| /* |
| * 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 |
| |
| import java.util |
| import java.util.UUID |
| |
| import scala.collection.JavaConverters._ |
| import scala.collection.mutable |
| |
| import org.apache.spark.SparkContext |
| import org.apache.spark.sql.SQLContext |
| import org.apache.spark.sql.catalyst.TableIdentifier |
| import org.apache.spark.sql.util.CarbonException |
| |
| import org.apache.carbondata.common.logging.LogServiceFactory |
| import org.apache.carbondata.core.constants.CarbonCommonConstants |
| import org.apache.carbondata.core.datamap.Segment |
| import org.apache.carbondata.core.indexstore.PartitionSpec |
| import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier |
| import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes, DecimalType} |
| import org.apache.carbondata.core.metadata.encoder.Encoding |
| import org.apache.carbondata.core.metadata.schema._ |
| import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, RelationIdentifier, TableInfo, TableSchema} |
| import org.apache.carbondata.core.metadata.schema.table.column.{ColumnSchema, ParentColumnTableRelation} |
| import org.apache.carbondata.core.service.impl.ColumnUniqueIdGenerator |
| import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentUpdateStatusManager} |
| import org.apache.carbondata.core.util.{CarbonUtil, DataTypeUtil} |
| import org.apache.carbondata.processing.loading.FailureCauses |
| import org.apache.carbondata.processing.loading.model.CarbonLoadModel |
| import org.apache.carbondata.processing.merger.CompactionType |
| import org.apache.carbondata.spark.CarbonSparkFactory |
| import org.apache.carbondata.spark.util.{CarbonScalaUtil, DataTypeConverterUtil} |
| |
| case class TableModel( |
| ifNotExistsSet: Boolean, |
| databaseNameOp: Option[String], |
| tableName: String, |
| tableProperties: Map[String, String], |
| dimCols: Seq[Field], |
| msrCols: Seq[Field], |
| sortKeyDims: Option[Seq[String]], |
| varcharCols: Option[Seq[String]], |
| highcardinalitydims: Option[Seq[String]], |
| noInvertedIdxCols: Option[Seq[String]], |
| innvertedIdxCols: Option[Seq[String]], |
| colProps: Option[util.Map[String, util.List[ColumnProperty]]] = None, |
| bucketFields: Option[BucketFields], |
| partitionInfo: Option[PartitionInfo], |
| tableComment: Option[String] = None, |
| var parentTable: Option[CarbonTable] = None, |
| var dataMapRelation: Option[scala.collection.mutable.LinkedHashMap[Field, DataMapField]] = None) |
| |
| case class Field(column: String, var dataType: Option[String], name: Option[String], |
| children: Option[List[Field]], parent: String = null, |
| storeType: Option[String] = Some("columnar"), |
| var schemaOrdinal: Int = -1, |
| var precision: Int = 0, var scale: Int = 0, var rawSchema: String = "", |
| var columnComment: String = "") { |
| override def equals(o: Any) : Boolean = o match { |
| case that: Field => |
| that.column.equalsIgnoreCase(this.column) |
| case _ => false |
| } |
| override def hashCode : Int = column.hashCode |
| } |
| |
| case class DataMapField(var aggregateFunction: String = "", |
| columnTableRelationList: Option[Seq[ColumnTableRelation]] = None) { |
| } |
| |
| case class ColumnTableRelation(parentColumnName: String, parentColumnId: String, |
| parentTableName: String, parentDatabaseName: String, parentTableId: String) { |
| } |
| |
| case class ColumnProperty(key: String, value: String) |
| |
| case class ComplexField(complexType: String, primitiveField: Option[Field], |
| complexField: Option[ComplexField]) |
| |
| case class PartitionerField(partitionColumn: String, dataType: Option[String], |
| columnComment: String) |
| |
| case class BucketFields(bucketColumns: Seq[String], numberOfBuckets: Int) |
| |
| case class DataLoadTableFileMapping(table: String, loadPath: String) |
| |
| case class ExecutionErrors(var failureCauses: FailureCauses, var errorMsg: String ) |
| |
| case class CarbonMergerMapping( |
| hdfsStoreLocation: String, |
| metadataFilePath: String, |
| var mergedLoadName: String, |
| databaseName: String, |
| factTableName: String, |
| validSegments: Array[Segment], |
| tableId: String, |
| campactionType: CompactionType, |
| // maxSegmentColCardinality is Cardinality of last segment of compaction |
| var maxSegmentColCardinality: Array[Int], |
| // maxSegmentColumnSchemaList is list of column schema of last segment of compaction |
| var maxSegmentColumnSchemaList: List[ColumnSchema], |
| @transient currentPartitions: Option[Seq[PartitionSpec]]) |
| |
| case class NodeInfo(TaskId: String, noOfBlocks: Int) |
| |
| case class AlterTableModel( |
| dbName: Option[String], |
| tableName: String, |
| segmentUpdateStatusManager: Option[SegmentUpdateStatusManager], |
| compactionType: String, |
| factTimeStamp: Option[Long], |
| var alterSql: String, |
| customSegmentIds: Option[List[String]] = None) |
| |
| case class UpdateTableModel( |
| isUpdate: Boolean, |
| updatedTimeStamp: Long, |
| var executorErrors: ExecutionErrors, |
| deletedSegments: Seq[Segment]) |
| |
| case class CompactionModel(compactionSize: Long, |
| compactionType: CompactionType, |
| carbonTable: CarbonTable, |
| isDDLTrigger: Boolean, |
| currentPartitions: Option[Seq[PartitionSpec]], |
| customSegmentIds: Option[List[String]]) |
| |
| case class CompactionCallableModel(carbonLoadModel: CarbonLoadModel, |
| carbonTable: CarbonTable, |
| loadsToMerge: util.List[LoadMetadataDetails], |
| sqlContext: SQLContext, |
| compactionType: CompactionType, |
| currentPartitions: Option[Seq[PartitionSpec]], |
| compactedSegments: java.util.List[String]) |
| |
| case class AlterPartitionModel(carbonLoadModel: CarbonLoadModel, |
| segmentId: String, |
| oldPartitionIds: List[Int], |
| sqlContext: SQLContext |
| ) |
| |
| case class SplitPartitionCallableModel(carbonLoadModel: CarbonLoadModel, |
| segmentId: String, |
| partitionId: String, |
| oldPartitionIds: List[Int], |
| sqlContext: SQLContext) |
| |
| case class DropPartitionCallableModel(carbonLoadModel: CarbonLoadModel, |
| segmentId: Segment, |
| partitionId: String, |
| oldPartitionIds: List[Int], |
| dropWithData: Boolean, |
| carbonTable: CarbonTable, |
| sqlContext: SQLContext) |
| |
| case class DataTypeInfo(dataType: String, precision: Int = 0, scale: Int = 0) |
| |
| class AlterTableColumnRenameModel(columnName: String, |
| newColumnName: String, |
| isColumnRename: Boolean) |
| |
| case class AlterTableDataTypeChangeModel(dataTypeInfo: DataTypeInfo, |
| databaseName: Option[String], |
| tableName: String, |
| columnName: String, |
| newColumnName: String, |
| isColumnRename: Boolean) |
| extends AlterTableColumnRenameModel(columnName, newColumnName, isColumnRename) |
| |
| case class AlterTableRenameModel( |
| oldTableIdentifier: TableIdentifier, |
| newTableIdentifier: TableIdentifier |
| ) |
| |
| case class AlterTableAddColumnsModel( |
| databaseName: Option[String], |
| tableName: String, |
| tableProperties: Map[String, String], |
| dimCols: Seq[Field], |
| msrCols: Seq[Field], |
| highCardinalityDims: Seq[String]) |
| |
| case class AlterTableDropColumnModel(databaseName: Option[String], |
| tableName: String, |
| columns: List[String]) |
| |
| case class AlterTableDropPartitionModel(databaseName: Option[String], |
| tableName: String, |
| partitionId: String, |
| dropWithData: Boolean) |
| |
| case class AlterTableSplitPartitionModel(databaseName: Option[String], |
| tableName: String, |
| partitionId: String, |
| splitInfo: List[String]) |
| |
| class AlterTableColumnSchemaGenerator( |
| alterTableModel: AlterTableAddColumnsModel, |
| dbName: String, |
| tableInfo: TableInfo, |
| tableIdentifier: AbsoluteTableIdentifier, |
| sc: SparkContext) { |
| |
| private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) |
| |
| private def isSortColumn(columnName: String): Boolean = { |
| val sortColumns = alterTableModel.tableProperties.get("sort_columns") |
| if(sortColumns.isDefined) { |
| sortColumns.get.contains(columnName) |
| } else { |
| true |
| } |
| } |
| |
| private def isVarcharColumn(columnName: String): Boolean = { |
| val varcharColumns = alterTableModel.tableProperties.get("long_string_columns") |
| if (varcharColumns.isDefined) { |
| varcharColumns.get.contains(columnName) |
| } else { |
| false |
| } |
| } |
| |
| def process: Seq[ColumnSchema] = { |
| val tableSchema = tableInfo.getFactTable |
| val tableCols = tableSchema.getListOfColumns.asScala |
| val existingColsSize = tableCols.size |
| var longStringCols = Seq[ColumnSchema]() |
| // get all original dimension columns |
| // but exclude complex type columns and long string columns |
| var allColumns = tableCols.filter(x => |
| (x.isDimensionColumn && !x.getDataType.isComplexType() |
| && x.getSchemaOrdinal != -1 && (x.getDataType != DataTypes.VARCHAR))) |
| var newCols = Seq[ColumnSchema]() |
| var invertedIndxCols: Array[String] = Array[String]() |
| if (alterTableModel.tableProperties.get(CarbonCommonConstants.INVERTED_INDEX).isDefined) { |
| invertedIndxCols = alterTableModel.tableProperties(CarbonCommonConstants.INVERTED_INDEX) |
| .split(',').map(_.trim) |
| } |
| |
| // add new dimension columns |
| alterTableModel.dimCols.foreach(field => { |
| val encoders = new java.util.ArrayList[Encoding]() |
| encoders.add(Encoding.DICTIONARY) |
| val columnSchema: ColumnSchema = TableNewProcessor.createColumnSchema( |
| field, |
| encoders, |
| isDimensionCol = true, |
| field.precision, |
| field.scale, |
| field.schemaOrdinal + existingColsSize, |
| alterTableModel.highCardinalityDims, |
| alterTableModel.databaseName.getOrElse(dbName), |
| isSortColumn(field.name.getOrElse(field.column)), |
| isVarcharColumn(field.name.getOrElse(field.column))) |
| if (columnSchema.getDataType == DataTypes.VARCHAR) { |
| // put the new long string columns in 'longStringCols' |
| // and add them after old long string columns |
| longStringCols ++= Seq(columnSchema) |
| } else { |
| allColumns ++= Seq(columnSchema) |
| } |
| newCols ++= Seq(columnSchema) |
| }) |
| // put the old long string columns |
| allColumns ++= tableCols.filter(x => |
| (x.isDimensionColumn && (x.getDataType == DataTypes.VARCHAR))) |
| // put the new long string columns |
| allColumns ++= longStringCols |
| // put complex type columns at the end of dimension columns |
| allColumns ++= tableCols.filter(x => |
| (x.isDimensionColumn && (x.getDataType.isComplexType() || x.getSchemaOrdinal == -1))) |
| // original measure columns |
| allColumns ++= tableCols.filter(x => !x.isDimensionColumn) |
| // add new measure columns |
| alterTableModel.msrCols.foreach(field => { |
| val encoders = new java.util.ArrayList[Encoding]() |
| val columnSchema: ColumnSchema = TableNewProcessor.createColumnSchema( |
| field, |
| encoders, |
| isDimensionCol = false, |
| field.precision, |
| field.scale, |
| field.schemaOrdinal + existingColsSize, |
| alterTableModel.highCardinalityDims, |
| alterTableModel.databaseName.getOrElse(dbName) |
| ) |
| allColumns ++= Seq(columnSchema) |
| newCols ++= Seq(columnSchema) |
| }) |
| |
| if (invertedIndxCols.nonEmpty) { |
| for (column <- newCols) { |
| if (invertedIndxCols.contains(column.getColumnName) && column.isDimensionColumn) { |
| column.setUseInvertedIndex(true) |
| } |
| } |
| } |
| |
| // Check if there is any duplicate measures or dimensions. |
| // Its based on the dimension name and measure name |
| allColumns.filter(x => !x.isInvisible).groupBy(_.getColumnName) |
| .foreach(f => if (f._2.size > 1) { |
| val name = f._1 |
| sys.error(s"Duplicate column found with name: $name") |
| }) |
| |
| if (newCols.exists(_.getDataType.isComplexType)) { |
| sys.error(s"Complex column cannot be added") |
| } |
| |
| val columnValidator = CarbonSparkFactory.getCarbonColumnValidator |
| columnValidator.validateColumns(allColumns) |
| |
| allColumns = CarbonScalaUtil.reArrangeColumnSchema(allColumns) |
| |
| if (tableInfo.getFactTable.getPartitionInfo != null) { |
| val par = tableInfo.getFactTable.getPartitionInfo.getColumnSchemaList |
| allColumns = allColumns.filterNot(b => par.contains(b)) ++= par.asScala |
| } |
| |
| def getLocalDictColumnList(tableProperties: scala.collection.mutable.Map[String, String], |
| columns: scala.collection.mutable.ListBuffer[ColumnSchema]): (scala.collection.mutable |
| .ListBuffer[ColumnSchema], scala.collection.mutable.ListBuffer[ColumnSchema]) = { |
| val includeColumns = new scala.collection.mutable.ListBuffer[ColumnSchema] |
| val excludeColumns = new scala.collection.mutable.ListBuffer[ColumnSchema] |
| val localDictIncludeColumns = if (tableProperties |
| .get(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE).isDefined) { |
| tableProperties(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE) |
| } else { |
| null |
| } |
| val localDictExcludeColumns = if (tableProperties |
| .get(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE).isDefined) { |
| tableProperties(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE) |
| } else { |
| null |
| } |
| if (null != localDictIncludeColumns) { |
| if (null == localDictExcludeColumns) { |
| columns.foreach { column => |
| if (localDictIncludeColumns.contains(column.getColumnName)) { |
| includeColumns.append(column) |
| } else { |
| if (column.getDataType.equals(DataTypes.STRING) || |
| column.getDataType.equals(DataTypes.VARCHAR) || |
| column.getDataType.toString.equals("ARRAY") || |
| column.getDataType.toString.equals("STRUCT")) { |
| excludeColumns.append(column) |
| } |
| } |
| } |
| } else { |
| columns.foreach { column => |
| if (localDictIncludeColumns.contains(column.getColumnName) && |
| !localDictExcludeColumns.contains(column.getColumnName)) { |
| includeColumns.append(column) |
| } else if (localDictExcludeColumns.contains(column.getColumnName)) { |
| excludeColumns.append(column) |
| } |
| } |
| } |
| } else { |
| if (null == localDictExcludeColumns) { |
| columns.foreach { column => |
| if (column.getDataType.equals(DataTypes.STRING) || |
| column.getDataType.equals(DataTypes.VARCHAR) || |
| column.getDataType.toString.equals("ARRAY") || |
| column.getDataType.toString.equals("STRUCT")) { |
| includeColumns.append(column) |
| } |
| } |
| } else { |
| columns.foreach { column => |
| if (!localDictExcludeColumns.contains(column.getColumnName) && |
| (column.getDataType.equals(DataTypes.STRING) || |
| column.getDataType.equals(DataTypes.VARCHAR) || |
| column.getDataType.toString.equals("ARRAY") || |
| column.getDataType.toString.equals("STRUCT"))) { |
| includeColumns.append(column) |
| } else if (localDictExcludeColumns.contains(column.getColumnName)) { |
| excludeColumns.append(column) |
| } |
| } |
| } |
| } |
| |
| (includeColumns, excludeColumns) |
| } |
| |
| val columnsWithoutNewCols = new scala.collection.mutable.ListBuffer[ColumnSchema] |
| allColumns.foreach { column => |
| if (!newCols.exists(x => x.getColumnName.equalsIgnoreCase(column.getColumnName))) { |
| columnsWithoutNewCols += column |
| } |
| } |
| |
| val isLocalDictEnabledForMainTable = tableSchema.getTableProperties |
| .get(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE) |
| |
| val alterMutableTblProperties: scala.collection.mutable.Map[String, String] = mutable |
| .Map(alterTableModel.tableProperties.toSeq: _*) |
| |
| // if local dictionary is enabled, then validate include and exclude columns if defined |
| if (null != isLocalDictEnabledForMainTable && isLocalDictEnabledForMainTable.toBoolean) { |
| var localDictIncludeColumns: Seq[String] = Seq[String]() |
| var localDictExcludeColumns: Seq[String] = Seq[String]() |
| // validate local dictionary include columns if defined |
| if (alterTableModel.tableProperties.get(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE) |
| .isDefined) { |
| localDictIncludeColumns = |
| alterTableModel.tableProperties(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE).split(",") |
| .map(_.trim) |
| CarbonScalaUtil |
| .validateLocalDictionaryColumns(alterMutableTblProperties, localDictIncludeColumns) |
| CarbonScalaUtil |
| .validateLocalConfiguredDictionaryColumns( |
| alterTableModel.dimCols ++ alterTableModel.msrCols, |
| alterMutableTblProperties, |
| localDictIncludeColumns) |
| } |
| |
| // validate local dictionary exclude columns if defined |
| if (alterTableModel.tableProperties.get(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE) |
| .isDefined) { |
| localDictExcludeColumns = |
| alterTableModel.tableProperties(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE).split(",") |
| .map(_.trim) |
| CarbonScalaUtil |
| .validateLocalDictionaryColumns(alterMutableTblProperties, localDictExcludeColumns) |
| CarbonScalaUtil |
| .validateLocalConfiguredDictionaryColumns( |
| alterTableModel.dimCols ++ alterTableModel.msrCols, |
| alterMutableTblProperties, |
| localDictExcludeColumns) |
| } |
| |
| // validate if both local dictionary include and exclude contains same column |
| CarbonScalaUtil.validateDuplicateLocalDictIncludeExcludeColmns(alterMutableTblProperties) |
| |
| CarbonUtil |
| .setLocalDictColumnsToWrapperSchema(newCols.asJava, |
| alterTableModel.tableProperties.asJava, |
| isLocalDictEnabledForMainTable) |
| } |
| |
| val includeExcludeColOfMainTable = getLocalDictColumnList(tableSchema.getTableProperties |
| .asScala, |
| columnsWithoutNewCols) |
| val includeExcludeColOfAlterTable = getLocalDictColumnList(alterMutableTblProperties, |
| newCols.to[mutable.ListBuffer]) |
| |
| // Append all Local Dictionary Include and Exclude columns of Alter Table to that of Main Table |
| includeExcludeColOfMainTable._1.appendAll(includeExcludeColOfAlterTable._1) |
| includeExcludeColOfMainTable._2.appendAll(includeExcludeColOfAlterTable._2) |
| |
| val localDictionaryIncludeColumn = new StringBuilder |
| val localDictionaryExcludeColumn = new StringBuilder |
| includeExcludeColOfMainTable._1.foreach { column => |
| localDictionaryIncludeColumn.append(column.getColumnName).append(",") |
| } |
| includeExcludeColOfMainTable._2.foreach { column => |
| localDictionaryExcludeColumn.append(column.getColumnName).append(",") |
| } |
| |
| // populate table properties map |
| val tablePropertiesMap = tableSchema.getTableProperties |
| alterTableModel.tableProperties.foreach { |
| case (key, mapValue) => |
| val value = tablePropertiesMap.get(key) |
| if (null != value) { |
| if (value != mapValue) { |
| tablePropertiesMap.put(key, value + "," + mapValue) |
| } |
| } else { |
| tablePropertiesMap.put(key, mapValue) |
| } |
| } |
| |
| // The Final Map should contain the combined Local Dictionary Include and |
| // Local Dictionary Exclude Columns from both Main table and Alter table |
| tablePropertiesMap |
| .put(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE, localDictionaryIncludeColumn.toString()) |
| tablePropertiesMap |
| .put(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE, localDictionaryExcludeColumn.toString()) |
| // This part will create dictionary file for all newly added dictionary columns |
| // if valid default value is provided, |
| // then that value will be included while creating dictionary file |
| val defaultValueString = "default.value." |
| newCols.foreach { col => |
| var rawData: String = null |
| for (elem <- alterTableModel.tableProperties) { |
| if (elem._1.toLowerCase.startsWith(defaultValueString)) { |
| if (col.getColumnName.equalsIgnoreCase(elem._1.substring(defaultValueString.length))) { |
| rawData = elem._2 |
| val data = DataTypeUtil.convertDataToBytesBasedOnDataType(elem._2, col) |
| if (null != data) { |
| col.setDefaultValue(data) |
| } else { |
| LOGGER |
| .error( |
| "Invalid default value for new column " + dbName + "." + |
| alterTableModel.tableName + |
| "." + col.getColumnName + " : " + elem._2) |
| } |
| } |
| } |
| else if (elem._1.equalsIgnoreCase("no_inverted_index") && |
| (elem._2.split(",").contains(col.getColumnName))) { |
| col.getEncodingList.remove(Encoding.INVERTED_INDEX) |
| } |
| } |
| } |
| tableSchema.setListOfColumns(allColumns.asJava) |
| tableInfo.setLastUpdatedTime(System.currentTimeMillis()) |
| tableInfo.setFactTable(tableSchema) |
| newCols |
| } |
| |
| } |
| |
| // TODO: move this to carbon store API |
| object TableNewProcessor { |
| def apply( |
| cm: TableModel): TableInfo = { |
| new TableNewProcessor(cm).process |
| } |
| |
| def createColumnSchema( |
| field: Field, |
| encoders: java.util.List[Encoding], |
| isDimensionCol: Boolean, |
| precision: Int, |
| scale: Int, |
| schemaOrdinal: Int, |
| highCardinalityDims: Seq[String], |
| databaseName: String, |
| isSortColumn: Boolean = false, |
| isVarcharColumn: Boolean = false): ColumnSchema = { |
| val dataType = DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")) |
| if (DataTypes.isDecimal(dataType)) { |
| dataType.asInstanceOf[DecimalType].setPrecision(field.precision) |
| dataType.asInstanceOf[DecimalType].setScale(field.scale) |
| } |
| val columnSchema = new ColumnSchema() |
| if (isVarcharColumn) { |
| columnSchema.setDataType(DataTypes.VARCHAR) |
| } else { |
| columnSchema.setDataType(dataType) |
| } |
| val colName = field.name.getOrElse(field.column) |
| columnSchema.setColumnName(colName) |
| if (highCardinalityDims.contains(colName)) { |
| encoders.remove(Encoding.DICTIONARY) |
| } |
| if (dataType == DataTypes.DATE) { |
| encoders.add(Encoding.DIRECT_DICTIONARY) |
| } |
| if (dataType == DataTypes.TIMESTAMP && !highCardinalityDims.contains(colName)) { |
| encoders.add(Encoding.DIRECT_DICTIONARY) |
| } |
| columnSchema.setEncodingList(encoders) |
| val colUniqueIdGenerator = ColumnUniqueIdGenerator.getInstance |
| val columnUniqueId = colUniqueIdGenerator.generateUniqueId(columnSchema) |
| columnSchema.setColumnUniqueId(columnUniqueId) |
| columnSchema.setColumnReferenceId(columnUniqueId) |
| columnSchema.setDimensionColumn(isDimensionCol) |
| columnSchema.setPrecision(precision) |
| columnSchema.setScale(scale) |
| columnSchema.setSchemaOrdinal(schemaOrdinal) |
| columnSchema.setSortColumn(isSortColumn) |
| columnSchema |
| } |
| } |
| |
| class TableNewProcessor(cm: TableModel) { |
| |
| def getAllChildren(fieldChildren: Option[List[Field]], |
| useDictionaryEncoding: Boolean): Seq[ColumnSchema] = { |
| var allColumns: Seq[ColumnSchema] = Seq[ColumnSchema]() |
| fieldChildren.foreach(fields => { |
| fields.foreach(field => { |
| val encoders = new java.util.ArrayList[Encoding]() |
| if (useDictionaryEncoding) { |
| encoders.add(Encoding.DICTIONARY) |
| } |
| val columnSchema: ColumnSchema = getColumnSchema( |
| DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), |
| field.name.getOrElse(field.column), |
| encoders, |
| true, |
| field, |
| cm.dataMapRelation, |
| useDictionaryEncoding = useDictionaryEncoding) |
| allColumns ++= Seq(columnSchema) |
| if (field.children.get != null) { |
| columnSchema.setNumberOfChild(field.children.get.size) |
| allColumns ++= getAllChildren(field.children, useDictionaryEncoding) |
| } |
| }) |
| }) |
| allColumns |
| } |
| |
| // varchar column is a string column that in long_string_columns |
| private def isVarcharColumn(colName : String): Boolean = { |
| cm.varcharCols.get.contains(colName) |
| } |
| |
| def getColumnSchema( |
| dataType: DataType, |
| colName: String, |
| encoders: java.util.List[Encoding], |
| isDimensionCol: Boolean, |
| field: Field, |
| map: Option[scala.collection.mutable.LinkedHashMap[Field, DataMapField]], |
| useDictionaryEncoding: Boolean = true) : ColumnSchema = { |
| val columnSchema = new ColumnSchema() |
| columnSchema.setDataType(dataType) |
| columnSchema.setColumnName(colName) |
| val isParentColumnRelation = map.isDefined && map.get.get(field).isDefined |
| if (!isParentColumnRelation) { |
| val highCardinalityDims = cm.highcardinalitydims.getOrElse(Seq()) |
| if (highCardinalityDims.contains(colName)) { |
| encoders.remove(Encoding.DICTIONARY) |
| } |
| if (dataType == DataTypes.DATE && useDictionaryEncoding) { |
| encoders.add(Encoding.DIRECT_DICTIONARY) |
| } |
| if (dataType == DataTypes.TIMESTAMP && |
| !highCardinalityDims.contains(colName) && useDictionaryEncoding) { |
| encoders.add(Encoding.DIRECT_DICTIONARY) |
| } |
| } |
| columnSchema.setEncodingList(encoders) |
| val colUniqueIdGenerator = ColumnUniqueIdGenerator.getInstance |
| val columnUniqueId = colUniqueIdGenerator.generateUniqueId(columnSchema) |
| columnSchema.setColumnUniqueId(columnUniqueId) |
| columnSchema.setColumnReferenceId(columnUniqueId) |
| columnSchema.setDimensionColumn(isDimensionCol) |
| columnSchema.setPrecision(field.precision) |
| columnSchema.setScale(field.scale) |
| columnSchema.setSchemaOrdinal(field.schemaOrdinal) |
| columnSchema.setSortColumn(false) |
| if (isVarcharColumn(colName)) { |
| columnSchema.setDataType(DataTypes.VARCHAR) |
| } |
| if (isParentColumnRelation) { |
| val dataMapField = map.get.get(field).get |
| columnSchema.setFunction(dataMapField.aggregateFunction) |
| val columnRelationList = dataMapField.columnTableRelationList.get |
| val parentColumnTableRelationList = new util.ArrayList[ParentColumnTableRelation] |
| columnRelationList.foreach { |
| columnRelation => |
| val relationIdentifier = new RelationIdentifier( |
| columnRelation.parentDatabaseName, |
| columnRelation.parentTableName, |
| columnRelation.parentTableId) |
| if (cm.parentTable.isDefined) { |
| relationIdentifier.setTablePath(cm.parentTable.get.getTablePath) |
| } |
| val parentColumnTableRelation = new ParentColumnTableRelation( |
| relationIdentifier, |
| columnRelation.parentColumnId, |
| columnRelation.parentColumnName) |
| parentColumnTableRelationList.add(parentColumnTableRelation) |
| } |
| columnSchema.setParentColumnTableRelations(parentColumnTableRelationList) |
| } |
| // TODO: Need to fill RowGroupID, converted type |
| // & Number of Children after DDL finalization |
| if (field.columnComment.nonEmpty) { |
| var columnProperties = columnSchema.getColumnProperties |
| if (columnProperties == null) { |
| columnProperties = new util.HashMap[String, String]() |
| columnSchema.setColumnProperties(columnProperties) |
| } |
| columnProperties.put(CarbonCommonConstants.COLUMN_COMMENT, field.columnComment) |
| } |
| columnSchema |
| } |
| |
| // process create dml fields and create wrapper TableInfo object |
| def process: TableInfo = { |
| val LOGGER = LogServiceFactory.getLogService(TableNewProcessor.getClass.getName) |
| var allColumns = Seq[ColumnSchema]() |
| var index = 0 |
| var measureCount = 0 |
| |
| // Sort columns should be at the begin of all columns |
| cm.sortKeyDims.get.foreach { keyDim => |
| val field = cm.dimCols.find(keyDim equals _.column).get |
| val encoders = if (getEncoderFromParent(field)) { |
| cm.parentTable.get.getColumnByName( |
| cm.dataMapRelation.get(field).columnTableRelationList.get.head.parentColumnName |
| ).getEncoder |
| } else { |
| val encoders = new java.util.ArrayList[Encoding]() |
| encoders.add(Encoding.DICTIONARY) |
| encoders |
| } |
| val columnSchema = getColumnSchema( |
| DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), |
| field.name.getOrElse(field.column), |
| encoders, |
| true, |
| field, |
| cm.dataMapRelation) |
| columnSchema.setSortColumn(true) |
| allColumns :+= columnSchema |
| index = index + 1 |
| } |
| |
| val dictionaryIncludeCols = cm.tableProperties |
| .getOrElse(CarbonCommonConstants.DICTIONARY_INCLUDE, "") |
| |
| def addDimensionCol(field: Field): Unit = { |
| val sortField = cm.sortKeyDims.get.find(field.column equals _) |
| if (sortField.isEmpty) { |
| val encoders = if (getEncoderFromParent(field)) { |
| cm.parentTable.get.getColumnByName(cm.dataMapRelation.get.get(field).get. |
| columnTableRelationList.get(0).parentColumnName).getEncoder |
| } else { |
| val encoders = new java.util.ArrayList[Encoding]() |
| encoders.add(Encoding.DICTIONARY) |
| encoders |
| } |
| val columnSchema = getColumnSchema( |
| DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), |
| field.name.getOrElse(field.column), |
| encoders, |
| true, |
| field, |
| cm.dataMapRelation) |
| allColumns :+= columnSchema |
| index = index + 1 |
| if (field.children.isDefined && field.children.get != null) { |
| val includeDictionaryEncoding = dictionaryIncludeCols.contains(field.column) |
| if (!includeDictionaryEncoding) { |
| columnSchema.getEncodingList.remove(Encoding.DICTIONARY) |
| } |
| columnSchema.setNumberOfChild(field.children.get.size) |
| allColumns ++= getAllChildren(field.children, includeDictionaryEncoding) |
| } |
| } |
| } |
| // add all dimensions |
| cm.dimCols.foreach(addDimensionCol(_)) |
| |
| // check whether the column is a local dictionary column and set in column schema |
| if (null != cm.tableProperties) { |
| CarbonUtil |
| .setLocalDictColumnsToWrapperSchema(allColumns.asJava, |
| cm.tableProperties.asJava, |
| cm.tableProperties(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)) |
| } |
| cm.msrCols.foreach { field => |
| // if aggregate function is defined in case of preaggregate and agg function is sum or avg |
| // then it can be stored as measure |
| var isAggFunPresent = false |
| // getting the encoder from maintable so whatever encoding is applied in maintable |
| // same encoder can be applied on aggregate table |
| val encoders = if (getEncoderFromParent(field)) { |
| isAggFunPresent = |
| cm.dataMapRelation.get(field).aggregateFunction.equalsIgnoreCase("sum") || |
| cm.dataMapRelation.get(field).aggregateFunction.equals("avg") || |
| cm.dataMapRelation.get(field).aggregateFunction.equals("count") |
| if (!isAggFunPresent) { |
| cm.parentTable.get.getColumnByName( |
| cm.dataMapRelation.get(field).columnTableRelationList.get.head.parentColumnName |
| ).getEncoder |
| } else { |
| new java.util.ArrayList[Encoding]() |
| } |
| } else { |
| new java.util.ArrayList[Encoding]() |
| } |
| // check if it can be dimension column |
| val isDimColumn = !encoders.isEmpty && !isAggFunPresent |
| val columnSchema = getColumnSchema( |
| DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), |
| field.name.getOrElse(field.column), |
| encoders, |
| isDimColumn, |
| field, |
| cm.dataMapRelation) |
| allColumns :+= columnSchema |
| index = index + 1 |
| if (!isDimColumn) { |
| measureCount += 1 |
| } |
| } |
| |
| // Check if there is any duplicate measures or dimensions. |
| // Its based on the dimension name and measure name |
| allColumns.groupBy(_.getColumnName).foreach { f => |
| if (f._2.size > 1) { |
| val name = f._1 |
| CarbonException.analysisException(s"Duplicate dimensions found with name: $name") |
| } |
| } |
| |
| val highCardinalityDims = cm.highcardinalitydims.getOrElse(Seq()) |
| |
| // Setting the boolean value of useInvertedIndex in column schema, if Parent table is defined |
| // Encoding is already decided above |
| if (!cm.parentTable.isDefined) { |
| val noInvertedIndexCols = cm.noInvertedIdxCols.getOrElse(Seq()) |
| val invertedIndexCols = cm.innvertedIdxCols.getOrElse(Seq()) |
| LOGGER.info("NoINVERTEDINDEX columns are : " + noInvertedIndexCols.mkString(",")) |
| for (column <- allColumns) { |
| // When the column is measure or the specified no inverted index column in DDL, |
| // set useInvertedIndex to false, otherwise true. |
| if (invertedIndexCols.contains(column.getColumnName) && |
| !cm.msrCols.exists(_.column.equalsIgnoreCase(column.getColumnName))) { |
| column.setUseInvertedIndex(true) |
| } else { |
| column.setUseInvertedIndex(false) |
| } |
| } |
| } |
| // Adding dummy measure if no measure is provided |
| if (measureCount == 0) { |
| val encoders = new java.util.ArrayList[Encoding]() |
| val field = Field( |
| CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE, |
| Some(DataTypes.DOUBLE.getName), |
| Some(CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE), |
| None |
| ) |
| val columnSchema: ColumnSchema = getColumnSchema( |
| DataTypes.DOUBLE, |
| CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE, |
| encoders, |
| false, |
| field, |
| cm.dataMapRelation) |
| columnSchema.setInvisible(true) |
| allColumns :+= columnSchema |
| } |
| val columnValidator = CarbonSparkFactory.getCarbonColumnValidator |
| columnValidator.validateColumns(allColumns) |
| |
| val tableInfo = new TableInfo() |
| val tableSchema = new TableSchema() |
| |
| val schemaEvol = new SchemaEvolution() |
| schemaEvol.setSchemaEvolutionEntryList(new util.ArrayList[SchemaEvolutionEntry]()) |
| tableSchema.setTableId(UUID.randomUUID().toString) |
| // populate table properties map |
| val tablePropertiesMap = new java.util.HashMap[String, String]() |
| cm.tableProperties.foreach { |
| x => tablePropertiesMap.put(x._1, x._2) |
| } |
| // Add table comment to table properties |
| tablePropertiesMap.put("comment", cm.tableComment.getOrElse("")) |
| val badRecordsPath = getBadRecordsPath(tablePropertiesMap, |
| cm.tableName, |
| tableSchema.getTableId, |
| cm.databaseNameOp.getOrElse("default")) |
| tablePropertiesMap.put("bad_record_path", badRecordsPath) |
| tableSchema.setTableProperties(tablePropertiesMap) |
| if (cm.bucketFields.isDefined) { |
| val bucketCols = cm.bucketFields.get.bucketColumns.map { b => |
| val col = allColumns.find(_.getColumnName.equalsIgnoreCase(b)) |
| col match { |
| case Some(colSchema: ColumnSchema) => |
| if (colSchema.isDimensionColumn && !colSchema.getDataType.isComplexType) { |
| colSchema |
| } else { |
| LOGGER.error(s"Bucket field must be dimension column and " + |
| s"should not be measure or complex column: ${ colSchema.getColumnName }") |
| CarbonException.analysisException(s"Bucket field must be dimension column and " + |
| s"should not be measure or complex column: ${ |
| colSchema |
| .getColumnName |
| }") |
| } |
| case _ => |
| LOGGER.error(s"Bucket field is not present in table columns") |
| CarbonException.analysisException(s"Bucket field is not present in table columns") |
| } |
| } |
| tableSchema.setBucketingInfo( |
| new BucketingInfo(bucketCols.asJava, cm.bucketFields.get.numberOfBuckets)) |
| } |
| if (cm.partitionInfo.isDefined) { |
| val partitionInfo = cm.partitionInfo.get |
| val partitionColumnSchema = partitionInfo.getColumnSchemaList.asScala |
| val partitionCols = allColumns.filter { column => |
| partitionColumnSchema.exists(_.getColumnName.equalsIgnoreCase(column.getColumnName)) |
| } |
| val orderCols = |
| partitionColumnSchema.map( |
| f => partitionCols.find(_.getColumnName.equalsIgnoreCase(f.getColumnName)).get).asJava |
| partitionInfo.setColumnSchemaList(orderCols) |
| tableSchema.setPartitionInfo(partitionInfo) |
| } |
| tableSchema.setTableName(cm.tableName) |
| tableSchema.setListOfColumns(allColumns.asJava) |
| tableSchema.setSchemaEvolution(schemaEvol) |
| tableInfo.setDatabaseName(cm.databaseNameOp.getOrElse(null)) |
| tableInfo.setTableUniqueName(CarbonTable.buildUniqueName(cm.databaseNameOp.getOrElse(null), |
| cm.tableName)) |
| tableInfo.setLastUpdatedTime(System.currentTimeMillis()) |
| tableInfo.setFactTable(tableSchema) |
| tableInfo |
| } |
| |
| private def getBadRecordsPath(tablePropertiesMap: util.HashMap[String, String], |
| tableName: String, |
| tableId: String, |
| databaseName: String): String = { |
| val badRecordsPath = tablePropertiesMap.asScala |
| .getOrElse("bad_record_path", CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL) |
| if (badRecordsPath == null || badRecordsPath.isEmpty) { |
| CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL |
| } else { |
| badRecordsPath + CarbonCommonConstants.FILE_SEPARATOR + databaseName + |
| CarbonCommonConstants.FILE_SEPARATOR + s"${tableName}_$tableId" |
| } |
| } |
| |
| /** |
| * Method to check to get the encoder from parent or not |
| * |
| * @param field column field |
| * @return get encoder from parent |
| */ |
| private def getEncoderFromParent(field: Field): Boolean = { |
| cm.parentTable.isDefined && |
| cm.dataMapRelation.get.get(field).isDefined && |
| cm.dataMapRelation.get.get(field).get.columnTableRelationList.size == 1 |
| } |
| |
| // For checking if the specified col group columns are specified in fields list. |
| protected def checkColGroupsValidity(colGrps: Seq[String], |
| allCols: Seq[ColumnSchema], |
| highCardCols: Seq[String]): Unit = { |
| if (null != colGrps) { |
| colGrps.foreach(columngroup => { |
| val rowCols = columngroup.split(",") |
| rowCols.foreach(colForGrouping => { |
| var found: Boolean = false |
| // check for dimensions + measures |
| allCols.foreach(eachCol => { |
| if (eachCol.getColumnName.equalsIgnoreCase(colForGrouping.trim())) { |
| found = true |
| } |
| }) |
| // check for No Dicitonary dimensions |
| highCardCols.foreach(noDicCol => { |
| if (colForGrouping.trim.equalsIgnoreCase(noDicCol)) { |
| found = true |
| } |
| }) |
| |
| if (!found) { |
| CarbonException.analysisException( |
| s"column $colForGrouping is not present in Field list") |
| } |
| }) |
| }) |
| } |
| } |
| } |
