| /* |
| * 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.api |
| |
| import java.io.InputStreamReader |
| import java.time.{Duration, Instant} |
| import java.util |
| import java.util.{Collections, Comparator} |
| |
| import scala.collection.JavaConverters._ |
| |
| import com.google.gson.Gson |
| import org.apache.commons.lang3.StringUtils |
| import org.apache.hadoop.conf.Configuration |
| import org.apache.spark.sql.catalyst.util.DateTimeUtils |
| import org.apache.spark.unsafe.types.UTF8String |
| |
| import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException |
| 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.indexstore.PartitionSpec |
| import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage} |
| import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, SegmentFileStore} |
| import org.apache.carbondata.core.metadata.schema.table.CarbonTable |
| import org.apache.carbondata.core.mutate.CarbonUpdateUtil |
| import org.apache.carbondata.core.statusmanager.{FileFormat, LoadMetadataDetails, SegmentStatus, SegmentStatusManager, StageInput} |
| import org.apache.carbondata.core.util.path.CarbonTablePath |
| import org.apache.carbondata.streaming.segment.StreamSegment |
| |
| object CarbonStore { |
| private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName) |
| |
| def readSegments( |
| tablePath: String, |
| showHistory: Boolean, |
| limit: Option[Int]): Array[LoadMetadataDetails] = { |
| val metaFolder = CarbonTablePath.getMetadataPath(tablePath) |
| var segmentsMetadataDetails = if (showHistory) { |
| SegmentStatusManager.readLoadMetadata(metaFolder) ++ |
| SegmentStatusManager.readLoadHistoryMetadata(metaFolder) |
| } else { |
| SegmentStatusManager.readLoadMetadata(metaFolder) |
| } |
| if (!showHistory) { |
| segmentsMetadataDetails = segmentsMetadataDetails |
| .filter(_.getVisibility.equalsIgnoreCase("true")) |
| segmentsMetadataDetails = segmentsMetadataDetails.sortWith { (l1, l2) => |
| java.lang.Double.parseDouble(l1.getLoadName) > |
| java.lang.Double.parseDouble(l2.getLoadName) |
| } |
| } |
| |
| if (limit.isDefined) { |
| segmentsMetadataDetails.slice(0, limit.get) |
| } else { |
| segmentsMetadataDetails |
| } |
| } |
| |
| /** |
| * Read stage files and return input files |
| */ |
| def readStages(tablePath: String): Seq[StageInput] = { |
| val stageFiles = listStageFiles(CarbonTablePath.getStageDir(tablePath)) |
| var output = Collections.synchronizedList(new util.ArrayList[StageInput]()) |
| output.addAll(readStageInput(stageFiles._1, |
| StageInput.StageStatus.Unload).asJavaCollection) |
| output.addAll(readStageInput(stageFiles._2, |
| StageInput.StageStatus.Loading).asJavaCollection) |
| Collections.sort(output, new Comparator[StageInput]() { |
| def compare(stageInput1: StageInput, stageInput2: StageInput): Int = { |
| (stageInput2.getCreateTime - stageInput1.getCreateTime).intValue() |
| } |
| }) |
| output.asScala |
| } |
| |
| /** |
| * Read stage files and return input files |
| */ |
| def readStageInput( |
| stageFiles: Seq[CarbonFile], |
| status: StageInput.StageStatus): Seq[StageInput] = { |
| val gson = new Gson() |
| val output = Collections.synchronizedList(new util.ArrayList[StageInput]()) |
| stageFiles.map { stage => |
| val filePath = stage.getAbsolutePath |
| val stream = FileFactory.getDataInputStream(filePath) |
| try { |
| val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput]) |
| stageInput.setCreateTime(stage.getLastModifiedTime) |
| stageInput.setStatus(status) |
| output.add(stageInput) |
| } finally { |
| stream.close() |
| } |
| } |
| output.asScala |
| } |
| |
| /* |
| * Collect all stage files and matched success files and loading files. |
| * return unloaded stagefiles and loading stagefiles in the end. |
| */ |
| def listStageFiles( |
| loadDetailsDir: String): (Array[CarbonFile], Array[CarbonFile]) = { |
| val dir = FileFactory.getCarbonFile(loadDetailsDir) |
| if (dir.exists()) { |
| // 1. List all files in the stage dictionary. |
| val allFiles = dir.listFiles() |
| |
| // 2. Get StageFile list. |
| // Firstly, get the stage files in the stage dictionary. |
| // which exclude the success files and loading files |
| // Second, only collect the stage files having success tag. |
| val stageFiles = allFiles.filterNot { file => |
| file.getName.endsWith(CarbonTablePath.SUCCESS_FILE_SUFFIX) |
| }.filterNot { file => |
| file.getName.endsWith(CarbonTablePath.LOADING_FILE_SUFFIX) |
| }.filter { file => |
| allFiles.contains(file.getName + CarbonTablePath.SUCCESS_FILE_SUFFIX) |
| }.sortWith { |
| (file1, file2) => file1.getLastModifiedTime > file2.getLastModifiedTime |
| } |
| // 3. Get the unloaded stage files, which haven't loading tag. |
| val unloadedFiles = stageFiles.filterNot { file => |
| allFiles.contains(file.getName + CarbonTablePath.LOADING_FILE_SUFFIX) |
| } |
| // 4. Get the loading stage files, which have loading tag. |
| val loadingFiles = stageFiles.filter { file => |
| allFiles.contains(file.getName + CarbonTablePath.LOADING_FILE_SUFFIX) |
| } |
| (unloadedFiles, loadingFiles) |
| } else { |
| (Array.empty, Array.empty) |
| } |
| } |
| |
| def getPartitions(tablePath: String, load: LoadMetadataDetails): Seq[String] = { |
| val segmentFile = SegmentFileStore.readSegmentFile( |
| CarbonTablePath.getSegmentFilePath(tablePath, load.getSegmentFile)) |
| if (segmentFile == null) { |
| return Seq.empty |
| } |
| val locationMap = segmentFile.getLocationMap |
| if (locationMap != null) { |
| locationMap.asScala.map { |
| case (_, detail) => |
| s"{${ detail.getPartitions.asScala.mkString(",") }}" |
| }.toSeq |
| } else { |
| Seq.empty |
| } |
| } |
| |
| def getMergeTo(load: LoadMetadataDetails): String = { |
| if (load.getMergedLoadName != null) { |
| load.getMergedLoadName |
| } else { |
| "NA" |
| } |
| } |
| |
| def getExternalSegmentPath(load: LoadMetadataDetails): String = { |
| if (StringUtils.isNotEmpty(load.getPath)) { |
| load.getPath |
| } else { |
| "NA" |
| } |
| } |
| |
| def getLoadStartTime(load: LoadMetadataDetails): String = { |
| val startTime = |
| if (load.getLoadStartTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) { |
| "NA" |
| } else { |
| new java.sql.Timestamp(load.getLoadStartTime).toString |
| } |
| startTime |
| } |
| |
| def getLoadEndTime(load: LoadMetadataDetails): String = { |
| val endTime = |
| if (load.getLoadStartTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) { |
| "NA" |
| } else { |
| new java.sql.Timestamp(load.getLoadEndTime).toString |
| } |
| endTime |
| } |
| |
| def getLoadTimeTaken(load: LoadMetadataDetails): String = { |
| if (load.getLoadEndTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) { |
| "NA" |
| } else { |
| Duration.between( |
| Instant.ofEpochMilli(load.getLoadStartTime), |
| Instant.ofEpochMilli(load.getLoadEndTime) |
| ).toString.replace("PT", "") |
| } |
| } |
| |
| def getLoadTimeTakenAsMillis(load: LoadMetadataDetails): Long = { |
| if (load.getLoadEndTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) { |
| // loading in progress |
| -1L |
| } else { |
| load.getLoadEndTime - load.getLoadStartTime |
| } |
| } |
| |
| def getDataAndIndexSize( |
| tablePath: String, |
| load: LoadMetadataDetails): (Long, Long) = { |
| val (dataSize, indexSize) = if (load.getFileFormat.equals(FileFormat.ROW_V1)) { |
| // for streaming segment, we should get the actual size from the index file |
| // since it is continuously inserting data |
| val segmentDir = CarbonTablePath.getSegmentPath(tablePath, load.getLoadName) |
| val indexPath = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir) |
| val indexFile = FileFactory.getCarbonFile(indexPath) |
| if (indexFile.exists()) { |
| val indices = |
| StreamSegment.readIndexFile(indexPath) |
| (indices.asScala.map(_.getFile_size).sum, indexFile.getSize) |
| } else { |
| (-1L, -1L) |
| } |
| } else { |
| // If the added segment is other than carbon segment then we can only display the data |
| // size and not index size, we can get the data size from table status file directly |
| if (!load.getFileFormat.isCarbonFormat) { |
| (if (load.getDataSize == null) -1L else load.getDataSize.toLong, -1L) |
| } else { |
| (if (load.getDataSize == null) -1L else load.getDataSize.toLong, |
| if (load.getIndexSize == null) -1L else load.getIndexSize.toLong) |
| } |
| } |
| (dataSize, indexSize) |
| } |
| |
| /** |
| * The method deletes all data if forceTableCLean <true> and lean 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, |
| 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) |
| CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, true) |
| currentTablePartitions match { |
| case Some(partitions) => |
| SegmentFileStore.cleanSegments( |
| carbonTable, |
| currentTablePartitions.map(_.asJava).orNull, |
| 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 tablepath |
| 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 partitionLoacation |
| 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)) { |
| // delete the file |
| FileFactory.deleteFile(filePath) |
| } |
| } |
| } |
| |
| // validates load ids |
| private def validateLoadIds(loadids: Seq[String]): Unit = { |
| if (loadids.isEmpty) { |
| val errorMessage = "Error: Segment id(s) should not be empty." |
| throw new MalformedCarbonCommandException(errorMessage) |
| } |
| } |
| |
| // TODO: move dbName and tableName to caller, caller should handle the log and error |
| def deleteLoadById( |
| loadids: Seq[String], |
| dbName: String, |
| tableName: String, |
| carbonTable: CarbonTable): Unit = { |
| |
| validateLoadIds(loadids) |
| |
| val path = carbonTable.getMetadataPath |
| |
| try { |
| val invalidLoadIds = SegmentStatusManager.updateDeletionStatus( |
| carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala |
| if (invalidLoadIds.isEmpty) { |
| LOGGER.info(s"Delete segment by Id is successfull for $dbName.$tableName.") |
| } else { |
| sys.error(s"Delete segment by Id is failed. Invalid ID is: ${invalidLoadIds.mkString(",")}") |
| } |
| } catch { |
| case ex: Exception => |
| sys.error(ex.getMessage) |
| } |
| Seq.empty |
| } |
| |
| // TODO: move dbName and tableName to caller, caller should handle the log and error |
| def deleteLoadByDate( |
| timestamp: String, |
| dbName: String, |
| tableName: String, |
| carbonTable: CarbonTable): Unit = { |
| |
| val time = validateTimeFormat(timestamp) |
| val path = carbonTable.getMetadataPath |
| |
| try { |
| val invalidLoadTimestamps = |
| SegmentStatusManager.updateDeletionStatus( |
| carbonTable.getAbsoluteTableIdentifier, |
| timestamp, |
| path, |
| time).asScala |
| if (invalidLoadTimestamps.isEmpty) { |
| LOGGER.info(s"Delete segment by date is successful for $dbName.$tableName.") |
| } else { |
| sys.error("Delete segment by date is failed. No matching segment found.") |
| } |
| } catch { |
| case ex: Exception => |
| sys.error(ex.getMessage) |
| } |
| } |
| |
| // this function is for test only |
| def isSegmentValid( |
| dbName: String, |
| tableName: String, |
| storePath: String, |
| segmentId: String): Boolean = { |
| val identifier = AbsoluteTableIdentifier.from(storePath, dbName, tableName, tableName) |
| val validAndInvalidSegments: SegmentStatusManager.ValidAndInvalidSegmentsInfo = new |
| SegmentStatusManager( |
| identifier).getValidAndInvalidSegments |
| validAndInvalidSegments.getValidSegments.contains(segmentId) |
| } |
| |
| private def validateTimeFormat(timestamp: String): Long = { |
| try { |
| DateTimeUtils.stringToTimestamp(UTF8String.fromString(timestamp)).get |
| } catch { |
| case e: Exception => |
| val errorMessage = "Error: Invalid load start time format: " + timestamp |
| throw new MalformedCarbonCommandException(errorMessage) |
| } |
| } |
| |
| } |
