vikramahuja1001 commented on a change in pull request #3917:
URL: https://github.com/apache/carbondata/pull/3917#discussion_r512553653
##########
File path: docs/dml-of-carbondata.md
##########
@@ -562,3 +563,50 @@ CarbonData DML statements are documented here,which
includes:
```
CLEAN FILES FOR TABLE carbon_table
```
+
+## CLEAN FILES
+
+ Clean files command is used to remove the Compacted and Marked
+ For Delete Segments from the store. Carbondata also supports Trash
+ Folder where all the stale data is moved to after clean files
+ is called
+
+ There are several types of compaction
+
+ ```
+ CLEAN FILES ON TABLE TableName
+ ```
+
+ - **Minor Compaction**
Review comment:
removed
##########
File path: docs/dml-of-carbondata.md
##########
@@ -562,3 +563,50 @@ CarbonData DML statements are documented here,which
includes:
```
CLEAN FILES FOR TABLE carbon_table
```
+
+## CLEAN FILES
+
+ Clean files command is used to remove the Compacted and Marked
Review comment:
linked
##########
File path:
integration/spark/src/main/scala/org/apache/carbondata/cleanfiles/CleanFilesUtil.scala
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.cleanfiles
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.spark.sql.{AnalysisException, CarbonEnv, Row, SparkSession}
+import org.apache.spark.sql.index.CarbonIndexUtil
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.exception.ConcurrentOperationException
+import org.apache.carbondata.core.indexstore.PartitionSpec
+import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil,
ICarbonLock, LockUsage}
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier,
CarbonMetadata, SegmentFileStore}
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil
+import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails,
SegmentStatus, SegmentStatusManager}
+import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
+import org.apache.carbondata.core.util.path.{CarbonTablePath, TrashUtil}
+import org.apache.carbondata.processing.loading.TableProcessingOperations
+import org.apache.carbondata.processing.loading.model.CarbonLoadModel
+
+object CleanFilesUtil {
+ private val LOGGER =
LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+
+ /**
+ * The method deletes all data if forceTableClean <true> and clean garbage
segment
+ * (MARKED_FOR_DELETE state) if forceTableClean <false>
+ *
+ * @param dbName : Database name
+ * @param tableName : Table name
+ * @param tablePath : Table path
+ * @param carbonTable : CarbonTable Object <null> in case of
force clean
+ * @param forceTableClean : <true> for force clean it will delete all
data
+ * <false> it will clean garbage segment
(MARKED_FOR_DELETE state)
+ * @param currentTablePartitions : Hive Partitions details
+ */
+ def cleanFiles(
+ dbName: String,
+ tableName: String,
+ tablePath: String,
+ timeStamp: String,
+ carbonTable: CarbonTable,
+ forceTableClean: Boolean,
+ currentTablePartitions: Option[Seq[PartitionSpec]] = None,
+ truncateTable: Boolean = false): Unit = {
+ var carbonCleanFilesLock: ICarbonLock = null
+ val absoluteTableIdentifier = if (forceTableClean) {
+ AbsoluteTableIdentifier.from(tablePath, dbName, tableName, tableName)
+ } else {
+ carbonTable.getAbsoluteTableIdentifier
+ }
+ try {
+ val errorMsg = "Clean files request is failed for " +
+ s"$dbName.$tableName" +
+ ". Not able to acquire the clean files lock due to another clean files
" +
+ "operation is running in the background."
+ // in case of force clean the lock is not required
+ if (forceTableClean) {
+ FileFactory.deleteAllCarbonFilesOfDir(
+ FileFactory.getCarbonFile(absoluteTableIdentifier.getTablePath))
+ } else {
+ carbonCleanFilesLock =
+ CarbonLockUtil
+ .getLockObject(absoluteTableIdentifier,
LockUsage.CLEAN_FILES_LOCK, errorMsg)
+ if (truncateTable) {
+ SegmentStatusManager.truncateTable(carbonTable)
+ }
+ SegmentStatusManager.deleteLoadsAndUpdateMetadata(
+ carbonTable, true, currentTablePartitions.map(_.asJava).orNull,
timeStamp)
+ CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, true)
+ currentTablePartitions match {
+ case Some(partitions) =>
+ SegmentFileStore.cleanSegments(
+ carbonTable,
+ currentTablePartitions.map(_.asJava).orNull,
+ timeStamp,
+ true)
+ case _ =>
+ }
+ }
+ } finally {
+ if (currentTablePartitions.equals(None)) {
+ cleanUpPartitionFoldersRecursively(carbonTable,
List.empty[PartitionSpec])
+ } else {
+ cleanUpPartitionFoldersRecursively(carbonTable,
currentTablePartitions.get.toList)
+ }
+
+ if (carbonCleanFilesLock != null) {
+ CarbonLockUtil.fileUnlock(carbonCleanFilesLock,
LockUsage.CLEAN_FILES_LOCK)
+ }
+ }
+ }
+
+
+ /**
+ * delete partition folders recursively
+ *
+ * @param carbonTable
+ * @param partitionSpecList
+ */
+ def cleanUpPartitionFoldersRecursively(carbonTable: CarbonTable,
+ partitionSpecList: List[PartitionSpec]): Unit = {
+ if (carbonTable != null && carbonTable.isHivePartitionTable) {
+ val loadMetadataDetails = SegmentStatusManager
+ .readLoadMetadata(carbonTable.getMetadataPath)
+
+ val carbonFile = FileFactory.getCarbonFile(carbonTable.getTablePath)
+
+ // list all files from table path
+ val listOfDefaultPartFilesIterator = carbonFile.listFiles(true)
+ loadMetadataDetails.foreach { metadataDetail =>
+ if
(metadataDetail.getSegmentStatus.equals(SegmentStatus.MARKED_FOR_DELETE) &&
+ metadataDetail.getSegmentFile == null) {
+ val loadStartTime: Long = metadataDetail.getLoadStartTime
+ // delete all files of @loadStartTime from table path
+ cleanCarbonFilesInFolder(listOfDefaultPartFilesIterator,
loadStartTime)
+ partitionSpecList.foreach {
+ partitionSpec =>
+ val partitionLocation = partitionSpec.getLocation
+ // For partition folder outside the tablePath
+ if
(!partitionLocation.toString.startsWith(carbonTable.getTablePath)) {
+ val partitionCarbonFile = FileFactory
+ .getCarbonFile(partitionLocation.toString)
+ // list all files from partitionLocation
+ val listOfExternalPartFilesIterator =
partitionCarbonFile.listFiles(true)
+ // delete all files of @loadStartTime from externalPath
+ cleanCarbonFilesInFolder(listOfExternalPartFilesIterator,
loadStartTime)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ * @param carbonFiles
+ * @param timestamp
+ */
+ private def cleanCarbonFilesInFolder(carbonFiles: java.util.List[CarbonFile],
+ timestamp: Long): Unit = {
+ carbonFiles.asScala.foreach { carbonFile =>
+ val filePath = carbonFile.getPath
+ val fileName = carbonFile.getName
+ if (CarbonTablePath.DataFileUtil.compareCarbonFileTimeStamp(fileName,
timestamp)) {
+ FileFactory.deleteFile(filePath)
+ }
+ }
+ }
+
+ /**
+ * The in-progress segments which are in stale state will be marked as
deleted
+ * when driver is initializing.
+ *
+ * @param databaseLocation
+ * @param dbName
+ */
+ def cleanInProgressSegments(databaseLocation: String, dbName: String,
timeStamp: String): Unit = {
+ val loaderDriver = CarbonProperties.getInstance().
+ getProperty(CarbonCommonConstants.DATA_MANAGEMENT_DRIVER,
+ CarbonCommonConstants.DATA_MANAGEMENT_DRIVER_DEFAULT).toBoolean
+ if (!loaderDriver) {
+ return
+ }
+ try {
+ if (FileFactory.isFileExist(databaseLocation)) {
+ val file = FileFactory.getCarbonFile(databaseLocation)
+ if (file.isDirectory) {
+ val tableFolders = file.listFiles()
+ tableFolders.foreach { tableFolder =>
+ if (tableFolder.isDirectory) {
+ val tablePath = databaseLocation +
CarbonCommonConstants.FILE_SEPARATOR +
+ tableFolder.getName
+ val tableUniqueName = CarbonTable.buildUniqueName(dbName,
tableFolder.getName)
+ val tableStatusFile =
+ CarbonTablePath.getTableStatusFilePath(tablePath)
+ if (FileFactory.isFileExist(tableStatusFile)) {
+ try {
+ val carbonTable = CarbonMetadata.getInstance
+ .getCarbonTable(tableUniqueName)
+
SegmentStatusManager.deleteLoadsAndUpdateMetadata(carbonTable, true, null,
+ timeStamp)
+ } catch {
+ case _: Exception =>
+ LOGGER.warn(s"Error while cleaning table " +
s"$tableUniqueName")
+ }
+ }
+ }
+ }
+ }
+ }
+ } catch {
+ case s: java.io.FileNotFoundException =>
+ LOGGER.error(s)
+ }
+ }
+
+ /**
+ * The below method deletes all the files and folders in the trash folders
of all carbon tables
+ * in all databases
+ */
+ def deleteDataFromTrashFolderInAllTables(sparkSession: SparkSession): Unit =
{
+ try {
+ val databases = sparkSession.sessionState.catalog.listDatabases()
+ databases.foreach(dbName => {
+ val databaseLocation = CarbonEnv.getDatabaseLocation(dbName,
sparkSession)
+ if (FileFactory.isFileExist(databaseLocation)) {
+ val file = FileFactory.getCarbonFile(databaseLocation)
+ if (file.isDirectory) {
+ val tableFolders = file.listFiles()
+ tableFolders.foreach { tableFolder =>
+ if (tableFolder.isDirectory) {
+ val tablePath = databaseLocation +
+ CarbonCommonConstants.FILE_SEPARATOR + tableFolder.getName
+ TrashUtil.deleteAllDataFromTrashFolder(tablePath)
+ }
+ }
+ }
+ }
+ })
+ } catch {
+ case e: Throwable =>
+ // catch all exceptions to avoid failure
+ LogServiceFactory.getLogService(this.getClass.getCanonicalName)
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -1427,6 +1427,25 @@ private CarbonCommonConstants() {
public static final String BITSET_PIPE_LINE_DEFAULT = "true";
+ public static final String MICROSECONDS_IN_A_DAY = "86400000";
+
+ /**
+ * this is the user defined time(in days), when a specific timestamp
subdirectory in
+ * trash folder will expire
+ */
+ @CarbonProperty
+ public static final String TRASH_EXPIRATION_TIME =
"carbon.trash.expiration.time";
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -1427,6 +1427,25 @@ private CarbonCommonConstants() {
public static final String BITSET_PIPE_LINE_DEFAULT = "true";
+ public static final String MICROSECONDS_IN_A_DAY = "86400000";
+
+ /**
+ * this is the user defined time(in days), when a specific timestamp
subdirectory in
+ * trash folder will expire
+ */
+ @CarbonProperty
+ public static final String TRASH_EXPIRATION_TIME =
"carbon.trash.expiration.time";
+
+ /**
+ * Default expiration time of trash folder is 3 days.
+ */
+ public static final String TRASH_EXPIRATION_TIME_DEFAULT = "3";
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
##########
@@ -1033,7 +1034,7 @@ public static void commitDropPartitions(CarbonTable
carbonTable, String uniqueId
* @throws IOException
*/
public static void cleanSegments(CarbonTable table, List<PartitionSpec>
partitionSpecs,
- boolean forceDelete) throws IOException {
+ String timeStamp, boolean forceDelete) throws IOException {
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
##########
@@ -1105,28 +1109,79 @@ public static void cleanSegments(CarbonTable table,
List<PartitionSpec> partitio
* @throws IOException
*/
public static void deleteSegment(String tablePath, Segment segment,
- List<PartitionSpec> partitionSpecs,
- SegmentUpdateStatusManager updateStatusManager) throws Exception {
+ List<PartitionSpec> partitionSpecs, SegmentUpdateStatusManager
updateStatusManager,
+ SegmentStatus segmentStatus, Boolean isPartitionTable, String timeStamp)
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -1427,6 +1428,25 @@ private CarbonCommonConstants() {
public static final String BITSET_PIPE_LINE_DEFAULT = "true";
+ public static final long MILLIS_SECONDS_IN_A_DAY = TimeUnit.DAYS.toMillis(1);
+
+ /**
+ * this is the user defined time(in days), when a specific timestamp
subdirectory in
+ * trash folder will expire
+ */
+ @CarbonProperty
+ public static final String TRASH_EXPIRATION_DAYS =
"carbon.trash.expiration.days";
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -1427,6 +1428,25 @@ private CarbonCommonConstants() {
public static final String BITSET_PIPE_LINE_DEFAULT = "true";
+ public static final long MILLIS_SECONDS_IN_A_DAY = TimeUnit.DAYS.toMillis(1);
+
+ /**
+ * this is the user defined time(in days), when a specific timestamp
subdirectory in
+ * trash folder will expire
+ */
+ @CarbonProperty
+ public static final String TRASH_EXPIRATION_DAYS =
"carbon.trash.expiration.days";
+
+ /**
+ * Default expiration time of trash folder is 3 days.
+ */
+ public static final String TRASH_EXPIRATION_DAYS_DEFAULT = "3";
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
##########
@@ -1049,7 +1049,7 @@ private static ReturnTuple isUpdateRequired(boolean
isForceDeletion, CarbonTable
}
public static void deleteLoadsAndUpdateMetadata(CarbonTable carbonTable,
boolean isForceDeletion,
- List<PartitionSpec> partitionSpecs) throws IOException {
+ List<PartitionSpec> partitionSpecs, String timeStamp) throws IOException
{
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
##########
@@ -2116,6 +2086,20 @@ public int getMaxSIRepairLimit(String dbName, String
tableName) {
return Math.abs(Integer.parseInt(thresholdValue));
}
+ /**
+ * The below method returns the microseconds after which the trash folder
will expire
+ */
+ public long getTrashFolderExpirationTime() {
+ String configuredValue =
getProperty(CarbonCommonConstants.TRASH_EXPIRATION_DAYS,
+ CarbonCommonConstants.TRASH_EXPIRATION_DAYS_DEFAULT);
+ int result = Integer.parseInt(configuredValue);
Review comment:
added try catch
##########
File path:
core/src/main/java/org/apache/carbondata/core/util/DeleteLoadFolders.java
##########
@@ -67,22 +69,23 @@ private static String
getSegmentPath(AbsoluteTableIdentifier identifier,
}
public static void physicalFactAndMeasureMetadataDeletion(CarbonTable
carbonTable,
- LoadMetadataDetails[] newAddedLoadHistoryList,
- boolean isForceDelete,
- List<PartitionSpec> specs) {
+ LoadMetadataDetails[] newAddedLoadHistoryList, boolean isForceDelete,
+ List<PartitionSpec> specs, String timeStamp) {
Review comment:
done
##########
File path:
core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
##########
@@ -792,4 +793,9 @@ public static String getParentPath(String dataFilePath) {
return dataFilePath;
}
}
+
+ public static String getTrashFolder(String carbonTablePath) {
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
