11443

Build # 11443

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/84ce094f2667758cb6820e71a332a1a259db3079">84ce094f2<a href="https://github.com/apache/carbondata/commit/84ce094f2667758cb6820e71a332a1a259db3079">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/85c000e91a1dc84784088f3dc2bc5a13b138344a">85c000e91</a>">85c000e91</a><a href="https://github.com/apache/carbondata/commit/84ce094f2667758cb6820e71a332a1a259db3079"> into 24753a924">24753a924</a>

Run Details

908 of 908 new or added lines in 42 files covered. (100.0%)

63939 of 81027 relevant lines covered (78.91%)

1.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.75

/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala

/*
 * 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.lang.Long

import scala.collection.JavaConverters._

import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.util.CarbonException
import org.apache.spark.unsafe.types.UTF8String

import org.apache.carbondata.common.Strings
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, SegmentStatus, SegmentStatusManager}
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 showSegments(
      limit: Option[String],
      tablePath: String,
      showHistory: Boolean): Seq[Row] = {
    val metaFolder = CarbonTablePath.getMetadataPath(tablePath)
    val loadMetadataDetailsArray = if (showHistory) {
      SegmentStatusManager.readLoadMetadata(metaFolder) ++
      SegmentStatusManager.readLoadHistoryMetadata(metaFolder)
    } else {
      SegmentStatusManager.readLoadMetadata(metaFolder)
    }

    if (loadMetadataDetailsArray.nonEmpty) {
      var loadMetadataDetailsSortedArray = loadMetadataDetailsArray.sortWith { (l1, l2) =>
        java.lang.Double.parseDouble(l1.getLoadName) > java.lang.Double.parseDouble(l2.getLoadName)
      }
      if (!showHistory) {
        loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray
          .filter(_.getVisibility.equalsIgnoreCase("true"))
      }
      if (limit.isDefined) {
        val limitLoads = limit.get
        try {
          val lim = Integer.parseInt(limitLoads)
          loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray.slice(0, lim)
        } catch {
          case _: NumberFormatException =>
            CarbonException.analysisException("Entered limit is not a valid Number")
        }
      }

      loadMetadataDetailsSortedArray
        .map { load =>
          val mergedTo =
            if (load.getMergedLoadName != null) {
              load.getMergedLoadName
            } else {
              "NA"
            }

          val startTime =
            if (load.getLoadStartTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) {
              "NA"
            } else {
              new java.sql.Timestamp(load.getLoadStartTime).toString
            }

          val endTime =
            if (load.getLoadEndTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) {
              "NA"
            } else {
              new java.sql.Timestamp(load.getLoadEndTime).toString
            }

          val (dataSize, indexSize) = if (load.getFileFormat == 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, FileFactory.getFileType(indexPath))
              (indices.asScala.map(_.getFile_size).sum, indexFile.getSize)
            } else {
              (-1L, -1L)
            }
          } else {
            // for batch segment, we can get the data size from table status file directly
            (if (load.getDataSize == null) -1L else load.getDataSize.toLong,
              if (load.getIndexSize == null) -1L else load.getIndexSize.toLong)
          }

          if (showHistory) {
            Row(
              load.getLoadName,
              load.getSegmentStatus.getMessage,
              startTime,
              endTime,
              mergedTo,
              load.getFileFormat.toString,
              load.getVisibility,
              Strings.formatSize(dataSize.toFloat),
              Strings.formatSize(indexSize.toFloat))
          } else {
            Row(
              load.getLoadName,
              load.getSegmentStatus.getMessage,
              startTime,
              endTime,
              mergedTo,
              load.getFileFormat.toString,
              Strings.formatSize(dataSize.toFloat),
              Strings.formatSize(indexSize.toFloat))
          }
        }.toSeq
    } else {
      Seq.empty
    }
  }

  /**
   * 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): 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,
            FileFactory.getFileType(absoluteTableIdentifier.getTablePath)))
      } else {
        carbonCleanFilesLock =
          CarbonLockUtil
            .getLockObject(absoluteTableIdentifier, LockUsage.CLEAN_FILES_LOCK, errorMsg)
        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) {
      val loadMetadataDetails = SegmentStatusManager
        .readLoadMetadata(carbonTable.getMetadataPath)

      val fileType = FileFactory.getFileType(carbonTable.getTablePath)
      val carbonFile = FileFactory.getCarbonFile(carbonTable.getTablePath, fileType)

      // 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 fileType = FileFactory.getFileType(partitionLocation.toString)
                val partitionCarbonFile = FileFactory
                  .getCarbonFile(partitionLocation.toString, fileType)
                // 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, FileFactory.getFileType(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)
    }
  }

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.api
19
20	import java.lang.Long
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.sql.Row
25	import org.apache.spark.sql.catalyst.util.DateTimeUtils
26	import org.apache.spark.sql.util.CarbonException
27	import org.apache.spark.unsafe.types.UTF8String
28
29	import org.apache.carbondata.common.Strings
30	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
31	import org.apache.carbondata.common.logging.LogServiceFactory
32	import org.apache.carbondata.core.constants.CarbonCommonConstants
33	import org.apache.carbondata.core.datastore.filesystem.CarbonFile
34	import org.apache.carbondata.core.datastore.impl.FileFactory
35	import org.apache.carbondata.core.indexstore.PartitionSpec
36	import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage}
37	import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, SegmentFileStore}
38	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
39	import org.apache.carbondata.core.mutate.CarbonUpdateUtil
40	import org.apache.carbondata.core.statusmanager.{FileFormat, SegmentStatus, SegmentStatusManager}
41	import org.apache.carbondata.core.util.path.CarbonTablePath
42	import org.apache.carbondata.streaming.segment.StreamSegment
43
44	object CarbonStore {
45	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
46
47	def showSegments(
48	limit: Option[String],
49	tablePath: String,
50	showHistory: Boolean): Seq[Row] = {	1✔
51	val metaFolder = CarbonTablePath.getMetadataPath(tablePath)	1✔
52	val loadMetadataDetailsArray = if (showHistory) {	1✔
53	SegmentStatusManager.readLoadMetadata(metaFolder) ++	1✔
54	SegmentStatusManager.readLoadHistoryMetadata(metaFolder)	1✔
55	} else {
56	SegmentStatusManager.readLoadMetadata(metaFolder)	1✔
57	}
58
59	if (loadMetadataDetailsArray.nonEmpty) {	1✔
60	var loadMetadataDetailsSortedArray = loadMetadataDetailsArray.sortWith { (l1, l2) =>	1✔
61	java.lang.Double.parseDouble(l1.getLoadName) > java.lang.Double.parseDouble(l2.getLoadName)	1✔
62	}
63	if (!showHistory) {	1✔
64	loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray	1✔
65	.filter(_.getVisibility.equalsIgnoreCase("true"))	1✔
66	}
67	if (limit.isDefined) {	1✔
68	val limitLoads = limit.get	1✔
69	try {	×
70	val lim = Integer.parseInt(limitLoads)	1✔
71	loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray.slice(0, lim)	1✔
72	} catch {
73	case _: NumberFormatException =>
74	CarbonException.analysisException("Entered limit is not a valid Number")	×
75	}
76	}
77
78	loadMetadataDetailsSortedArray	1✔
79	.map { load =>	1✔
80	val mergedTo =	1✔
81	if (load.getMergedLoadName != null) {	1✔
82	load.getMergedLoadName	1✔
83	} else {
84	"NA"	1✔
85	}
86
87	val startTime =	1✔
88	if (load.getLoadStartTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) {	1✔
89	"NA"	×
90	} else {
91	new java.sql.Timestamp(load.getLoadStartTime).toString	1✔
92	}
93
94	val endTime =	1✔
95	if (load.getLoadEndTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) {	1✔
96	"NA"	1✔
97	} else {
98	new java.sql.Timestamp(load.getLoadEndTime).toString	1✔
99	}
100
101	val (dataSize, indexSize) = if (load.getFileFormat == FileFormat.ROW_V1) {	1✔
102	// for streaming segment, we should get the actual size from the index file
103	// since it is continuously inserting data
104	val segmentDir = CarbonTablePath.getSegmentPath(tablePath, load.getLoadName)	1✔
105	val indexPath = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir)	1✔
106	val indexFile = FileFactory.getCarbonFile(indexPath)	1✔
107	if (indexFile.exists()) {	1✔
108	val indices =	1✔
109	StreamSegment.readIndexFile(indexPath, FileFactory.getFileType(indexPath))	1✔
110	(indices.asScala.map(_.getFile_size).sum, indexFile.getSize)	1✔
111	} else {
112	(-1L, -1L)	×
113	}
114	} else {
115	// for batch segment, we can get the data size from table status file directly
116	(if (load.getDataSize == null) -1L else load.getDataSize.toLong,	1✔
117	if (load.getIndexSize == null) -1L else load.getIndexSize.toLong)	1✔
118	}
119
120	if (showHistory) {	1✔
121	Row(	1✔
122	load.getLoadName,	1✔
123	load.getSegmentStatus.getMessage,	1✔
124	startTime,	1✔
125	endTime,	1✔
126	mergedTo,	1✔
127	load.getFileFormat.toString,	1✔
128	load.getVisibility,	1✔
129	Strings.formatSize(dataSize.toFloat),	1✔
130	Strings.formatSize(indexSize.toFloat))	1✔
131	} else {
132	Row(	1✔
133	load.getLoadName,	1✔
134	load.getSegmentStatus.getMessage,	1✔
135	startTime,	1✔
136	endTime,	1✔
137	mergedTo,	1✔
138	load.getFileFormat.toString,	1✔
139	Strings.formatSize(dataSize.toFloat),	1✔
140	Strings.formatSize(indexSize.toFloat))	1✔
141	}
142	}.toSeq	1✔
143	} else {
144	Seq.empty	×
145	}
146	}
147
148	/**
149	* The method deletes all data if forceTableCLean <true> and lean garbage segment
150	* (MARKED_FOR_DELETE state) if forceTableCLean <false>
151	*
152	* @param dbName : Database name
153	* @param tableName : Table name
154	* @param tablePath : Table path
155	* @param carbonTable : CarbonTable Object <null> in case of force clean
156	* @param forceTableClean : <true> for force clean it will delete all data
157	* <false> it will clean garbage segment (MARKED_FOR_DELETE state)
158	* @param currentTablePartitions : Hive Partitions details
159	*/
160	def cleanFiles(
161	dbName: String,
162	tableName: String,
163	tablePath: String,
164	carbonTable: CarbonTable,
165	forceTableClean: Boolean,
166	currentTablePartitions: Option[Seq[PartitionSpec]] = None): Unit = {	1✔
167	var carbonCleanFilesLock: ICarbonLock = null	1✔
168	val absoluteTableIdentifier = if (forceTableClean) {	1✔
169	AbsoluteTableIdentifier.from(tablePath, dbName, tableName, tableName)	1✔
170	} else {
171	carbonTable.getAbsoluteTableIdentifier	1✔
172	}
173	try {
174	val errorMsg = "Clean files request is failed for " +	1✔
175	s"$dbName.$tableName" +	1✔
176	". Not able to acquire the clean files lock due to another clean files " +	1✔
177	"operation is running in the background."	1✔
178	// in case of force clean the lock is not required
179	if (forceTableClean) {	1✔
180	FileFactory.deleteAllCarbonFilesOfDir(	1✔
181	FileFactory.getCarbonFile(absoluteTableIdentifier.getTablePath,	1✔
182	FileFactory.getFileType(absoluteTableIdentifier.getTablePath)))	1✔
183	} else {
184	carbonCleanFilesLock =	1✔
185	CarbonLockUtil
186	.getLockObject(absoluteTableIdentifier, LockUsage.CLEAN_FILES_LOCK, errorMsg)	1✔
187	SegmentStatusManager.deleteLoadsAndUpdateMetadata(	1✔
188	carbonTable, true, currentTablePartitions.map(_.asJava).orNull)	1✔
189	CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, true)	1✔
190	currentTablePartitions match {	1✔
191	case Some(partitions) =>	1✔
192	SegmentFileStore.cleanSegments(	1✔
193	carbonTable,	1✔
194	currentTablePartitions.map(_.asJava).orNull,	1✔
195	true)	1✔
196	case _ =>	1✔
197	}
198	}
199	} finally {	1✔
200	if (currentTablePartitions.equals(None)) {	1✔
201	cleanUpPartitionFoldersRecursively(carbonTable, List.empty[PartitionSpec])	1✔
202	} else {
203	cleanUpPartitionFoldersRecursively(carbonTable, currentTablePartitions.get.toList)	1✔
204	}
205
206	if (carbonCleanFilesLock != null) {	1✔
207	CarbonLockUtil.fileUnlock(carbonCleanFilesLock, LockUsage.CLEAN_FILES_LOCK)	1✔
208	}
209	}
210	}
211
212	/**
213	* delete partition folders recursively
214	*
215	* @param carbonTable
216	* @param partitionSpecList
217	*/
218	def cleanUpPartitionFoldersRecursively(carbonTable: CarbonTable,
219	partitionSpecList: List[PartitionSpec]): Unit = {
220	if (carbonTable != null) {	1✔
221	val loadMetadataDetails = SegmentStatusManager	1✔
222	.readLoadMetadata(carbonTable.getMetadataPath)	1✔
223
224	val fileType = FileFactory.getFileType(carbonTable.getTablePath)	1✔
225	val carbonFile = FileFactory.getCarbonFile(carbonTable.getTablePath, fileType)	1✔
226
227	// list all files from table path
228	val listOfDefaultPartFilesIterator = carbonFile.listFiles(true)	1✔
229	loadMetadataDetails.foreach { metadataDetail =>	1✔
230	if (metadataDetail.getSegmentStatus.equals(SegmentStatus.MARKED_FOR_DELETE) &&	1✔
231	metadataDetail.getSegmentFile == null) {	1✔
232	val loadStartTime: Long = metadataDetail.getLoadStartTime	×
233	// delete all files of @loadStartTime from tablepath
234	cleanCarbonFilesInFolder(listOfDefaultPartFilesIterator, loadStartTime)	×
235	partitionSpecList.foreach {	×
236	partitionSpec =>	×
237	val partitionLocation = partitionSpec.getLocation	×
238	// For partition folder outside the tablePath
239	if (!partitionLocation.toString.startsWith(carbonTable.getTablePath)) {	×
240	val fileType = FileFactory.getFileType(partitionLocation.toString)	×
241	val partitionCarbonFile = FileFactory	×
242	.getCarbonFile(partitionLocation.toString, fileType)	×
243	// list all files from partitionLoacation
244	val listOfExternalPartFilesIterator = partitionCarbonFile.listFiles(true)	×
245	// delete all files of @loadStartTime from externalPath
246	cleanCarbonFilesInFolder(listOfExternalPartFilesIterator, loadStartTime)	×
247	}
248	}
249	}
250	}
251	}
252	}
253
254	/**
255	*
256	* @param carbonFiles
257	* @param timestamp
258	*/
259	private def cleanCarbonFilesInFolder(carbonFiles: java.util.List[CarbonFile],
260	timestamp: Long): Unit = {
261	carbonFiles.asScala.foreach {	×
262	carbonFile =>	×
263	val filePath = carbonFile.getPath	×
264	val fileName = carbonFile.getName	×
265	if (CarbonTablePath.DataFileUtil.compareCarbonFileTimeStamp(fileName, timestamp)) {	×
266	// delete the file
267	FileFactory.deleteFile(filePath, FileFactory.getFileType(filePath))	×
268
269	}
270	}
271	}
272
273	// validates load ids
274	private def validateLoadIds(loadids: Seq[String]): Unit = {
275	if (loadids.isEmpty) {	1✔
276	val errorMessage = "Error: Segment id(s) should not be empty."	1✔
277	throw new MalformedCarbonCommandException(errorMessage)	1✔
278	}
279	}
280
281	// TODO: move dbName and tableName to caller, caller should handle the log and error
282	def deleteLoadById(
283	loadids: Seq[String],
284	dbName: String,
285	tableName: String,
286	carbonTable: CarbonTable): Unit = {
287
288	validateLoadIds(loadids)	1✔
289
290	val path = carbonTable.getMetadataPath	1✔
291
292	try {	1✔
293	val invalidLoadIds = SegmentStatusManager.updateDeletionStatus(	1✔
294	carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala	1✔
295	if (invalidLoadIds.isEmpty) {	1✔
296	LOGGER.info(s"Delete segment by Id is successfull for $dbName.$tableName.")	1✔
297	} else {
298	sys.error(s"Delete segment by Id is failed. Invalid ID is: ${invalidLoadIds.mkString(",")}")	×
299	}
300	} catch {
301	case ex: Exception =>
302	sys.error(ex.getMessage)	×
303	}
304	Seq.empty	1✔
305	}
306
307	// TODO: move dbName and tableName to caller, caller should handle the log and error
308	def deleteLoadByDate(
309	timestamp: String,
310	dbName: String,
311	tableName: String,
312	carbonTable: CarbonTable): Unit = {	1✔
313
314	val time = validateTimeFormat(timestamp)	1✔
315	val path = carbonTable.getMetadataPath	1✔
316
317	try {	1✔
318	val invalidLoadTimestamps =	1✔
319	SegmentStatusManager.updateDeletionStatus(	1✔
320	carbonTable.getAbsoluteTableIdentifier,	1✔
321	timestamp,	1✔
322	path,	1✔
323	time).asScala	1✔
324	if (invalidLoadTimestamps.isEmpty) {	1✔
325	LOGGER.info(s"Delete segment by date is successful for $dbName.$tableName.")	1✔
326	} else {
327	sys.error("Delete segment by date is failed. No matching segment found.")	×
328	}
329	} catch {
330	case ex: Exception =>
331	sys.error(ex.getMessage)	×
332	}
333	}
334
335	// this function is for test only
336	def isSegmentValid(
337	dbName: String,
338	tableName: String,
339	storePath: String,
340	segmentId: String): Boolean = {
341	val identifier = AbsoluteTableIdentifier.from(storePath, dbName, tableName, tableName)	1✔
342	val validAndInvalidSegments: SegmentStatusManager.ValidAndInvalidSegmentsInfo = new	1✔
343	SegmentStatusManager(
344	identifier).getValidAndInvalidSegments	1✔
345	validAndInvalidSegments.getValidSegments.contains(segmentId)	1✔
346	}
347
348	private def validateTimeFormat(timestamp: String): Long = {
349	try {	1✔
350	DateTimeUtils.stringToTimestamp(UTF8String.fromString(timestamp)).get	1✔
351	} catch {
352	case e: Exception =>	1✔
353	val errorMessage = "Error: Invalid load start time format: " + timestamp	1✔
354	throw new MalformedCarbonCommandException(errorMessage)	1✔
355	}
356	}
357
358	}	1✔

apache / carbondata / 11443

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous