2132

Build # 2132

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/32b1e35f9f5de9b08a09ecde6f6f45bc4c2b3a87">32b1e35f9<a href="https://github.com/apache/carbondata/commit/32b1e35f9f5de9b08a09ecde6f6f45bc4c2b3a87">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/0c777259d54d3b99acd9d62b75cddde7e726afc1">0c777259d</a>">0c777259d</a><a href="https://github.com/apache/carbondata/commit/32b1e35f9f5de9b08a09ecde6f6f45bc4c2b3a87"> into bd752e9d5">bd752e9d5</a>

Run Details

61515 of 77838 relevant lines covered (79.03%)

1.05 hits per line

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

97.73

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableAddHivePartitionCommand.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.spark.sql.execution.command.partition

import java.util

import scala.collection.JavaConverters._

import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.execution.command.{AlterTableAddPartitionCommand, AlterTableDropPartitionCommand, AtomicRunnableCommand}
import org.apache.spark.sql.optimizer.CarbonFilters

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.compression.CompressorFactory
import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.metadata.SegmentFileStore
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.SegmentStatus
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{OperationContext, OperationListenerBus, PostAlterTableHivePartitionCommandEvent, PreAlterTableHivePartitionCommandEvent}
import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
import org.apache.carbondata.processing.util.CarbonLoaderUtil

/**
 * Adding the partition to the hive and create a new segment if the location has data.
 *
 */
case class CarbonAlterTableAddHivePartitionCommand(
    tableName: TableIdentifier,
    partitionSpecsAndLocs: Seq[(TablePartitionSpec, Option[String])],
    ifNotExists: Boolean)
  extends AtomicRunnableCommand {

  var partitionSpecsAndLocsTobeAdded : util.List[PartitionSpec] = _
  var table: CarbonTable = _

  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
    table = CarbonEnv.getCarbonTable(tableName)(sparkSession)
    setAuditTable(table)
    setAuditInfo(Map("partition" -> partitionSpecsAndLocs.mkString(", ")))
    if (table.isHivePartitionTable) {
      if (table.isChildDataMap) {
        throw new UnsupportedOperationException("Cannot add partition directly on aggregate tables")
      }
      val partitionWithLoc = partitionSpecsAndLocs.filter(_._2.isDefined)
      if (partitionWithLoc.nonEmpty) {
        val partitionSpecs = partitionWithLoc.map{ case (part, location) =>
          new PartitionSpec(
            new util.ArrayList(part.map(p => p._1 + "=" + p._2).toList.asJava),
            location.get)
        }
        // Get all the partitions which are not already present in hive.
        val currParts = CarbonFilters.getCurrentPartitions(sparkSession, tableName).get
        partitionSpecsAndLocsTobeAdded =
          new util.ArrayList(partitionSpecs.filterNot { part =>
          currParts.exists(p => part.equals(p))
        }.asJava)
      }
      val operationContext = new OperationContext
      val preAlterTableHivePartitionCommandEvent = PreAlterTableHivePartitionCommandEvent(
        sparkSession,
        table)
      OperationListenerBus.getInstance()
        .fireEvent(preAlterTableHivePartitionCommandEvent, operationContext)
      AlterTableAddPartitionCommand(tableName, partitionSpecsAndLocs, ifNotExists).run(sparkSession)
      val postAlterTableHivePartitionCommandEvent = PostAlterTableHivePartitionCommandEvent(
        sparkSession,
        table)
      OperationListenerBus.getInstance()
        .fireEvent(postAlterTableHivePartitionCommandEvent, operationContext)
    } else {
      throw new UnsupportedOperationException(
        "Cannot add partition directly on non partitioned table")
    }
    Seq.empty[Row]
  }


  override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
    AlterTableDropPartitionCommand(
      tableName,
      partitionSpecsAndLocs.map(_._1),
      ifExists = true,
      purge = false,
      retainData = true).run(sparkSession)
    val msg = s"Got exception $exception when processing data of add partition." +
              "Dropping partitions to the metadata"
    LogServiceFactory.getLogService(this.getClass.getCanonicalName).error(msg)
    Seq.empty[Row]
  }

  override def processData(sparkSession: SparkSession): Seq[Row] = {
    // Partitions with physical data should be registered to as a new segment.
    if (partitionSpecsAndLocsTobeAdded != null && partitionSpecsAndLocsTobeAdded.size() > 0) {
      val segmentFile = SegmentFileStore.getSegmentFileForPhysicalDataPartitions(table.getTablePath,
        partitionSpecsAndLocsTobeAdded)
      if (segmentFile != null) {
        val indexToSchemas = SegmentFileStore.getSchemaFiles(segmentFile, table.getTablePath)
        val tableColums = table.getTableInfo.getFactTable.getListOfColumns.asScala
        var isSameSchema = indexToSchemas.asScala.exists{ case(key, columnSchemas) =>
          columnSchemas.asScala.exists { col =>
            tableColums.exists(p => p.getColumnUniqueId.equals(col.getColumnUniqueId))
          } && columnSchemas.size() == tableColums.length
        }
        if (!isSameSchema) {
          throw new UnsupportedOperationException(
            "Schema of index files located in location is not matching with current table schema")
        }
        val loadModel = new CarbonLoadModel
        val columnCompressor = table.getTableInfo.getFactTable.getTableProperties.asScala
          .getOrElse(CarbonCommonConstants.COMPRESSOR,
            CompressorFactory.getInstance().getCompressor.getName)
        loadModel.setColumnCompressor(columnCompressor)
        loadModel.setCarbonTransactionalTable(true)
        loadModel.setCarbonDataLoadSchema(new CarbonDataLoadSchema(table))
        // Create new entry in tablestatus file
        CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, false)
        val newMetaEntry = loadModel.getCurrentLoadMetadataDetail
        val segmentFileName =
          SegmentFileStore.genSegmentFileName(
            loadModel.getSegmentId, String.valueOf(loadModel.getFactTimeStamp)) +
          CarbonTablePath.SEGMENT_EXT
        newMetaEntry.setSegmentFile(segmentFileName)
        val segmentsLoc = CarbonTablePath.getSegmentFilesLocation(table.getTablePath)
        CarbonUtil.checkAndCreateFolderWithPermission(segmentsLoc)
        val segmentPath = segmentsLoc + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
        SegmentFileStore.writeSegmentFile(segmentFile, segmentPath)
        CarbonLoaderUtil.populateNewLoadMetaEntry(
          newMetaEntry,
          SegmentStatus.SUCCESS,
          loadModel.getFactTimeStamp,
          true)
        // Add size to the entry
        CarbonLoaderUtil.addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId, table)
        // Make the load as success in table status
        CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false)
      }
    }
    Seq.empty[Row]
  }

  override protected def opName: String = "ADD HIVE PARTITION"
}

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.spark.sql.execution.command.partition
19
20	import java.util
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
25	import org.apache.spark.sql.catalyst.TableIdentifier
26	import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
27	import org.apache.spark.sql.execution.command.{AlterTableAddPartitionCommand, AlterTableDropPartitionCommand, AtomicRunnableCommand}
28	import org.apache.spark.sql.optimizer.CarbonFilters
29
30	import org.apache.carbondata.common.logging.LogServiceFactory
31	import org.apache.carbondata.core.constants.CarbonCommonConstants
32	import org.apache.carbondata.core.datastore.compression.CompressorFactory
33	import org.apache.carbondata.core.indexstore.PartitionSpec
34	import org.apache.carbondata.core.metadata.SegmentFileStore
35	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
36	import org.apache.carbondata.core.statusmanager.SegmentStatus
37	import org.apache.carbondata.core.util.CarbonUtil
38	import org.apache.carbondata.core.util.path.CarbonTablePath
39	import org.apache.carbondata.events.{OperationContext, OperationListenerBus, PostAlterTableHivePartitionCommandEvent, PreAlterTableHivePartitionCommandEvent}
40	import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
41	import org.apache.carbondata.processing.util.CarbonLoaderUtil
42
43	/**
44	* Adding the partition to the hive and create a new segment if the location has data.
45	*
46	*/
47	case class CarbonAlterTableAddHivePartitionCommand(	×
48	tableName: TableIdentifier,	1✔
49	partitionSpecsAndLocs: Seq[(TablePartitionSpec, Option[String])],	1✔
50	ifNotExists: Boolean)	1✔
51	extends AtomicRunnableCommand {	1✔
52
53	var partitionSpecsAndLocsTobeAdded : util.List[PartitionSpec] = _	1✔
54	var table: CarbonTable = _	1✔
55
56	override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
57	table = CarbonEnv.getCarbonTable(tableName)(sparkSession)	1✔
58	setAuditTable(table)	1✔
59	setAuditInfo(Map("partition" -> partitionSpecsAndLocs.mkString(", ")))	1✔
60	if (table.isHivePartitionTable) {	1✔
61	if (table.isChildDataMap) {	1✔
62	throw new UnsupportedOperationException("Cannot add partition directly on aggregate tables")	×
63	}
64	val partitionWithLoc = partitionSpecsAndLocs.filter(_._2.isDefined)	1✔
65	if (partitionWithLoc.nonEmpty) {	1✔
66	val partitionSpecs = partitionWithLoc.map{ case (part, location) =>	1✔
67	new PartitionSpec(	1✔
68	new util.ArrayList(part.map(p => p._1 + "=" + p._2).toList.asJava),	1✔
69	location.get)	1✔
70	}
71	// Get all the partitions which are not already present in hive.
72	val currParts = CarbonFilters.getCurrentPartitions(sparkSession, tableName).get	1✔
73	partitionSpecsAndLocsTobeAdded =	1✔
74	new util.ArrayList(partitionSpecs.filterNot { part =>	1✔
75	currParts.exists(p => part.equals(p))	1✔
76	}.asJava)	1✔
77	}
78	val operationContext = new OperationContext	1✔
79	val preAlterTableHivePartitionCommandEvent = PreAlterTableHivePartitionCommandEvent(	1✔
80	sparkSession,	1✔
81	table)	1✔
82	OperationListenerBus.getInstance()	1✔
83	.fireEvent(preAlterTableHivePartitionCommandEvent, operationContext)	1✔
84	AlterTableAddPartitionCommand(tableName, partitionSpecsAndLocs, ifNotExists).run(sparkSession)	1✔
85	val postAlterTableHivePartitionCommandEvent = PostAlterTableHivePartitionCommandEvent(	1✔
86	sparkSession,	1✔
87	table)	1✔
88	OperationListenerBus.getInstance()	1✔
89	.fireEvent(postAlterTableHivePartitionCommandEvent, operationContext)	1✔
90	} else {
91	throw new UnsupportedOperationException(	1✔
92	"Cannot add partition directly on non partitioned table")	1✔
93	}
94	Seq.empty[Row]	1✔
95	}
96
97
98	override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
99	AlterTableDropPartitionCommand(	1✔
100	tableName,	1✔
101	partitionSpecsAndLocs.map(_._1),	1✔
102	ifExists = true,	1✔
103	purge = false,	1✔
104	retainData = true).run(sparkSession)	1✔
105	val msg = s"Got exception $exception when processing data of add partition." +	1✔
106	"Dropping partitions to the metadata"	1✔
107	LogServiceFactory.getLogService(this.getClass.getCanonicalName).error(msg)	1✔
108	Seq.empty[Row]	1✔
109	}
110
111	override def processData(sparkSession: SparkSession): Seq[Row] = {
112	// Partitions with physical data should be registered to as a new segment.
113	if (partitionSpecsAndLocsTobeAdded != null && partitionSpecsAndLocsTobeAdded.size() > 0) {	1✔
114	val segmentFile = SegmentFileStore.getSegmentFileForPhysicalDataPartitions(table.getTablePath,	1✔
115	partitionSpecsAndLocsTobeAdded)	1✔
116	if (segmentFile != null) {	1✔
117	val indexToSchemas = SegmentFileStore.getSchemaFiles(segmentFile, table.getTablePath)	1✔
118	val tableColums = table.getTableInfo.getFactTable.getListOfColumns.asScala	1✔
119	var isSameSchema = indexToSchemas.asScala.exists{ case(key, columnSchemas) =>	1✔
120	columnSchemas.asScala.exists { col =>	1✔
121	tableColums.exists(p => p.getColumnUniqueId.equals(col.getColumnUniqueId))	1✔
122	} && columnSchemas.size() == tableColums.length	1✔
123	}
124	if (!isSameSchema) {	1✔
125	throw new UnsupportedOperationException(	1✔
126	"Schema of index files located in location is not matching with current table schema")	1✔
127	}
128	val loadModel = new CarbonLoadModel	1✔
129	val columnCompressor = table.getTableInfo.getFactTable.getTableProperties.asScala	1✔
130	.getOrElse(CarbonCommonConstants.COMPRESSOR,	1✔
131	CompressorFactory.getInstance().getCompressor.getName)	1✔
132	loadModel.setColumnCompressor(columnCompressor)	1✔
133	loadModel.setCarbonTransactionalTable(true)	1✔
134	loadModel.setCarbonDataLoadSchema(new CarbonDataLoadSchema(table))	1✔
135	// Create new entry in tablestatus file
136	CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, false)	1✔
137	val newMetaEntry = loadModel.getCurrentLoadMetadataDetail	1✔
138	val segmentFileName =
139	SegmentFileStore.genSegmentFileName(	1✔
140	loadModel.getSegmentId, String.valueOf(loadModel.getFactTimeStamp)) +	1✔
141	CarbonTablePath.SEGMENT_EXT	1✔
142	newMetaEntry.setSegmentFile(segmentFileName)	1✔
143	val segmentsLoc = CarbonTablePath.getSegmentFilesLocation(table.getTablePath)	1✔
144	CarbonUtil.checkAndCreateFolderWithPermission(segmentsLoc)	1✔
145	val segmentPath = segmentsLoc + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName	1✔
146	SegmentFileStore.writeSegmentFile(segmentFile, segmentPath)	1✔
147	CarbonLoaderUtil.populateNewLoadMetaEntry(	1✔
148	newMetaEntry,	1✔
149	SegmentStatus.SUCCESS,	1✔
150	loadModel.getFactTimeStamp,	1✔
151	true)	1✔
152	// Add size to the entry
153	CarbonLoaderUtil.addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId, table)	1✔
154	// Make the load as success in table status
155	CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false)	1✔
156	}
157	}
158	Seq.empty[Row]	1✔
159	}
160
161	override protected def opName: String = "ADD HIVE PARTITION"	1✔
162	}

apache / carbondata / 2132

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