apache / carbondata / 46

Build # 46

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/2e4ac74431d29556a8b5e6da0a99dd4506a0a3b4">2e4ac7443<a href="https://github.com/apache/carbondata/commit/2e4ac74431d29556a8b5e6da0a99dd4506a0a3b4">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/d402633d0363a2e36b57872f3a551d366d21dfe8">d402633d0</a>">d402633d0</a><a href="https://github.com/apache/carbondata/commit/2e4ac74431d29556a8b5e6da0a99dd4506a0a3b4"> into 2a9604cd8">2a9604cd8</a>

Run Details

40 of 40 new or added lines in 6 files covered. (100.0%)

52938 of 71137 relevant lines covered (74.42%)

0.89 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/spark/rdd/PartitionDropper.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.spark.rdd

import java.io.IOException

import org.apache.spark.sql.execution.command.{AlterPartitionModel, DropPartitionCallableModel}
import org.apache.spark.util.PartitionUtils

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.metadata.schema.partition.PartitionType
import org.apache.carbondata.spark.{AlterPartitionResultImpl, PartitionFactory}

object PartitionDropper {

  val logger = LogServiceFactory.getLogService(PartitionDropper.getClass.getName)

  def triggerPartitionDrop(dropPartitionCallableModel: DropPartitionCallableModel): Unit = {
    val alterPartitionModel = new AlterPartitionModel(dropPartitionCallableModel.carbonLoadModel,
      dropPartitionCallableModel.segmentId.getSegmentNo,
      dropPartitionCallableModel.oldPartitionIds,
      dropPartitionCallableModel.sqlContext
    )
    val partitionId = dropPartitionCallableModel.partitionId
    val oldPartitionIds = dropPartitionCallableModel.oldPartitionIds
    val dropWithData = dropPartitionCallableModel.dropWithData
    val carbonTable = dropPartitionCallableModel.carbonTable
    val dbName = carbonTable.getDatabaseName
    val tableName = carbonTable.getTableName
    val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
    val partitionInfo = carbonTable.getPartitionInfo(tableName)
    val partitioner = PartitionFactory.getPartitioner(partitionInfo)

    var finalDropStatus = false
    val bucketInfo = carbonTable.getBucketingInfo(tableName)
    val bucketNumber = bucketInfo match {
      case null => 1
      case _ => bucketInfo.getNumOfRanges
    }
    val partitionIndex = oldPartitionIds.indexOf(Integer.valueOf(partitionId))
    val targetPartitionId = partitionInfo.getPartitionType match {
      case PartitionType.RANGE => if (partitionIndex == oldPartitionIds.length - 1) {
        "0"
      } else {
        String.valueOf(oldPartitionIds(partitionIndex + 1))
      }
      case PartitionType.LIST => "0"
    }

    if (!dropWithData) {
      try {
        for (i <- 0 until bucketNumber) {
          val bucketId = i
          val rdd = new CarbonScanPartitionRDD(alterPartitionModel,
            absoluteTableIdentifier,
            Seq(partitionId, targetPartitionId),
            bucketId
          ).partitionBy(partitioner).map(_._2)

          val dropStatus = new AlterTableLoadPartitionRDD(alterPartitionModel,
            new AlterPartitionResultImpl(),
            Seq(partitionId),
            bucketId,
            absoluteTableIdentifier,
            rdd).collect()

          if (dropStatus.length == 0) {
            finalDropStatus = false
          } else {
            finalDropStatus = dropStatus.forall(_._2)
          }
          if (!finalDropStatus) {
            logger.audit(s"Drop Partition request failed for table " +
                         s"${ dbName }.${ tableName }")
            logger.error(s"Drop Partition request failed for table " +
                         s"${ dbName }.${ tableName }")
          }
        }

        if (finalDropStatus) {
          try {
            PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,
              Seq(partitionId, targetPartitionId).toList, dbName,
              tableName, partitionInfo)
          } catch {
            case e: IOException => sys.error(s"Exception while delete original carbon files " +
                                             e.getMessage)
          }
          logger.audit(s"Drop Partition request completed for table " +
                       s"${ dbName }.${ tableName }")
          logger.info(s"Drop Partition request completed for table " +
                      s"${ dbName }.${ tableName }")
        }
      } catch {
        case e: Exception => sys.error(s"Exception in dropping partition action: ${ e.getMessage }")
      }
    } else {
      PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,
        Seq(partitionId).toList, dbName, tableName, partitionInfo)
      logger.audit(s"Drop Partition request completed for table " +
                   s"${ dbName }.${ tableName }")
      logger.info(s"Drop Partition request completed for table " +
                  s"${ dbName }.${ tableName }")
    }
  }
}

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.spark.rdd
19
20	import java.io.IOException
21
22	import org.apache.spark.sql.execution.command.{AlterPartitionModel, DropPartitionCallableModel}
23	import org.apache.spark.util.PartitionUtils
24
25	import org.apache.carbondata.common.logging.LogServiceFactory
26	import org.apache.carbondata.core.metadata.schema.partition.PartitionType
27	import org.apache.carbondata.spark.{AlterPartitionResultImpl, PartitionFactory}
28
29	object PartitionDropper {
30
31	val logger = LogServiceFactory.getLogService(PartitionDropper.getClass.getName)	×
32
33	def triggerPartitionDrop(dropPartitionCallableModel: DropPartitionCallableModel): Unit = {	×
34	val alterPartitionModel = new AlterPartitionModel(dropPartitionCallableModel.carbonLoadModel,	×
35	dropPartitionCallableModel.segmentId.getSegmentNo,	×
36	dropPartitionCallableModel.oldPartitionIds,	×
37	dropPartitionCallableModel.sqlContext	×
38	)
39	val partitionId = dropPartitionCallableModel.partitionId	×
40	val oldPartitionIds = dropPartitionCallableModel.oldPartitionIds	×
41	val dropWithData = dropPartitionCallableModel.dropWithData	×
42	val carbonTable = dropPartitionCallableModel.carbonTable	×
43	val dbName = carbonTable.getDatabaseName	×
44	val tableName = carbonTable.getTableName	×
45	val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier	×
46	val partitionInfo = carbonTable.getPartitionInfo(tableName)	×
47	val partitioner = PartitionFactory.getPartitioner(partitionInfo)	×
48
49	var finalDropStatus = false	×
50	val bucketInfo = carbonTable.getBucketingInfo(tableName)	×
51	val bucketNumber = bucketInfo match {	×
52	case null => 1	×
53	case _ => bucketInfo.getNumOfRanges	×
54	}
55	val partitionIndex = oldPartitionIds.indexOf(Integer.valueOf(partitionId))	×
56	val targetPartitionId = partitionInfo.getPartitionType match {	×
57	case PartitionType.RANGE => if (partitionIndex == oldPartitionIds.length - 1) {	×
58	"0"	×
59	} else {
60	String.valueOf(oldPartitionIds(partitionIndex + 1))	×
61	}
62	case PartitionType.LIST => "0"	×
63	}
64
65	if (!dropWithData) {	×
66	try {	×
67	for (i <- 0 until bucketNumber) {	×
68	val bucketId = i	×
69	val rdd = new CarbonScanPartitionRDD(alterPartitionModel,	×
70	absoluteTableIdentifier,	×
71	Seq(partitionId, targetPartitionId),	×
72	bucketId	×
73	).partitionBy(partitioner).map(_._2)	×
74
75	val dropStatus = new AlterTableLoadPartitionRDD(alterPartitionModel,	×
76	new AlterPartitionResultImpl(),	×
77	Seq(partitionId),	×
78	bucketId,	×
79	absoluteTableIdentifier,	×
80	rdd).collect()	×
81
82	if (dropStatus.length == 0) {	×
83	finalDropStatus = false	×
84	} else {
85	finalDropStatus = dropStatus.forall(_._2)	×
86	}
87	if (!finalDropStatus) {	×
88	logger.audit(s"Drop Partition request failed for table " +	×
89	s"${ dbName }.${ tableName }")	×
90	logger.error(s"Drop Partition request failed for table " +	×
91	s"${ dbName }.${ tableName }")	×
92	}
93	}
94
95	if (finalDropStatus) {	×
96	try {	×
97	PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,	×
98	Seq(partitionId, targetPartitionId).toList, dbName,	×
99	tableName, partitionInfo)	×
100	} catch {
101	case e: IOException => sys.error(s"Exception while delete original carbon files " +	×
102	e.getMessage)	×
103	}
104	logger.audit(s"Drop Partition request completed for table " +	×
105	s"${ dbName }.${ tableName }")	×
106	logger.info(s"Drop Partition request completed for table " +	×
107	s"${ dbName }.${ tableName }")	×
108	}
109	} catch {
110	case e: Exception => sys.error(s"Exception in dropping partition action: ${ e.getMessage }")
111	}
112	} else {
113	PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,	×
114	Seq(partitionId).toList, dbName, tableName, partitionInfo)	×
115	logger.audit(s"Drop Partition request completed for table " +	×
116	s"${ dbName }.${ tableName }")	×
117	logger.info(s"Drop Partition request completed for table " +	×
118	s"${ dbName }.${ tableName }")	×
119	}
120	}
121	}	×

apache / carbondata / 46

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