6674

Build # 6674

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ef6eab4c385bb5b9df71263a313e8dfd682e2a57">ef6eab4c3<a href="https://github.com/apache/carbondata/commit/ef6eab4c385bb5b9df71263a313e8dfd682e2a57">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/f892e26c84b582cfc7060ab98b976b72df4a1450">f892e26c8</a>">f892e26c8</a><a href="https://github.com/apache/carbondata/commit/ef6eab4c385bb5b9df71263a313e8dfd682e2a57"> into 9ee0f3579">9ee0f3579</a>

Run Details

8 of 8 new or added lines in 2 files covered. (100.0%)

57199 of 70657 relevant lines covered (80.95%)

1.08 hits per line

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

81.05

/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParserUtil.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.parser

import scala.collection.JavaConverters._
import scala.collection.mutable

import org.antlr.v4.runtime.tree.TerminalNode
import org.apache.spark.sql.{CarbonEnv, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.parser.ParserUtils.operationNotAllowed
import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.command.{PartitionerField, TableModel, TableNewProcessor}
import org.apache.spark.sql.execution.command.table.{CarbonCreateTableAsSelectCommand, CarbonCreateTableCommand}
import org.apache.spark.sql.types.StructField

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.datatype.DataTypes
import org.apache.carbondata.core.metadata.schema.SchemaReader
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.spark.CarbonOption
import org.apache.carbondata.spark.util.{CarbonScalaUtil, CommonUtil}

/**
 * Utility class to validate the create table and CTAS command,
 * and to prepare the logical plan of create table and CTAS command.
 */
object CarbonSparkSqlParserUtil {
  /**
   * The method validate that the property configured for the streaming attribute is valid.
   *
   * @param carbonOption Instance of CarbonOption having all the required option for datasource.
   */
  private def validateStreamingProperty(carbonOption: CarbonOption): Unit = {
    try {
      carbonOption.isStreaming
    } catch {
      case _: IllegalArgumentException =>
        throw new MalformedCarbonCommandException(
          "Table property 'streaming' should be either 'true' or 'false'")
    }
  }

  /**
   * The method validates the create table command and returns the create table or
   * ctas table LogicalPlan.
   *
   * @param createTableTuple a tuple of (CreateTableHeaderContext, SkewSpecContext,
   *                         BucketSpecContext, ColTypeListContext, ColTypeListContext,
   *                         TablePropertyListContext,
   *                         LocationSpecContext, Option[String], TerminalNode, QueryContext,
   *                         String)
   * @param extraTableTuple  A tupple of (Seq[StructField], Boolean, TableIdentifier, Boolean,
   *                         Seq[String],
   *                         Option[String], mutable.Map[String, String], Map[String, String],
   *                         Seq[StructField],
   *                         Seq[PartitionerField], CarbonSpark2SqlParser, SparkSession,
   *                         Option[LogicalPlan])
   * @return <LogicalPlan> of create table or ctas table
   *
   */
  def createCarbonTable(createTableTuple: (CreateTableHeaderContext, SkewSpecContext,
    BucketSpecContext, ColTypeListContext, ColTypeListContext, TablePropertyListContext,
    LocationSpecContext, Option[String], TerminalNode, QueryContext, String),
      extraTableTuple: (Seq[StructField], Boolean, TableIdentifier, Boolean, Seq[String],
        Option[String], mutable.Map[String, String], Map[String, String], Seq[StructField],
        Seq[PartitionerField], CarbonSpark2SqlParser, SparkSession,
        Option[LogicalPlan])): LogicalPlan = {
    val (tableHeader, skewSpecContext, bucketSpecContext, partitionColumns, columns,
    tablePropertyList, locationSpecContext, tableComment, ctas, query, provider) = createTableTuple
    val (cols, external, tableIdentifier, ifNotExists, colNames, tablePath,
    tableProperties, properties, partitionByStructFields, partitionFields,
    parser, sparkSession, selectQuery) = extraTableTuple
    val options = new CarbonOption(properties)
    // validate streaming property
    validateStreamingProperty(options)
    var fields = parser.getFields(cols ++ partitionByStructFields)
    // validate for create table as select
    selectQuery match {
      case Some(q) =>
        // create table as select does not allow creation of partitioned table
        if (partitionFields.nonEmpty) {
          val errorMessage = "A Create Table As Select (CTAS) statement is not allowed to " +
                             "create a partitioned table using Carbondata file formats."
          operationNotAllowed(errorMessage, partitionColumns)
        }
        // create table as select does not allow to explicitly specify schema
        if (fields.nonEmpty) {
          operationNotAllowed(
            "Schema may not be specified in a Create Table As Select (CTAS) statement", columns)
        }
        // external table is not allow
        if (external) {
          operationNotAllowed("Create external table as select", tableHeader)
        }
        fields = parser
          .getFields(CarbonEnv.getInstance(sparkSession).carbonMetastore
            .getSchemaFromUnresolvedRelation(sparkSession, Some(q).get))
      case _ =>
      // ignore this case
    }
    if (partitionFields.nonEmpty && options.isStreaming) {
      operationNotAllowed("Streaming is not allowed on partitioned table", partitionColumns)
    }
    // validate tblProperties
    val bucketFields = parser.getBucketFields(tableProperties, fields, options)
    var isTransactionalTable: Boolean = true

    val tableInfo = if (external) {
      if (fields.nonEmpty) {
        // user provided schema for this external table, this is not allow currently
        // see CARBONDATA-2866
        operationNotAllowed(
          "Schema may not be specified for external table", columns)
      }
      if (partitionByStructFields.nonEmpty) {
        operationNotAllowed(
          "Partition is not supported for external table", partitionColumns)
      }
      // read table info from schema file in the provided table path
      // external table also must convert table name to lower case
      val identifier = AbsoluteTableIdentifier.from(
        tablePath.get,
        CarbonEnv.getDatabaseName(tableIdentifier.database)(sparkSession).toLowerCase(),
        tableIdentifier.table.toLowerCase())
      val table = try {
        val schemaPath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath)
        if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {
          if (provider.equalsIgnoreCase("'carbonfile'")) {
            SchemaReader.inferSchema(identifier, true)
          } else {
            isTransactionalTable = false
            SchemaReader.inferSchema(identifier, false)
          }
        } else {
          SchemaReader.getTableInfo(identifier)
        }
      } catch {
        case e: Throwable =>
          operationNotAllowed(s"Invalid table path provided: ${ tablePath.get } ", tableHeader)
      }

      // set "_external" property, so that DROP TABLE will not delete the data
      if (provider.equalsIgnoreCase("'carbonfile'")) {
        table.getFactTable.getTableProperties.put("_filelevelformat", "true")
        table.getFactTable.getTableProperties.put("_external", "false")
      } else {
        table.getFactTable.getTableProperties.put("_external", "true")
        table.getFactTable.getTableProperties.put("_filelevelformat", "false")
      }
      var isLocalDic_enabled = table.getFactTable.getTableProperties
        .get(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)
      if (null == isLocalDic_enabled) {
        table.getFactTable.getTableProperties
          .put(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE,
            CarbonProperties.getInstance()
              .getProperty(CarbonCommonConstants.LOCAL_DICTIONARY_SYSTEM_ENABLE,
                CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE_DEFAULT))
      }
      isLocalDic_enabled = table.getFactTable.getTableProperties
        .get(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)
      if (CarbonScalaUtil.validateLocalDictionaryEnable(isLocalDic_enabled) &&
          isLocalDic_enabled.toBoolean) {
        val allcolumns = table.getFactTable.getListOfColumns
        for (i <- 0 until allcolumns.size()) {
          val cols = allcolumns.get(i)
          if (cols.getDataType == DataTypes.STRING || cols.getDataType == DataTypes.VARCHAR) {
            cols.setLocalDictColumn(true)
          }
          allcolumns.set(i, cols)
        }
        table.getFactTable.setListOfColumns(allcolumns)
      }
      table
    } else {
      // prepare table model of the collected tokens
      val tableModel: TableModel = parser.prepareTableModel(
        ifNotExists,
        convertDbNameToLowerCase(tableIdentifier.database),
        tableIdentifier.table.toLowerCase,
        fields,
        partitionFields,
        tableProperties,
        bucketFields,
        isAlterFlow = false,
        false,
        tableComment)
      TableNewProcessor(tableModel)
    }
    tableInfo.setTransactionalTable(isTransactionalTable)
    selectQuery match {
      case query@Some(q) =>
        CarbonCreateTableAsSelectCommand(
          tableInfo = tableInfo,
          query = query.get,
          ifNotExistsSet = ifNotExists,
          tableLocation = tablePath)
      case _ =>
        CarbonCreateTableCommand(
          tableInfo = tableInfo,
          ifNotExistsSet = ifNotExists,
          tableLocation = tablePath,
          external)
    }
  }

  /**
   * This method will convert the database name to lower case
   *
   * @param dbName database name.
   * @return Option of String
   */
  def convertDbNameToLowerCase(dbName: Option[String]): Option[String] = {
    dbName match {
      case Some(databaseName) => Some(databaseName.toLowerCase)
      case None => dbName
    }
  }

  /**
   * Validates the partition columns and return's A tuple of partition columns and partitioner
   * fields.
   *
   * @param partitionColumns        An instance of ColTypeListContext having parser rules for
   *                                column.
   * @param colNames                <Seq[String]> Sequence of Table column names.
   * @param tableProperties         <Map[String, String]> Table property map.
   * @param partitionByStructFields Seq[StructField] Sequence of partition fields.
   * @return <Seq[PartitionerField]> A Seq of partitioner fields.
   */
  def validatePartitionFields(
      partitionColumns: ColTypeListContext,
      colNames: Seq[String],
      tableProperties: mutable.Map[String, String],
      partitionByStructFields: Seq[StructField]): Seq[PartitionerField] = {

    val partitionerFields = partitionByStructFields.map { structField =>
      PartitionerField(structField.name, Some(structField.dataType.toString), null)
    }
    // validate partition clause
    if (partitionerFields.nonEmpty) {
      if (!CommonUtil.validatePartitionColumns(tableProperties, partitionerFields)) {
        throw new MalformedCarbonCommandException("Error: Invalid partition definition")
      }
      // partition columns should not be part of the schema
      val badPartCols = partitionerFields.map(_.partitionColumn.toLowerCase).toSet
        .intersect(colNames.map(_.toLowerCase).toSet)
      if (badPartCols.nonEmpty) {
        operationNotAllowed(s"Partition columns should not be specified in the schema: " +
                            badPartCols.map("\"" + _ + "\"").mkString("[", ",", "]")
          , partitionColumns: ColTypeListContext)
      }
    }
    partitionerFields
  }

  /**
   * Parse a key-value map from a [[TablePropertyListContext]], assuming all values are specified.
   *
   * @param ctx   Instance of TablePropertyListContext defining parser rule for the table
   *              properties.
   * @param props <Map[String, String]> Map of table property list
   * @return <Map[String, String]> Map of transformed table property.
   */
  def visitPropertyKeyValues(ctx: TablePropertyListContext,
      props: Map[String, String]): Map[String, String] = {
    val badKeys = props.filter { case (_, v) => v == null }.keys
    if (badKeys.nonEmpty) {
      operationNotAllowed(
        s"Values must be specified for key(s): ${ badKeys.mkString("[", ",", "]") }", ctx)
    }
    props.map { case (key, value) =>
      if (needToConvertToLowerCase(key)) {
        (key.toLowerCase, value.toLowerCase)
      } else {
        (key.toLowerCase, value)
      }
    }
  }

  /**
   * check's whether need to convert to lower case
   *
   * @param key <String> property key
   * @return returns <true> if lower case conversion is needed else <false>
   */
  def needToConvertToLowerCase(key: String): Boolean = {
    val noConvertList = Array("LIST_INFO", "RANGE_INFO", "PATH")
    !noConvertList.exists(x => x.equalsIgnoreCase(key))
  }

  /**
   * The method validate the create table command and returns the table's columns.
   *
   * @param tableHeader       An instance of CreateTableHeaderContext having parser rules for
   *                          create table.
   * @param skewSpecContext   An instance of SkewSpecContext having parser rules for create table.
   * @param bucketSpecContext An instance of BucketSpecContext having parser rules for create table.
   * @param columns           An instance of ColTypeListContext having parser rules for columns
   *                          of the table.
   * @param cols              Table;s columns.
   * @param tableIdentifier   Instance of table identifier.
   * @param isTempTable       Flag to identify temp table.
   * @return Table's column names <Seq[String]>.
   */
  def validateCreateTableReqAndGetColumns(tableHeader: CreateTableHeaderContext,
      skewSpecContext: SkewSpecContext,
      bucketSpecContext: BucketSpecContext,
      columns: ColTypeListContext,
      cols: Seq[StructField],
      tableIdentifier: TableIdentifier,
      isTempTable: Boolean): Seq[String] = {
    // TODO: implement temporary tables
    if (isTempTable) {
      throw new ParseException(
        "CREATE TEMPORARY TABLE is not supported yet. " +
        "Please use CREATE TEMPORARY VIEW as an alternative.", tableHeader)
    }
    if (skewSpecContext != null) {
      operationNotAllowed("CREATE TABLE ... SKEWED BY", skewSpecContext)
    }
    if (bucketSpecContext != null) {
      operationNotAllowed("CREATE TABLE ... CLUSTERED BY", bucketSpecContext)
    }

    // Ensuring whether no duplicate name is used in table definition
    val colNames: Seq[String] = cols.map(_.name)
    if (colNames.length != colNames.distinct.length) {
      val duplicateColumns = colNames.groupBy(identity).collect {
        case (x, ys) if ys.length > 1 => "\"" + x + "\""
      }
      operationNotAllowed(s"Duplicated column names found in table definition of " +
                          s"$tableIdentifier: ${ duplicateColumns.mkString("[", ",", "]") }",
        columns)
    }
    colNames
  }
  /**
   * The method return's the storage type
   * @param createFileFormat
   * @return
   */
  def getFileStorage(createFileFormat: CreateFileFormatContext): String = {
    Option(createFileFormat) match {
      case Some(value) =>
        val result = value.children.get(1).getText
        if (result.equalsIgnoreCase("by")) {
          value.storageHandler().STRING().getSymbol.getText
        } else if (result.equalsIgnoreCase("as") && value.children.size() > 1) {
          value.children.get(2).getText
        } else {
          // The case of "STORED AS PARQUET/ORC"
          ""
        }
      case _ => ""
    }
  }
}

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.parser
19
20	import scala.collection.JavaConverters._
21	import scala.collection.mutable
22
23	import org.antlr.v4.runtime.tree.TerminalNode
24	import org.apache.spark.sql.{CarbonEnv, SparkSession}
25	import org.apache.spark.sql.catalyst.TableIdentifier
26	import org.apache.spark.sql.catalyst.parser.ParseException
27	import org.apache.spark.sql.catalyst.parser.ParserUtils.operationNotAllowed
28	import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
29	import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
30	import org.apache.spark.sql.execution.command.{PartitionerField, TableModel, TableNewProcessor}
31	import org.apache.spark.sql.execution.command.table.{CarbonCreateTableAsSelectCommand, CarbonCreateTableCommand}
32	import org.apache.spark.sql.types.StructField
33
34	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
35	import org.apache.carbondata.core.constants.CarbonCommonConstants
36	import org.apache.carbondata.core.datastore.impl.FileFactory
37	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
38	import org.apache.carbondata.core.metadata.datatype.DataTypes
39	import org.apache.carbondata.core.metadata.schema.SchemaReader
40	import org.apache.carbondata.core.util.CarbonProperties
41	import org.apache.carbondata.core.util.path.CarbonTablePath
42	import org.apache.carbondata.spark.CarbonOption
43	import org.apache.carbondata.spark.util.{CarbonScalaUtil, CommonUtil}
44
45	/**
46	* Utility class to validate the create table and CTAS command,
47	* and to prepare the logical plan of create table and CTAS command.
48	*/
49	object CarbonSparkSqlParserUtil {
50	/**
51	* The method validate that the property configured for the streaming attribute is valid.
52	*
53	* @param carbonOption Instance of CarbonOption having all the required option for datasource.
54	*/
55	private def validateStreamingProperty(carbonOption: CarbonOption): Unit = {
56	try {	2✔
57	carbonOption.isStreaming	2✔
58	} catch {
59	case _: IllegalArgumentException =>
60	throw new MalformedCarbonCommandException(	2✔
61	"Table property 'streaming' should be either 'true' or 'false'")	2✔
62	}
63	}
64
65	/**
66	* The method validates the create table command and returns the create table or
67	* ctas table LogicalPlan.
68	*
69	* @param createTableTuple a tuple of (CreateTableHeaderContext, SkewSpecContext,
70	* BucketSpecContext, ColTypeListContext, ColTypeListContext,
71	* TablePropertyListContext,
72	* LocationSpecContext, Option[String], TerminalNode, QueryContext,
73	* String)
74	* @param extraTableTuple A tupple of (Seq[StructField], Boolean, TableIdentifier, Boolean,
75	* Seq[String],
76	* Option[String], mutable.Map[String, String], Map[String, String],
77	* Seq[StructField],
78	* Seq[PartitionerField], CarbonSpark2SqlParser, SparkSession,
79	* Option[LogicalPlan])
80	* @return <LogicalPlan> of create table or ctas table
81	*
82	*/
83	def createCarbonTable(createTableTuple: (CreateTableHeaderContext, SkewSpecContext,
84	BucketSpecContext, ColTypeListContext, ColTypeListContext, TablePropertyListContext,
85	LocationSpecContext, Option[String], TerminalNode, QueryContext, String),
86	extraTableTuple: (Seq[StructField], Boolean, TableIdentifier, Boolean, Seq[String],
87	Option[String], mutable.Map[String, String], Map[String, String], Seq[StructField],
88	Seq[PartitionerField], CarbonSpark2SqlParser, SparkSession,
89	Option[LogicalPlan])): LogicalPlan = {
90	val (tableHeader, skewSpecContext, bucketSpecContext, partitionColumns, columns,	2✔
91	tablePropertyList, locationSpecContext, tableComment, ctas, query, provider) = createTableTuple	2✔
92	val (cols, external, tableIdentifier, ifNotExists, colNames, tablePath,	2✔
93	tableProperties, properties, partitionByStructFields, partitionFields,	2✔
94	parser, sparkSession, selectQuery) = extraTableTuple	2✔
95	val options = new CarbonOption(properties)	2✔
96	// validate streaming property
97	validateStreamingProperty(options)	2✔
98	var fields = parser.getFields(cols ++ partitionByStructFields)	2✔
99	// validate for create table as select
100	selectQuery match {	2✔
101	case Some(q) =>	2✔
102	// create table as select does not allow creation of partitioned table
103	if (partitionFields.nonEmpty) {	1✔
104	val errorMessage = "A Create Table As Select (CTAS) statement is not allowed to " +	×
105	"create a partitioned table using Carbondata file formats."
106	operationNotAllowed(errorMessage, partitionColumns)	×
107	}
108	// create table as select does not allow to explicitly specify schema
109	if (fields.nonEmpty) {	1✔
110	operationNotAllowed(	×
111	"Schema may not be specified in a Create Table As Select (CTAS) statement", columns)	×
112	}
113	// external table is not allow
114	if (external) {	1✔
115	operationNotAllowed("Create external table as select", tableHeader)	×
116	}
117	fields = parser	1✔
118	.getFields(CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
119	.getSchemaFromUnresolvedRelation(sparkSession, Some(q).get))	1✔
120	case _ =>	2✔
121	// ignore this case
122	}
123	if (partitionFields.nonEmpty && options.isStreaming) {	2✔
124	operationNotAllowed("Streaming is not allowed on partitioned table", partitionColumns)	×
125	}
126	// validate tblProperties
127	val bucketFields = parser.getBucketFields(tableProperties, fields, options)	2✔
128	var isTransactionalTable: Boolean = true	2✔
129
130	val tableInfo = if (external) {	2✔
131	if (fields.nonEmpty) {	1✔
132	// user provided schema for this external table, this is not allow currently
133	// see CARBONDATA-2866
134	operationNotAllowed(	1✔
135	"Schema may not be specified for external table", columns)	1✔
136	}
137	if (partitionByStructFields.nonEmpty) {	1✔
138	operationNotAllowed(	×
139	"Partition is not supported for external table", partitionColumns)	×
140	}
141	// read table info from schema file in the provided table path
142	// external table also must convert table name to lower case
143	val identifier = AbsoluteTableIdentifier.from(	1✔
144	tablePath.get,	1✔
145	CarbonEnv.getDatabaseName(tableIdentifier.database)(sparkSession).toLowerCase(),	1✔
146	tableIdentifier.table.toLowerCase())	1✔
147	val table = try {	1✔
148	val schemaPath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath)	1✔
149	if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {	1✔
150	if (provider.equalsIgnoreCase("'carbonfile'")) {	1✔
151	SchemaReader.inferSchema(identifier, true)	1✔
152	} else {
153	isTransactionalTable = false	1✔
154	SchemaReader.inferSchema(identifier, false)	1✔
155	}
156	} else {
157	SchemaReader.getTableInfo(identifier)	1✔
158	}
159	} catch {
160	case e: Throwable =>	1✔
161	operationNotAllowed(s"Invalid table path provided: ${ tablePath.get } ", tableHeader)	×
162	}
163
164	// set "_external" property, so that DROP TABLE will not delete the data
165	if (provider.equalsIgnoreCase("'carbonfile'")) {	1✔
166	table.getFactTable.getTableProperties.put("_filelevelformat", "true")	1✔
167	table.getFactTable.getTableProperties.put("_external", "false")	1✔
168	} else {
169	table.getFactTable.getTableProperties.put("_external", "true")	1✔
170	table.getFactTable.getTableProperties.put("_filelevelformat", "false")	1✔
171	}
172	var isLocalDic_enabled = table.getFactTable.getTableProperties	1✔
173	.get(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)	1✔
174	if (null == isLocalDic_enabled) {	1✔
175	table.getFactTable.getTableProperties	1✔
176	.put(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE,	1✔
177	CarbonProperties.getInstance()	1✔
178	.getProperty(CarbonCommonConstants.LOCAL_DICTIONARY_SYSTEM_ENABLE,	1✔
179	CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE_DEFAULT))	1✔
180	}
181	isLocalDic_enabled = table.getFactTable.getTableProperties	1✔
182	.get(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)	1✔
183	if (CarbonScalaUtil.validateLocalDictionaryEnable(isLocalDic_enabled) &&	1✔
184	isLocalDic_enabled.toBoolean) {	1✔
185	val allcolumns = table.getFactTable.getListOfColumns	×
186	for (i <- 0 until allcolumns.size()) {	×
187	val cols = allcolumns.get(i)	×
188	if (cols.getDataType == DataTypes.STRING \|\| cols.getDataType == DataTypes.VARCHAR) {	×
189	cols.setLocalDictColumn(true)	×
190	}
191	allcolumns.set(i, cols)	×
192	}
193	table.getFactTable.setListOfColumns(allcolumns)	×
194	}
195	table	1✔
196	} else {
197	// prepare table model of the collected tokens
198	val tableModel: TableModel = parser.prepareTableModel(	2✔
199	ifNotExists,	2✔
200	convertDbNameToLowerCase(tableIdentifier.database),	2✔
201	tableIdentifier.table.toLowerCase,	2✔
202	fields,	2✔
203	partitionFields,	2✔
204	tableProperties,	2✔
205	bucketFields,	2✔
206	isAlterFlow = false,	2✔
207	false,	2✔
208	tableComment)	2✔
209	TableNewProcessor(tableModel)	2✔
210	}
211	tableInfo.setTransactionalTable(isTransactionalTable)	2✔
212	selectQuery match {	2✔
213	case query@Some(q) =>	2✔
214	CarbonCreateTableAsSelectCommand(	1✔
215	tableInfo = tableInfo,	1✔
216	query = query.get,	1✔
217	ifNotExistsSet = ifNotExists,	1✔
218	tableLocation = tablePath)	1✔
219	case _ =>
220	CarbonCreateTableCommand(	2✔
221	tableInfo = tableInfo,	2✔
222	ifNotExistsSet = ifNotExists,	2✔
223	tableLocation = tablePath,	2✔
224	external)	2✔
225	}
226	}
227
228	/**
229	* This method will convert the database name to lower case
230	*
231	* @param dbName database name.
232	* @return Option of String
233	*/
234	def convertDbNameToLowerCase(dbName: Option[String]): Option[String] = {
235	dbName match {	2✔
236	case Some(databaseName) => Some(databaseName.toLowerCase)	2✔
237	case None => dbName	2✔
238	}
239	}
240
241	/**
242	* Validates the partition columns and return's A tuple of partition columns and partitioner
243	* fields.
244	*
245	* @param partitionColumns An instance of ColTypeListContext having parser rules for
246	* column.
247	* @param colNames <Seq[String]> Sequence of Table column names.
248	* @param tableProperties <Map[String, String]> Table property map.
249	* @param partitionByStructFields Seq[StructField] Sequence of partition fields.
250	* @return <Seq[PartitionerField]> A Seq of partitioner fields.
251	*/
252	def validatePartitionFields(
253	partitionColumns: ColTypeListContext,
254	colNames: Seq[String],
255	tableProperties: mutable.Map[String, String],
256	partitionByStructFields: Seq[StructField]): Seq[PartitionerField] = {
257
258	val partitionerFields = partitionByStructFields.map { structField =>	2✔
259	PartitionerField(structField.name, Some(structField.dataType.toString), null)	2✔
260	}
261	// validate partition clause
262	if (partitionerFields.nonEmpty) {	2✔
263	if (!CommonUtil.validatePartitionColumns(tableProperties, partitionerFields)) {	2✔
264	throw new MalformedCarbonCommandException("Error: Invalid partition definition")	1✔
265	}
266	// partition columns should not be part of the schema
267	val badPartCols = partitionerFields.map(_.partitionColumn.toLowerCase).toSet	2✔
268	.intersect(colNames.map(_.toLowerCase).toSet)	2✔
269	if (badPartCols.nonEmpty) {	2✔
270	operationNotAllowed(s"Partition columns should not be specified in the schema: " +	1✔
271	badPartCols.map("\"" + _ + "\"").mkString("[", ",", "]")	1✔
272	, partitionColumns: ColTypeListContext)	1✔
273	}
274	}
275	partitionerFields	2✔
276	}
277
278	/**
279	* Parse a key-value map from a [[TablePropertyListContext]], assuming all values are specified.
280	*
281	* @param ctx Instance of TablePropertyListContext defining parser rule for the table
282	* properties.
283	* @param props <Map[String, String]> Map of table property list
284	* @return <Map[String, String]> Map of transformed table property.
285	*/
286	def visitPropertyKeyValues(ctx: TablePropertyListContext,
287	props: Map[String, String]): Map[String, String] = {
288	val badKeys = props.filter { case (_, v) => v == null }.keys	2✔
289	if (badKeys.nonEmpty) {	2✔
290	operationNotAllowed(	×
291	s"Values must be specified for key(s): ${ badKeys.mkString("[", ",", "]") }", ctx)	×
292	}
293	props.map { case (key, value) =>	2✔
294	if (needToConvertToLowerCase(key)) {	2✔
295	(key.toLowerCase, value.toLowerCase)	2✔
296	} else {
297	(key.toLowerCase, value)	2✔
298	}
299	}
300	}
301
302	/**
303	* check's whether need to convert to lower case
304	*
305	* @param key <String> property key
306	* @return returns <true> if lower case conversion is needed else <false>
307	*/
308	def needToConvertToLowerCase(key: String): Boolean = {	2✔
309	val noConvertList = Array("LIST_INFO", "RANGE_INFO", "PATH")	2✔
310	!noConvertList.exists(x => x.equalsIgnoreCase(key))	2✔
311	}
312
313	/**
314	* The method validate the create table command and returns the table's columns.
315	*
316	* @param tableHeader An instance of CreateTableHeaderContext having parser rules for
317	* create table.
318	* @param skewSpecContext An instance of SkewSpecContext having parser rules for create table.
319	* @param bucketSpecContext An instance of BucketSpecContext having parser rules for create table.
320	* @param columns An instance of ColTypeListContext having parser rules for columns
321	* of the table.
322	* @param cols Table;s columns.
323	* @param tableIdentifier Instance of table identifier.
324	* @param isTempTable Flag to identify temp table.
325	* @return Table's column names <Seq[String]>.
326	*/
327	def validateCreateTableReqAndGetColumns(tableHeader: CreateTableHeaderContext,
328	skewSpecContext: SkewSpecContext,
329	bucketSpecContext: BucketSpecContext,
330	columns: ColTypeListContext,
331	cols: Seq[StructField],
332	tableIdentifier: TableIdentifier,
333	isTempTable: Boolean): Seq[String] = {
334	// TODO: implement temporary tables
335	if (isTempTable) {	2✔
336	throw new ParseException(	×
337	"CREATE TEMPORARY TABLE is not supported yet. " +	×
338	"Please use CREATE TEMPORARY VIEW as an alternative.", tableHeader)	×
339	}
340	if (skewSpecContext != null) {	2✔
341	operationNotAllowed("CREATE TABLE ... SKEWED BY", skewSpecContext)	×
342	}
343	if (bucketSpecContext != null) {	2✔
344	operationNotAllowed("CREATE TABLE ... CLUSTERED BY", bucketSpecContext)	×
345	}
346
347	// Ensuring whether no duplicate name is used in table definition
348	val colNames: Seq[String] = cols.map(_.name)	2✔
349	if (colNames.length != colNames.distinct.length) {	2✔
350	val duplicateColumns = colNames.groupBy(identity).collect {	×
351	case (x, ys) if ys.length > 1 => "\"" + x + "\""	×
352	}
353	operationNotAllowed(s"Duplicated column names found in table definition of " +	×
354	s"$tableIdentifier: ${ duplicateColumns.mkString("[", ",", "]") }",	×
355	columns)	×
356	}
357	colNames	2✔
358	}
359	/**
360	* The method return's the storage type
361	* @param createFileFormat
362	* @return
363	*/
364	def getFileStorage(createFileFormat: CreateFileFormatContext): String = {
365	Option(createFileFormat) match {	2✔
366	case Some(value) =>	2✔
367	val result = value.children.get(1).getText	2✔
368	if (result.equalsIgnoreCase("by")) {	2✔
369	value.storageHandler().STRING().getSymbol.getText	2✔
370	} else if (result.equalsIgnoreCase("as") && value.children.size() > 1) {	2✔
371	value.children.get(2).getText	2✔
372	} else {
373	// The case of "STORED AS PARQUET/ORC"
374	""	×
375	}
376	case _ => ""	2✔
377	}
378	}
379	}	2✔

apache / carbondata / 6674

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