6651

Build # 6651

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ea4b9f56a14ffb5684c7fd15eb6ae251bc0b4345">ea4b9f56a<a href="https://github.com/apache/carbondata/commit/ea4b9f56a14ffb5684c7fd15eb6ae251bc0b4345">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/3daee1dde2033d76771aa2f0c276eb1e53f56c40">3daee1dde</a>">3daee1dde</a><a href="https://github.com/apache/carbondata/commit/ea4b9f56a14ffb5684c7fd15eb6ae251bc0b4345"> into 9ee0f3579">9ee0f3579</a>

Run Details

301 of 301 new or added lines in 73 files covered. (100.0%)

21940 of 70484 relevant lines covered (31.13%)

0.45 hits per line

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

0.0

/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.hadoop.fs.{LocatedFileStatus, Path, RemoteIterator}
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.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.{SegmentStatus, SegmentStatusManager}
import org.apache.carbondata.core.util.path.CarbonTablePath

object CarbonStore {
  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  def showSegments(
      limit: Option[String],
      tableFolderPath: String,
      showHistory: Boolean): Seq[Row] = {
    val loadMetadataDetailsArray = if (showHistory) {
      SegmentStatusManager.readLoadMetadata(tableFolderPath) ++
      SegmentStatusManager.readLoadHistoryMetadata(tableFolderPath)
    } else {
      SegmentStatusManager.readLoadMetadata(tableFolderPath)
    }

    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) {
              null
            } else {
              new java.sql.Timestamp(load.getLoadStartTime)
            }

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

          if (showHistory) {
            Row(
              load.getLoadName,
              load.getSegmentStatus.getMessage,
              startTime,
              endTime,
              mergedTo,
              load.getFileFormat.toString,
              load.getVisibility())
          } else {
            Row(
              load.getLoadName,
              load.getSegmentStatus.getMessage,
              startTime,
              endTime,
              mergedTo,
              load.getFileFormat.toString)
          }
        }.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 = {
    LOGGER.audit(s"The clean files request has been received for $dbName.$tableName")
    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)) {
        cleanUpPartitionFoldersRecurssively(carbonTable, List.empty[PartitionSpec])
      } else {
        cleanUpPartitionFoldersRecurssively(carbonTable, currentTablePartitions.get.toList)
      }

      if (carbonCleanFilesLock != null) {
        CarbonLockUtil.fileUnlock(carbonCleanFilesLock, LockUsage.CLEAN_FILES_LOCK)
      }
    }
    LOGGER.audit(s"Clean files operation is success for $dbName.$tableName.")
  }

  /**
   * delete partition folders recurssively
   *
   * @param carbonTable
   * @param partitionSpecList
   */
  def cleanUpPartitionFoldersRecurssively(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)
        }
    }
  }

  // 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 = {

    LOGGER.audit(s"Delete segment by Id request has been received for $dbName.$tableName")
    validateLoadIds(loadids)

    val path = carbonTable.getMetadataPath

    try {
      val invalidLoadIds = SegmentStatusManager.updateDeletionStatus(
        carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala
      if (invalidLoadIds.isEmpty) {
        LOGGER.audit(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 = {
    LOGGER.audit(s"Delete segment by Id request has been received for $dbName.$tableName")

    val time = validateTimeFormat(timestamp)
    val path = carbonTable.getMetadataPath

    try {
      val invalidLoadTimestamps =
        SegmentStatusManager.updateDeletionStatus(
          carbonTable.getAbsoluteTableIdentifier,
          timestamp,
          path,
          time).asScala
      if (invalidLoadTimestamps.isEmpty) {
        LOGGER.audit(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.hadoop.fs.{LocatedFileStatus, Path, RemoteIterator}
25	import org.apache.spark.sql.Row
26	import org.apache.spark.sql.catalyst.util.DateTimeUtils
27	import org.apache.spark.sql.util.CarbonException
28	import org.apache.spark.unsafe.types.UTF8String
29
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.{SegmentStatus, SegmentStatusManager}
41	import org.apache.carbondata.core.util.path.CarbonTablePath
42
43	object CarbonStore {
44	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	×
45
46	def showSegments(
47	limit: Option[String],
48	tableFolderPath: String,
49	showHistory: Boolean): Seq[Row] = {	×
50	val loadMetadataDetailsArray = if (showHistory) {	×
51	SegmentStatusManager.readLoadMetadata(tableFolderPath) ++	×
52	SegmentStatusManager.readLoadHistoryMetadata(tableFolderPath)	×
53	} else {
54	SegmentStatusManager.readLoadMetadata(tableFolderPath)	×
55	}
56
57	if (loadMetadataDetailsArray.nonEmpty) {	×
58	var loadMetadataDetailsSortedArray = loadMetadataDetailsArray.sortWith { (l1, l2) =>	×
59	java.lang.Double.parseDouble(l1.getLoadName) > java.lang.Double.parseDouble(l2.getLoadName)	×
60	}
61	if (!showHistory) {	×
62	loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray	×
63	.filter(_.getVisibility.equalsIgnoreCase("true"))	×
64	}
65	if (limit.isDefined) {	×
66	val limitLoads = limit.get	×
67	try {	×
68	val lim = Integer.parseInt(limitLoads)	×
69	loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray.slice(0, lim)	×
70	} catch {
71	case _: NumberFormatException =>
72	CarbonException.analysisException("Entered limit is not a valid Number")	×
73	}
74	}
75
76	loadMetadataDetailsSortedArray	×
77	.map { load =>	×
78	val mergedTo =	×
79	if (load.getMergedLoadName != null) {	×
80	load.getMergedLoadName	×
81	} else {
82	"NA"	×
83	}
84
85	val startTime =	×
86	if (load.getLoadStartTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) {	×
87	null	×
88	} else {
89	new java.sql.Timestamp(load.getLoadStartTime)	×
90	}
91
92	val endTime =	×
93	if (load.getLoadEndTime == CarbonCommonConstants.SEGMENT_LOAD_TIME_DEFAULT) {	×
94	null	×
95	} else {
96	new java.sql.Timestamp(load.getLoadEndTime)	×
97	}
98
99	if (showHistory) {	×
100	Row(	×
101	load.getLoadName,	×
102	load.getSegmentStatus.getMessage,	×
103	startTime,	×
104	endTime,	×
105	mergedTo,	×
106	load.getFileFormat.toString,	×
107	load.getVisibility())	×
108	} else {
109	Row(	×
110	load.getLoadName,	×
111	load.getSegmentStatus.getMessage,	×
112	startTime,	×
113	endTime,	×
114	mergedTo,	×
115	load.getFileFormat.toString)	×
116	}
117	}.toSeq	×
118	} else {
119	Seq.empty	×
120	}
121	}
122
123	/**
124	* The method deletes all data if forceTableCLean <true> and lean garbage segment
125	* (MARKED_FOR_DELETE state) if forceTableCLean <false>
126	*
127	* @param dbName : Database name
128	* @param tableName : Table name
129	* @param tablePath : Table path
130	* @param carbonTable : CarbonTable Object <null> in case of force clean
131	* @param forceTableClean : <true> for force clean it will delete all data
132	* <false> it will clean garbage segment (MARKED_FOR_DELETE state)
133	* @param currentTablePartitions : Hive Partitions details
134	*/
135	def cleanFiles(
136	dbName: String,
137	tableName: String,
138	tablePath: String,
139	carbonTable: CarbonTable,
140	forceTableClean: Boolean,
141	currentTablePartitions: Option[Seq[PartitionSpec]] = None): Unit = {	×
142	LOGGER.audit(s"The clean files request has been received for $dbName.$tableName")	×
143	var carbonCleanFilesLock: ICarbonLock = null	×
144	val absoluteTableIdentifier = if (forceTableClean) {	×
145	AbsoluteTableIdentifier.from(tablePath, dbName, tableName, tableName)	×
146	} else {
147	carbonTable.getAbsoluteTableIdentifier	×
148	}
149	try {	×
150	val errorMsg = "Clean files request is failed for " +	×
151	s"$dbName.$tableName" +	×
152	". Not able to acquire the clean files lock due to another clean files " +	×
153	"operation is running in the background."	×
154	// in case of force clean the lock is not required
155	if (forceTableClean) {	×
156	FileFactory.deleteAllCarbonFilesOfDir(	×
157	FileFactory.getCarbonFile(absoluteTableIdentifier.getTablePath,	×
158	FileFactory.getFileType(absoluteTableIdentifier.getTablePath)))	×
159	} else {	×
160	carbonCleanFilesLock =	×
161	CarbonLockUtil
162	.getLockObject(absoluteTableIdentifier, LockUsage.CLEAN_FILES_LOCK, errorMsg)	×
163	SegmentStatusManager.deleteLoadsAndUpdateMetadata(	×
164	carbonTable, true, currentTablePartitions.map(_.asJava).orNull)	×
165	CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, true)	×
166	currentTablePartitions match {	×
167	case Some(partitions) =>	×
168	SegmentFileStore.cleanSegments(	×
169	carbonTable,	×
170	currentTablePartitions.map(_.asJava).orNull,	×
171	true)	×
172	case _ =>	×
173	}
174	}
175	} finally {	×
176	if (currentTablePartitions.equals(None)) {	×
177	cleanUpPartitionFoldersRecurssively(carbonTable, List.empty[PartitionSpec])	×
178	} else {
179	cleanUpPartitionFoldersRecurssively(carbonTable, currentTablePartitions.get.toList)	×
180	}
181
182	if (carbonCleanFilesLock != null) {	×
183	CarbonLockUtil.fileUnlock(carbonCleanFilesLock, LockUsage.CLEAN_FILES_LOCK)	×
184	}
185	}
186	LOGGER.audit(s"Clean files operation is success for $dbName.$tableName.")	×
187	}
188
189	/**
190	* delete partition folders recurssively
191	*
192	* @param carbonTable
193	* @param partitionSpecList
194	*/
195	def cleanUpPartitionFoldersRecurssively(carbonTable: CarbonTable,
196	partitionSpecList: List[PartitionSpec]): Unit = {
197	if (carbonTable != null) {	×
198	val loadMetadataDetails = SegmentStatusManager	×
199	.readLoadMetadata(carbonTable.getMetadataPath)	×
200
201	val fileType = FileFactory.getFileType(carbonTable.getTablePath)	×
202	val carbonFile = FileFactory.getCarbonFile(carbonTable.getTablePath, fileType)	×
203
204	// list all files from table path
205	val listOfDefaultPartFilesIterator = carbonFile.listFiles(true)	×
206	loadMetadataDetails.foreach { metadataDetail =>	×
207	if (metadataDetail.getSegmentStatus.equals(SegmentStatus.MARKED_FOR_DELETE) &&	×
208	metadataDetail.getSegmentFile == null) {	×
209	val loadStartTime: Long = metadataDetail.getLoadStartTime	×
210	// delete all files of @loadStartTime from tablepath
211	cleanCarbonFilesInFolder(listOfDefaultPartFilesIterator, loadStartTime)	×
212	partitionSpecList.foreach {	×
213	partitionSpec =>	×
214	val partitionLocation = partitionSpec.getLocation	×
215	// For partition folder outside the tablePath
216	if (!partitionLocation.toString.startsWith(carbonTable.getTablePath)) {	×
217	val fileType = FileFactory.getFileType(partitionLocation.toString)	×
218	val partitionCarbonFile = FileFactory	×
219	.getCarbonFile(partitionLocation.toString, fileType)	×
220	// list all files from partitionLoacation
221	val listOfExternalPartFilesIterator = partitionCarbonFile.listFiles(true)	×
222	// delete all files of @loadStartTime from externalPath
223	cleanCarbonFilesInFolder(listOfExternalPartFilesIterator, loadStartTime)	×
224	}
225	}
226	}
227	}
228	}
229	}
230
231	/**
232	*
233	* @param carbonFiles
234	* @param timestamp
235	*/
236	private def cleanCarbonFilesInFolder(carbonFiles: java.util.List[CarbonFile],
237	timestamp: Long): Unit = {
238	carbonFiles.asScala.foreach {	×
239	carbonFile =>	×
240	val filePath = carbonFile.getPath	×
241	val fileName = carbonFile.getName	×
242	if (CarbonTablePath.DataFileUtil.compareCarbonFileTimeStamp(fileName, timestamp)) {	×
243	// delete the file
244	FileFactory.deleteFile(filePath)	×
245	}
246	}
247	}
248
249	// validates load ids
250	private def validateLoadIds(loadids: Seq[String]): Unit = {
251	if (loadids.isEmpty) {	×
252	val errorMessage = "Error: Segment id(s) should not be empty."	×
253	throw new MalformedCarbonCommandException(errorMessage)	×
254	}
255	}
256
257	// TODO: move dbName and tableName to caller, caller should handle the log and error
258	def deleteLoadById(
259	loadids: Seq[String],
260	dbName: String,
261	tableName: String,
262	carbonTable: CarbonTable): Unit = {
263
264	LOGGER.audit(s"Delete segment by Id request has been received for $dbName.$tableName")	×
265	validateLoadIds(loadids)	×
266
267	val path = carbonTable.getMetadataPath	×
268
269	try {	×
270	val invalidLoadIds = SegmentStatusManager.updateDeletionStatus(	×
271	carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala	×
272	if (invalidLoadIds.isEmpty) {	×
273	LOGGER.audit(s"Delete segment by Id is successfull for $dbName.$tableName.")	×
274	} else {
275	sys.error(s"Delete segment by Id is failed. Invalid ID is: ${invalidLoadIds.mkString(",")}")	×
276	}
277	} catch {
278	case ex: Exception =>
279	sys.error(ex.getMessage)	×
280	}
281	Seq.empty	×
282	}
283
284	// TODO: move dbName and tableName to caller, caller should handle the log and error
285	def deleteLoadByDate(
286	timestamp: String,
287	dbName: String,
288	tableName: String,
289	carbonTable: CarbonTable): Unit = {	×
290	LOGGER.audit(s"Delete segment by Id request has been received for $dbName.$tableName")	×
291
292	val time = validateTimeFormat(timestamp)	×
293	val path = carbonTable.getMetadataPath	×
294
295	try {	×
296	val invalidLoadTimestamps =	×
297	SegmentStatusManager.updateDeletionStatus(	×
298	carbonTable.getAbsoluteTableIdentifier,	×
299	timestamp,	×
300	path,	×
301	time).asScala	×
302	if (invalidLoadTimestamps.isEmpty) {	×
303	LOGGER.audit(s"Delete segment by date is successful for $dbName.$tableName.")	×
304	} else {
305	sys.error("Delete segment by date is failed. No matching segment found.")	×
306	}
307	} catch {
308	case ex: Exception =>
309	sys.error(ex.getMessage)	×
310	}
311	}
312
313	// this function is for test only
314	def isSegmentValid(
315	dbName: String,
316	tableName: String,
317	storePath: String,
318	segmentId: String): Boolean = {
319	val identifier = AbsoluteTableIdentifier.from(storePath, dbName, tableName, tableName)	×
320	val validAndInvalidSegments: SegmentStatusManager.ValidAndInvalidSegmentsInfo = new	×
321	SegmentStatusManager(
322	identifier).getValidAndInvalidSegments	×
323	validAndInvalidSegments.getValidSegments.contains(segmentId)	×
324	}
325
326	private def validateTimeFormat(timestamp: String): Long = {
327	try {	×
328	DateTimeUtils.stringToTimestamp(UTF8String.fromString(timestamp)).get	×
329	} catch {
330	case e: Exception =>	×
331	val errorMessage = "Error: Invalid load start time format: " + timestamp	×
332	throw new MalformedCarbonCommandException(errorMessage)	×
333	}
334	}
335
336	}	×

apache / carbondata / 6651

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