| /* |
| * 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.datamap |
| |
| import scala.collection.JavaConverters._ |
| |
| import org.apache.spark.sql._ |
| import org.apache.spark.sql.catalyst.TableIdentifier |
| import org.apache.spark.sql.execution.command._ |
| |
| import org.apache.carbondata.common.exceptions.sql.{MalformedCarbonCommandException, MalformedDataMapCommandException} |
| import org.apache.carbondata.common.logging.LogServiceFactory |
| import org.apache.carbondata.core.datamap.{DataMapProvider, DataMapStoreManager} |
| import org.apache.carbondata.core.datamap.status.DataMapStatusManager |
| import org.apache.carbondata.core.metadata.ColumnarFormatVersion |
| import org.apache.carbondata.core.metadata.datatype.DataTypes |
| import org.apache.carbondata.core.metadata.schema.datamap.{DataMapClassProvider, DataMapProperty} |
| import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema} |
| import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil} |
| import org.apache.carbondata.datamap.{DataMapManager, IndexDataMapProvider} |
| import org.apache.carbondata.events._ |
| |
| /** |
| * Below command class will be used to create datamap on table |
| * and updating the parent table about the datamap information |
| */ |
| case class CarbonCreateDataMapCommand( |
| dataMapName: String, |
| tableIdentifier: Option[TableIdentifier], |
| dmProviderName: String, |
| dmProperties: Map[String, String], |
| queryString: Option[String], |
| ifNotExistsSet: Boolean = false, |
| var deferredRebuild: Boolean = false) |
| extends AtomicRunnableCommand { |
| |
| private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) |
| private var dataMapProvider: DataMapProvider = _ |
| private var mainTable: CarbonTable = _ |
| private var dataMapSchema: DataMapSchema = _ |
| |
| override def processMetadata(sparkSession: SparkSession): Seq[Row] = { |
| // since streaming segment does not support building index and pre-aggregate yet, |
| // so streaming table does not support create datamap |
| mainTable = tableIdentifier match { |
| case Some(table) => |
| CarbonEnv.getCarbonTable(table.database, table.table)(sparkSession) |
| case _ => null |
| } |
| |
| if (mainTable != null) { |
| setAuditTable(mainTable) |
| } |
| setAuditInfo(Map("provider" -> dmProviderName, "dmName" -> dataMapName) ++ dmProperties) |
| |
| if (mainTable != null && !mainTable.getTableInfo.isTransactionalTable) { |
| throw new MalformedCarbonCommandException("Unsupported operation on non transactional table") |
| } |
| |
| if (mainTable != null && mainTable.getDataMapSchema(dataMapName) != null) { |
| if (!ifNotExistsSet) { |
| throw new MalformedDataMapCommandException(s"DataMap name '$dataMapName' already exist") |
| } else { |
| return Seq.empty |
| } |
| } |
| |
| if (mainTable != null && CarbonUtil.getFormatVersion(mainTable) != ColumnarFormatVersion.V3) { |
| throw new MalformedCarbonCommandException(s"Unsupported operation on table with " + |
| s"V1 or V2 format data") |
| } |
| |
| dataMapSchema = new DataMapSchema(dataMapName, dmProviderName) |
| |
| val property = dmProperties.map(x => (x._1.trim, x._2.trim)).asJava |
| val javaMap = new java.util.HashMap[String, String](property) |
| javaMap.put(DataMapProperty.DEFERRED_REBUILD, deferredRebuild.toString) |
| dataMapSchema.setProperties(javaMap) |
| |
| if (dataMapSchema.isIndexDataMap && mainTable == null) { |
| throw new MalformedDataMapCommandException( |
| "For this datamap, main table is required. Use `CREATE DATAMAP ... ON TABLE ...` ") |
| } |
| dataMapProvider = DataMapManager.get.getDataMapProvider(mainTable, dataMapSchema, sparkSession) |
| if (deferredRebuild && !dataMapProvider.supportRebuild()) { |
| throw new MalformedDataMapCommandException( |
| s"DEFERRED REBUILD is not supported on this datamap $dataMapName" + |
| s" with provider ${dataMapSchema.getProviderName}") |
| } |
| |
| if (null != mainTable) { |
| if (mainTable.isChildTableForMV) { |
| throw new MalformedDataMapCommandException( |
| "Cannot create DataMap on child table " + mainTable.getTableUniqueName) |
| } |
| } |
| if (!dataMapSchema.isIndexDataMap) { |
| if (DataMapStoreManager.getInstance().getAllDataMapSchemas.asScala |
| .exists(_.getDataMapName.equalsIgnoreCase(dataMapSchema.getDataMapName))) { |
| if (!ifNotExistsSet) { |
| throw new MalformedDataMapCommandException( |
| "DataMap with name " + dataMapSchema.getDataMapName + " already exists in storage") |
| } else { |
| return Seq.empty |
| } |
| } |
| } |
| |
| val systemFolderLocation: String = CarbonProperties.getInstance().getSystemFolderLocation |
| val operationContext: OperationContext = new OperationContext() |
| |
| // If it is index datamap, check whether the column has datamap created already |
| dataMapProvider match { |
| case provider: IndexDataMapProvider => |
| val isBloomFilter = DataMapClassProvider.BLOOMFILTER.getShortName |
| .equalsIgnoreCase(dmProviderName) |
| val datamaps = DataMapStoreManager.getInstance.getAllDataMap(mainTable).asScala |
| val thisDmProviderName = |
| dataMapProvider.asInstanceOf[IndexDataMapProvider].getDataMapSchema.getProviderName |
| val existingIndexColumn4ThisProvider = datamaps.filter { datamap => |
| thisDmProviderName.equalsIgnoreCase(datamap.getDataMapSchema.getProviderName) |
| }.flatMap { datamap => |
| datamap.getDataMapSchema.getIndexColumns |
| }.distinct |
| |
| provider.getIndexedColumns.asScala.foreach { column => |
| if (existingIndexColumn4ThisProvider.contains(column.getColName)) { |
| throw new MalformedDataMapCommandException(String.format( |
| "column '%s' already has %s index datamap created", |
| column.getColName, thisDmProviderName)) |
| } else if (isBloomFilter) { |
| if (column.getDataType == DataTypes.BINARY) { |
| throw new MalformedDataMapCommandException( |
| s"BloomFilter datamap does not support Binary datatype column: ${ |
| column.getColName |
| }") |
| } |
| // if datamap provider is bloomfilter,the index column datatype cannot be complex type |
| if (column.isComplex) { |
| throw new MalformedDataMapCommandException( |
| s"BloomFilter datamap does not support complex datatype column: ${ |
| column.getColName |
| }") |
| } |
| } |
| } |
| |
| val createDataMapPreExecutionEvent: CreateDataMapPreExecutionEvent = |
| new CreateDataMapPreExecutionEvent(sparkSession, |
| systemFolderLocation, tableIdentifier.get) |
| OperationListenerBus.getInstance().fireEvent(createDataMapPreExecutionEvent, |
| operationContext) |
| dataMapProvider.initMeta(queryString.orNull) |
| DataMapStatusManager.disableDataMap(dataMapName) |
| case _ => |
| val createDataMapPreExecutionEvent: CreateDataMapPreExecutionEvent = |
| CreateDataMapPreExecutionEvent(sparkSession, |
| systemFolderLocation, tableIdentifier.orNull) |
| OperationListenerBus.getInstance().fireEvent(createDataMapPreExecutionEvent, |
| operationContext) |
| dataMapProvider.initMeta(queryString.orNull) |
| } |
| val createDataMapPostExecutionEvent: CreateDataMapPostExecutionEvent = |
| CreateDataMapPostExecutionEvent(sparkSession, |
| systemFolderLocation, tableIdentifier, dmProviderName) |
| OperationListenerBus.getInstance().fireEvent(createDataMapPostExecutionEvent, |
| operationContext) |
| Seq.empty |
| } |
| |
| override def processData(sparkSession: SparkSession): Seq[Row] = { |
| if (dataMapProvider != null) { |
| dataMapProvider.initData() |
| // TODO: remove these checks once the preaggregate and preaggregate timeseries are deprecated |
| if (mainTable != null && !deferredRebuild && dataMapSchema.isIndexDataMap) { |
| dataMapProvider.rebuild() |
| if (dataMapSchema.isIndexDataMap) { |
| val operationContext: OperationContext = new OperationContext() |
| val systemFolderLocation: String = CarbonProperties.getInstance().getSystemFolderLocation |
| val updateDataMapPreExecutionEvent: UpdateDataMapPreExecutionEvent = |
| new UpdateDataMapPreExecutionEvent(sparkSession, |
| systemFolderLocation, tableIdentifier.get) |
| OperationListenerBus.getInstance().fireEvent(updateDataMapPreExecutionEvent, |
| operationContext) |
| DataMapStatusManager.enableDataMap(dataMapName) |
| val updateDataMapPostExecutionEvent: UpdateDataMapPostExecutionEvent = |
| new UpdateDataMapPostExecutionEvent(sparkSession, |
| systemFolderLocation, tableIdentifier.get) |
| OperationListenerBus.getInstance().fireEvent(updateDataMapPostExecutionEvent, |
| operationContext) |
| } |
| } |
| if (null != dataMapSchema.getRelationIdentifier && !dataMapSchema.isIndexDataMap && |
| !dataMapSchema.isLazy) { |
| DataMapStatusManager.enableDataMap(dataMapName) |
| } |
| } |
| Seq.empty |
| } |
| |
| override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = { |
| if (dataMapProvider != null) { |
| val table = |
| if (mainTable != null) { |
| Some(TableIdentifier(mainTable.getTableName, Some(mainTable.getDatabaseName))) |
| } else { |
| None |
| } |
| CarbonDropDataMapCommand( |
| dataMapName, |
| true, |
| table, |
| forceDrop = false).run(sparkSession) |
| } |
| Seq.empty |
| } |
| |
| override protected def opName: String = "CREATE DATAMAP" |
| } |
| |
