apache / carbondata / 813

Build # 813

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/668ec60ea0f913d8e2e83a8e11fb48cfd8344ff7">668ec60ea<a href="https://github.com/apache/carbondata/commit/668ec60ea0f913d8e2e83a8e11fb48cfd8344ff7">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/25ab0270021ec1c58219690c7389003c553768bf">25ab02700</a>">25ab02700</a><a href="https://github.com/apache/carbondata/commit/668ec60ea0f913d8e2e83a8e11fb48cfd8344ff7"> into 8427771fc">8427771fc</a>

Run Details

35 of 35 new or added lines in 1 file covered. (100.0%)

60356 of 75076 relevant lines covered (80.39%)

1.06 hits per line

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

97.65

/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)
    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]
  }

}

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	if (table.isHivePartitionTable) {	1✔
59	if (table.isChildDataMap) {	1✔
60	throw new UnsupportedOperationException("Cannot add partition directly on aggregate tables")	×
61	}
62	val partitionWithLoc = partitionSpecsAndLocs.filter(_._2.isDefined)	1✔
63	if (partitionWithLoc.nonEmpty) {	1✔
64	val partitionSpecs = partitionWithLoc.map{ case (part, location) =>	1✔
65	new PartitionSpec(	1✔
66	new util.ArrayList(part.map(p => p._1 + "=" + p._2).toList.asJava),	1✔
67	location.get)	1✔
68	}
69	// Get all the partitions which are not already present in hive.
70	val currParts = CarbonFilters.getCurrentPartitions(sparkSession, tableName).get	1✔
71	partitionSpecsAndLocsTobeAdded =	1✔
72	new util.ArrayList(partitionSpecs.filterNot { part =>	1✔
73	currParts.exists(p => part.equals(p))	1✔
74	}.asJava)	1✔
75	}
76	val operationContext = new OperationContext	1✔
77	val preAlterTableHivePartitionCommandEvent = PreAlterTableHivePartitionCommandEvent(	1✔
78	sparkSession,	1✔
79	table)	1✔
80	OperationListenerBus.getInstance()	1✔
81	.fireEvent(preAlterTableHivePartitionCommandEvent, operationContext)	1✔
82	AlterTableAddPartitionCommand(tableName, partitionSpecsAndLocs, ifNotExists).run(sparkSession)	1✔
83	val postAlterTableHivePartitionCommandEvent = PostAlterTableHivePartitionCommandEvent(	1✔
84	sparkSession,	1✔
85	table)	1✔
86	OperationListenerBus.getInstance()	1✔
87	.fireEvent(postAlterTableHivePartitionCommandEvent, operationContext)	1✔
88	} else {
89	throw new UnsupportedOperationException(	1✔
90	"Cannot add partition directly on non partitioned table")	1✔
91	}
92	Seq.empty[Row]	1✔
93	}
94
95
96	override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
97	AlterTableDropPartitionCommand(	1✔
98	tableName,	1✔
99	partitionSpecsAndLocs.map(_._1),	1✔
100	ifExists = true,	1✔
101	purge = false,	1✔
102	retainData = true).run(sparkSession)	1✔
103	val msg = s"Got exception $exception when processing data of add partition." +	1✔
104	"Dropping partitions to the metadata"	1✔
105	LogServiceFactory.getLogService(this.getClass.getCanonicalName).error(msg)	1✔
106	Seq.empty[Row]	1✔
107	}
108
109	override def processData(sparkSession: SparkSession): Seq[Row] = {
110	// Partitions with physical data should be registered to as a new segment.
111	if (partitionSpecsAndLocsTobeAdded != null && partitionSpecsAndLocsTobeAdded.size() > 0) {	1✔
112	val segmentFile = SegmentFileStore.getSegmentFileForPhysicalDataPartitions(table.getTablePath,	1✔
113	partitionSpecsAndLocsTobeAdded)	1✔
114	if (segmentFile != null) {	1✔
115	val indexToSchemas = SegmentFileStore.getSchemaFiles(segmentFile, table.getTablePath)	1✔
116	val tableColums = table.getTableInfo.getFactTable.getListOfColumns.asScala	1✔
117	var isSameSchema = indexToSchemas.asScala.exists{ case(key, columnSchemas) =>	1✔
118	columnSchemas.asScala.exists { col =>	1✔
119	tableColums.exists(p => p.getColumnUniqueId.equals(col.getColumnUniqueId))	1✔
120	} && columnSchemas.size() == tableColums.length	1✔
121	}
122	if (!isSameSchema) {	1✔
123	throw new UnsupportedOperationException(	1✔
124	"Schema of index files located in location is not matching with current table schema")	1✔
125	}
126	val loadModel = new CarbonLoadModel	1✔
127	val columnCompressor = table.getTableInfo.getFactTable.getTableProperties.asScala	1✔
128	.getOrElse(CarbonCommonConstants.COMPRESSOR,	1✔
129	CompressorFactory.getInstance().getCompressor.getName)	1✔
130	loadModel.setColumnCompressor(columnCompressor)	1✔
131	loadModel.setCarbonTransactionalTable(true)	1✔
132	loadModel.setCarbonDataLoadSchema(new CarbonDataLoadSchema(table))	1✔
133	// Create new entry in tablestatus file
134	CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, false)	1✔
135	val newMetaEntry = loadModel.getCurrentLoadMetadataDetail	1✔
136	val segmentFileName =
137	SegmentFileStore.genSegmentFileName(	1✔
138	loadModel.getSegmentId, String.valueOf(loadModel.getFactTimeStamp)) +	1✔
139	CarbonTablePath.SEGMENT_EXT	1✔
140	newMetaEntry.setSegmentFile(segmentFileName)	1✔
141	val segmentsLoc = CarbonTablePath.getSegmentFilesLocation(table.getTablePath)	1✔
142	CarbonUtil.checkAndCreateFolderWithPermission(segmentsLoc)	1✔
143	val segmentPath = segmentsLoc + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName	1✔
144	SegmentFileStore.writeSegmentFile(segmentFile, segmentPath)	1✔
145	CarbonLoaderUtil.populateNewLoadMetaEntry(	1✔
146	newMetaEntry,	1✔
147	SegmentStatus.SUCCESS,	1✔
148	loadModel.getFactTimeStamp,	1✔
149	true)	1✔
150	// Add size to the entry
151	CarbonLoaderUtil.addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId, table)	1✔
152	// Make the load as success in table status
153	CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false)	1✔
154	}
155	}
156	Seq.empty[Row]	1✔
157	}
158
159	}

apache / carbondata / 813

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