| /* |
| * 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, AlterTableModel, 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.{AlterTableMergeIndexEvent, 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.isChildTableForMV) { |
| throw new UnsupportedOperationException("Cannot add partition directly on child 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 |
| val 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) |
| |
| // Normally, application will use Carbon SDK to write files into a partition folder, then |
| // add the folder to partitioned carbon table. |
| // If there are many threads writes to the same partition folder, there will be many |
| // carbon index files, and it is not good for query performance since all index files |
| // need to be read to spark driver. |
| // So, here trigger to merge the index files by sending an event |
| val alterTableModel = AlterTableModel( |
| dbName = Some(table.getDatabaseName), |
| tableName = table.getTableName, |
| segmentUpdateStatusManager = None, |
| compactionType = "", // to trigger index merge, this is not required |
| factTimeStamp = Some(System.currentTimeMillis()), |
| alterSql = null, |
| customSegmentIds = Some(Seq(loadModel.getSegmentId).toList)) |
| val mergeIndexEvent = AlterTableMergeIndexEvent(sparkSession, table, alterTableModel) |
| OperationListenerBus.getInstance.fireEvent(mergeIndexEvent, new OperationContext) |
| } |
| } |
| Seq.empty[Row] |
| } |
| |
| override protected def opName: String = "ADD HIVE PARTITION" |
| } |
