apache / carbondata / 135

Build # 135

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/292fcf810b5978473dfad8d018f929e41371057d">292fcf810<a href="https://github.com/apache/carbondata/commit/292fcf810b5978473dfad8d018f929e41371057d">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/878f54fd288b8aecc6a9751c0fb245e83a386efb">878f54fd2</a>">878f54fd2</a><a href="https://github.com/apache/carbondata/commit/292fcf810b5978473dfad8d018f929e41371057d"> into b588cb655">b588cb655</a>

Run Details

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

57669 of 71449 relevant lines covered (80.71%)

1.07 hits per line

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

78.42

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

import java.util.Locale

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.language.implicitConversions

import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.command.{TableModel, TableNewProcessor}
import org.apache.spark.sql.execution.strategy.CarbonLateDecodeStrategy
import org.apache.spark.sql.execution.streaming.Sink
import org.apache.spark.sql.hive.CarbonMetaStore
import org.apache.spark.sql.optimizer.CarbonLateDecodeRule
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
import org.apache.spark.sql.sources._
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CarbonException

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.schema.SchemaEvolutionEntry
import org.apache.carbondata.core.metadata.schema.table.TableInfo
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
import org.apache.carbondata.spark.CarbonOption
import org.apache.carbondata.spark.util.CarbonScalaUtil
import org.apache.carbondata.streaming.{CarbonStreamException, CarbonStreamingQueryListener, StreamSinkFactory}

/**
 * Carbon relation provider compliant to data source api.
 * Creates carbon relations
 */
class CarbonSource extends CreatableRelationProvider with RelationProvider
  with SchemaRelationProvider with StreamSinkProvider with DataSourceRegister {

  override def shortName(): String = "carbondata"

  // will be called if hive supported create table command is provided
  override def createRelation(sqlContext: SQLContext,
      parameters: Map[String, String]): BaseRelation = {
    CarbonEnv.getInstance(sqlContext.sparkSession)
    // if path is provided we can directly create Hadoop relation. \
    // Otherwise create datasource relation
    val newParameters = CarbonScalaUtil.getDeserializedParameters(parameters)
    newParameters.get("tablePath") match {
      case Some(path) => CarbonDatasourceHadoopRelation(sqlContext.sparkSession,
        Array(path),
        newParameters,
        None)
      case _ =>
        val options = new CarbonOption(newParameters)
        val tablePath =
          CarbonEnv.getTablePath(options.dbName, options.tableName)(sqlContext.sparkSession)
        CarbonDatasourceHadoopRelation(sqlContext.sparkSession,
          Array(tablePath),
          newParameters,
          None)
    }
  }

  // called by any write operation like INSERT INTO DDL or DataFrame.write API
  override def createRelation(
      sqlContext: SQLContext,
      mode: SaveMode,
      parameters: Map[String, String],
      data: DataFrame): BaseRelation = {
    CarbonEnv.getInstance(sqlContext.sparkSession)
    var newParameters = CarbonScalaUtil.getDeserializedParameters(parameters)
    val options = new CarbonOption(newParameters)
    val isExists = CarbonEnv.getInstance(sqlContext.sparkSession).carbonMetastore.tableExists(
      options.tableName, options.dbName)(sqlContext.sparkSession)
    val (doSave, doAppend) = (mode, isExists) match {
      case (SaveMode.ErrorIfExists, true) =>
        CarbonException.analysisException(s"table path already exists.")
      case (SaveMode.Overwrite, true) =>
        newParameters += (("overwrite", "true"))
        (true, false)
      case (SaveMode.Overwrite, false) | (SaveMode.ErrorIfExists, false) =>
        newParameters += (("overwrite", "true"))
        (true, false)
      case (SaveMode.Append, _) =>
        (false, true)
      case (SaveMode.Ignore, exists) =>
        (!exists, false)
    }

    if (doSave) {
      // save data when the save mode is Overwrite.
      new CarbonDataFrameWriter(sqlContext, data).saveAsCarbonFile(
        CaseInsensitiveMap[String](newParameters))
    } else if (doAppend) {
      new CarbonDataFrameWriter(sqlContext, data).appendToCarbonFile(
        CaseInsensitiveMap[String](newParameters))
    }

    createRelation(sqlContext, newParameters, data.schema)
  }

  // called by DDL operation with a USING clause
  override def createRelation(
      sqlContext: SQLContext,
      parameters: Map[String, String],
      dataSchema: StructType): BaseRelation = {
    CarbonEnv.getInstance(sqlContext.sparkSession)
    addLateDecodeOptimization(sqlContext.sparkSession)
    val newParameters =
      CaseInsensitiveMap[String](CarbonScalaUtil.getDeserializedParameters(parameters))
    val dbName: String =
      CarbonEnv.getDatabaseName(newParameters.get("dbName"))(sqlContext.sparkSession)
    val tableOption: Option[String] = newParameters.get("tableName")
    if (tableOption.isEmpty) {
      CarbonException.analysisException("Table creation failed. Table name is not specified")
    }
    val tableName = tableOption.get.toLowerCase()
    if (tableName.contains(" ")) {
      CarbonException.analysisException(
        "Table creation failed. Table name cannot contain blank space")
    }
    val (path, updatedParams) = if (sqlContext.sparkSession.sessionState.catalog.listTables(dbName)
      .exists(_.table.equalsIgnoreCase(tableName))) {
        getPathForTable(sqlContext.sparkSession, dbName, tableName, newParameters)
    } else {
        createTableIfNotExists(sqlContext.sparkSession, newParameters, dataSchema)
    }

    CarbonDatasourceHadoopRelation(sqlContext.sparkSession, Array(path), updatedParams,
      Option(dataSchema))
  }

  private def addLateDecodeOptimization(ss: SparkSession): Unit = {
    if (ss.sessionState.experimentalMethods.extraStrategies.isEmpty) {
      ss.sessionState.experimentalMethods.extraStrategies = Seq(new CarbonLateDecodeStrategy)
      ss.sessionState.experimentalMethods.extraOptimizations = Seq(new CarbonLateDecodeRule)
    }
  }


  private def createTableIfNotExists(
      sparkSession: SparkSession,
      parameters: Map[String, String],
      dataSchema: StructType): (String, Map[String, String]) = {

    val dbName: String = CarbonEnv.getDatabaseName(parameters.get("dbName"))(sparkSession)
    val tableName: String = parameters.getOrElse("tableName", "").toLowerCase

    try {
      val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
      (carbonTable.getTablePath, parameters)
    } catch {
      case _: NoSuchTableException =>
        val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetastore
        val identifier = AbsoluteTableIdentifier.from(
          CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession),
          dbName,
          tableName)
        val updatedParams = CarbonSource.updateAndCreateTable(
          identifier, dataSchema, sparkSession, metaStore, parameters, None)
        (CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession), updatedParams)
      case ex: Exception =>
        throw new Exception("do not have dbname and tablename for carbon table", ex)
    }
  }

  /**
   * Returns the path of the table
   *
   * @param sparkSession
   * @param dbName
   * @param tableName
   * @return
   */
  private def getPathForTable(sparkSession: SparkSession, dbName: String,
      tableName : String, parameters: Map[String, String]): (String, Map[String, String]) = {

    if (StringUtils.isBlank(tableName)) {
      throw new MalformedCarbonCommandException("The Specified Table Name is Blank")
    }
    if (tableName.contains(" ")) {
      throw new MalformedCarbonCommandException("Table Name Should not have spaces ")
    }
    try {
      if (parameters.contains("tablePath")) {
        (parameters("tablePath"), parameters)
      } else if (!sparkSession.isInstanceOf[CarbonSession]) {
        (CarbonProperties.getStorePath + "/" + dbName + "/" + tableName, parameters)
      } else {
        (CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession), parameters)
      }
    } catch {
      case ex: Exception =>
        throw new Exception(s"Do not have $dbName and $tableName", ex)
    }
  }

  /**
   * produce a streaming `Sink` for a specific format
   * now it will create a default sink(CarbonAppendableStreamSink) for row format
   */
  override def createSink(sqlContext: SQLContext,
      parameters: Map[String, String],
      partitionColumns: Seq[String],
      outputMode: OutputMode): Sink = {

    // check "tablePath" option
    val options = new CarbonOption(parameters)
    val dbName = CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)
    val tableName = options.tableName
    if (tableName.contains(" ")) {
      throw new CarbonStreamException("Table creation failed. Table name cannot contain blank " +
                                      "space")
    }
    val sparkSession = sqlContext.sparkSession
    val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
    if (!carbonTable.isStreamingSink) {
      throw new CarbonStreamException(s"Table ${carbonTable.getDatabaseName}." +
                                      s"${carbonTable.getTableName} is not a streaming table")
    }

    // CarbonSession has added CarbonStreamingQueryListener during the initialization.
    // But other SparkSessions didn't, so here will add the listener once.
    if (!"CarbonSession".equals(sparkSession.getClass.getSimpleName)) {
      if (CarbonSource.listenerAdded.get(sparkSession.hashCode()).isEmpty) {
        synchronized {
          if (CarbonSource.listenerAdded.get(sparkSession.hashCode()).isEmpty) {
            sparkSession.streams.addListener(new CarbonStreamingQueryListener(sparkSession))
            CarbonSource.listenerAdded.put(sparkSession.hashCode(), true)
          }
        }
      }
    }

    // create sink
    StreamSinkFactory.createStreamTableSink(
      sqlContext.sparkSession,
      sqlContext.sparkSession.sessionState.newHadoopConf(),
      carbonTable,
      parameters)
  }

}

object CarbonSource {

  lazy val listenerAdded = new mutable.HashMap[Int, Boolean]()

  def createTableInfoFromParams(
      parameters: Map[String, String],
      dataSchema: StructType,
      identifier: AbsoluteTableIdentifier,
      query: Option[LogicalPlan],
      sparkSession: SparkSession): TableModel = {
    val sqlParser = new CarbonSpark2SqlParser
    val map = scala.collection.mutable.Map[String, String]()
    parameters.foreach { case (key, value) => map.put(key, value.toLowerCase()) }
    val options = new CarbonOption(parameters)
    val fields = query match {
      case Some(q) =>
        // if query is provided then it is a CTAS flow
        if (sqlParser.getFields(dataSchema).nonEmpty) {
          throw new AnalysisException(
            "Schema cannot be specified in a Create Table As Select (CTAS) statement")
        }
        sqlParser
          .getFields(CarbonEnv.getInstance(sparkSession).carbonMetastore
            .getSchemaFromUnresolvedRelation(sparkSession, q))
      case None =>
        sqlParser.getFields(dataSchema)
    }
    val bucketFields = sqlParser.getBucketFields(map, fields, options)
    sqlParser.prepareTableModel(ifNotExistPresent = false, Option(identifier.getDatabaseName),
      identifier.getTableName, fields, Nil, map, bucketFields)
  }

  /**
   * Update spark catalog table with schema information in case of schema storage is hive metastore
   * @param tableDesc
   * @param sparkSession
   * @return
   */
  def updateCatalogTableWithCarbonSchema(
      tableDesc: CatalogTable,
      sparkSession: SparkSession,
      query: Option[LogicalPlan] = None): CatalogTable = {
    val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetastore
    val storageFormat = tableDesc.storage
    val properties = storageFormat.properties
    if (!properties.contains("carbonSchemaPartsNo")) {
      val tablePath = CarbonEnv.getTablePath(
        tableDesc.identifier.database, tableDesc.identifier.table)(sparkSession)
      val dbName = CarbonEnv.getDatabaseName(tableDesc.identifier.database)(sparkSession)
      val identifier = AbsoluteTableIdentifier.from(tablePath, dbName, tableDesc.identifier.table)
      val map = updateAndCreateTable(
        identifier,
        tableDesc.schema,
        sparkSession,
        metaStore,
        properties,
        query)
      // updating params
      val updatedFormat = storageFormat.copy(properties = map)
      tableDesc.copy(storage = updatedFormat)
    } else {
      val tableInfo = CarbonUtil.convertGsonToTableInfo(properties.asJava)
      val isExternal = properties.getOrElse("isExternal", "false")
      val isTransactionalTable = properties.getOrElse("isTransactional", "true")
        .contains("true")
      tableInfo.setTransactionalTable(isTransactionalTable)
      if (isTransactionalTable && !metaStore.isReadFromHiveMetaStore) {
        // save to disk
        metaStore.saveToDisk(tableInfo, properties("tablePath"))
        // remove schema string from map as we don't store carbon schema to hive metastore
        val map = CarbonUtil.removeSchemaFromMap(properties.asJava)
        val updatedFormat = storageFormat.copy(properties = map.asScala.toMap)
        tableDesc.copy(storage = updatedFormat)
      } else {
        tableDesc
      }
    }
  }

  def updateAndCreateTable(
      identifier: AbsoluteTableIdentifier,
      dataSchema: StructType,
      sparkSession: SparkSession,
      metaStore: CarbonMetaStore,
      properties: Map[String, String],
      query: Option[LogicalPlan]): Map[String, String] = {
    val model = createTableInfoFromParams(properties, dataSchema, identifier, query, sparkSession)
    val tableInfo: TableInfo = TableNewProcessor(model)
    val isExternal = properties.getOrElse("isExternal", "false")
    val isTransactionalTable = properties.getOrElse("isTransactional", "true")
      .contains("true")
    val tablePath = properties.getOrElse("path", "")
    tableInfo.setTablePath(identifier.getTablePath)
    tableInfo.setTransactionalTable(isTransactionalTable)
    tableInfo.setDatabaseName(identifier.getDatabaseName)
    val schemaEvolutionEntry = new SchemaEvolutionEntry
    schemaEvolutionEntry.setTimeStamp(tableInfo.getLastUpdatedTime)
    tableInfo.getFactTable.getSchemaEvolution.getSchemaEvolutionEntryList.add(schemaEvolutionEntry)
    val map = if (!metaStore.isReadFromHiveMetaStore && isTransactionalTable) {
      metaStore.saveToDisk(tableInfo, identifier.getTablePath)
      new java.util.HashMap[String, String]()
    } else {
      CarbonUtil.convertToMultiStringMap(tableInfo)
    }
    properties.foreach(e => map.put(e._1, e._2))
    map.put("tablepath", identifier.getTablePath)
    map.put("dbname", identifier.getDatabaseName)
    if (map.containsKey("tableName")) {
      val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
      LOGGER.warn("tableName is not required in options, ignoring it")
    }
    map.put("tableName", identifier.getTableName)
    map.asScala.toMap
  }
}

/**
 * Code ported from Apache Spark
 * Builds a map in which keys are case insensitive. Input map can be accessed for cases where
 * case-sensitive information is required. The primary constructor is marked private to avoid
 * nested case-insensitive map creation, otherwise the keys in the original map will become
 * case-insensitive in this scenario.
 */
case class CaseInsensitiveMap[T] (originalMap: Map[String, T]) extends Map[String, T]
  with Serializable {

  val keyLowerCasedMap = originalMap.map(kv => kv.copy(_1 = kv._1.toLowerCase(Locale.ROOT)))

  override def get(k: String): Option[T] = keyLowerCasedMap.get(k.toLowerCase(Locale.ROOT))

  override def contains(k: String): Boolean =
    keyLowerCasedMap.contains(k.toLowerCase(Locale.ROOT))

  override def +[B1 >: T](kv: (String, B1)): Map[String, B1] = {
    new CaseInsensitiveMap(originalMap + kv)
  }

  override def iterator: Iterator[(String, T)] = keyLowerCasedMap.iterator

  override def -(key: String): Map[String, T] = {
    new CaseInsensitiveMap(originalMap.filterKeys(!_.equalsIgnoreCase(key)))
  }
}

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
19
20	import java.util.Locale
21
22	import scala.collection.JavaConverters._
23	import scala.collection.mutable
24	import scala.language.implicitConversions
25
26	import org.apache.commons.lang.StringUtils
27	import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
28	import org.apache.spark.sql.catalyst.catalog.CatalogTable
29	import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
30	import org.apache.spark.sql.execution.command.{TableModel, TableNewProcessor}
31	import org.apache.spark.sql.execution.strategy.CarbonLateDecodeStrategy
32	import org.apache.spark.sql.execution.streaming.Sink
33	import org.apache.spark.sql.hive.CarbonMetaStore
34	import org.apache.spark.sql.optimizer.CarbonLateDecodeRule
35	import org.apache.spark.sql.parser.CarbonSpark2SqlParser
36	import org.apache.spark.sql.sources._
37	import org.apache.spark.sql.streaming.OutputMode
38	import org.apache.spark.sql.types.StructType
39	import org.apache.spark.sql.util.CarbonException
40
41	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
42	import org.apache.carbondata.common.logging.LogServiceFactory
43	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
44	import org.apache.carbondata.core.metadata.schema.SchemaEvolutionEntry
45	import org.apache.carbondata.core.metadata.schema.table.TableInfo
46	import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
47	import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
48	import org.apache.carbondata.spark.CarbonOption
49	import org.apache.carbondata.spark.util.CarbonScalaUtil
50	import org.apache.carbondata.streaming.{CarbonStreamException, CarbonStreamingQueryListener, StreamSinkFactory}
51
52	/**
53	* Carbon relation provider compliant to data source api.
54	* Creates carbon relations
55	*/
56	class CarbonSource extends CreatableRelationProvider with RelationProvider	2✔
57	with SchemaRelationProvider with StreamSinkProvider with DataSourceRegister {
58
59	override def shortName(): String = "carbondata"	2✔
60
61	// will be called if hive supported create table command is provided
62	override def createRelation(sqlContext: SQLContext,
63	parameters: Map[String, String]): BaseRelation = {
64	CarbonEnv.getInstance(sqlContext.sparkSession)	×
65	// if path is provided we can directly create Hadoop relation. \
66	// Otherwise create datasource relation
67	val newParameters = CarbonScalaUtil.getDeserializedParameters(parameters)	×
68	newParameters.get("tablePath") match {	×
69	case Some(path) => CarbonDatasourceHadoopRelation(sqlContext.sparkSession,	×
70	Array(path),	×
71	newParameters,	×
72	None)	×
73	case _ =>	×
74	val options = new CarbonOption(newParameters)	×
75	val tablePath =	×
76	CarbonEnv.getTablePath(options.dbName, options.tableName)(sqlContext.sparkSession)	×
77	CarbonDatasourceHadoopRelation(sqlContext.sparkSession,	×
78	Array(tablePath),	×
79	newParameters,	×
80	None)	×
81	}
82	}
83
84	// called by any write operation like INSERT INTO DDL or DataFrame.write API
85	override def createRelation(
86	sqlContext: SQLContext,
87	mode: SaveMode,
88	parameters: Map[String, String],
89	data: DataFrame): BaseRelation = {
90	CarbonEnv.getInstance(sqlContext.sparkSession)	2✔
91	var newParameters = CarbonScalaUtil.getDeserializedParameters(parameters)	2✔
92	val options = new CarbonOption(newParameters)	2✔
93	val isExists = CarbonEnv.getInstance(sqlContext.sparkSession).carbonMetastore.tableExists(	2✔
94	options.tableName, options.dbName)(sqlContext.sparkSession)	2✔
95	val (doSave, doAppend) = (mode, isExists) match {	2✔
96	case (SaveMode.ErrorIfExists, true) =>	2✔
97	CarbonException.analysisException(s"table path already exists.")	×
98	case (SaveMode.Overwrite, true) =>	2✔
99	newParameters += (("overwrite", "true"))	1✔
100	(true, false)	1✔
101	case (SaveMode.Overwrite, false) \| (SaveMode.ErrorIfExists, false) =>	2✔
102	newParameters += (("overwrite", "true"))	2✔
103	(true, false)	2✔
104	case (SaveMode.Append, _) =>	1✔
105	(false, true)	1✔
106	case (SaveMode.Ignore, exists) =>	×
107	(!exists, false)	×
108	}
109
110	if (doSave) {	2✔
111	// save data when the save mode is Overwrite.
112	new CarbonDataFrameWriter(sqlContext, data).saveAsCarbonFile(	2✔
113	CaseInsensitiveMap[String](newParameters))	2✔
114	} else if (doAppend) {	1✔
115	new CarbonDataFrameWriter(sqlContext, data).appendToCarbonFile(	1✔
116	CaseInsensitiveMap[String](newParameters))	1✔
117	}
118
119	createRelation(sqlContext, newParameters, data.schema)	2✔
120	}
121
122	// called by DDL operation with a USING clause
123	override def createRelation(
124	sqlContext: SQLContext,
125	parameters: Map[String, String],
126	dataSchema: StructType): BaseRelation = {
127	CarbonEnv.getInstance(sqlContext.sparkSession)	2✔
128	addLateDecodeOptimization(sqlContext.sparkSession)	2✔
129	val newParameters =	2✔
130	CaseInsensitiveMap[String](CarbonScalaUtil.getDeserializedParameters(parameters))	2✔
131	val dbName: String =	2✔
132	CarbonEnv.getDatabaseName(newParameters.get("dbName"))(sqlContext.sparkSession)	2✔
133	val tableOption: Option[String] = newParameters.get("tableName")	2✔
134	if (tableOption.isEmpty) {	2✔
135	CarbonException.analysisException("Table creation failed. Table name is not specified")	×
136	}
137	val tableName = tableOption.get.toLowerCase()	2✔
138	if (tableName.contains(" ")) {	2✔
139	CarbonException.analysisException(	×
140	"Table creation failed. Table name cannot contain blank space")	×
141	}
142	val (path, updatedParams) = if (sqlContext.sparkSession.sessionState.catalog.listTables(dbName)	2✔
143	.exists(_.table.equalsIgnoreCase(tableName))) {	2✔
144	getPathForTable(sqlContext.sparkSession, dbName, tableName, newParameters)	2✔
145	} else {
146	createTableIfNotExists(sqlContext.sparkSession, newParameters, dataSchema)	2✔
147	}
148
149	CarbonDatasourceHadoopRelation(sqlContext.sparkSession, Array(path), updatedParams,	2✔
150	Option(dataSchema))	2✔
151	}
152
153	private def addLateDecodeOptimization(ss: SparkSession): Unit = {
154	if (ss.sessionState.experimentalMethods.extraStrategies.isEmpty) {	2✔
155	ss.sessionState.experimentalMethods.extraStrategies = Seq(new CarbonLateDecodeStrategy)	×
156	ss.sessionState.experimentalMethods.extraOptimizations = Seq(new CarbonLateDecodeRule)	×
157	}
158	}
159
160
161	private def createTableIfNotExists(
162	sparkSession: SparkSession,
163	parameters: Map[String, String],
164	dataSchema: StructType): (String, Map[String, String]) = {	2✔
165
166	val dbName: String = CarbonEnv.getDatabaseName(parameters.get("dbName"))(sparkSession)	2✔
167	val tableName: String = parameters.getOrElse("tableName", "").toLowerCase	2✔
168
169	try {	2✔
170	val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)	2✔
171	(carbonTable.getTablePath, parameters)	2✔
172	} catch {
173	case _: NoSuchTableException =>
174	val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
175	val identifier = AbsoluteTableIdentifier.from(	2✔
176	CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession),	2✔
177	dbName,	2✔
178	tableName)	2✔
179	val updatedParams = CarbonSource.updateAndCreateTable(	2✔
180	identifier, dataSchema, sparkSession, metaStore, parameters, None)	2✔
181	(CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession), updatedParams)	2✔
182	case ex: Exception =>
183	throw new Exception("do not have dbname and tablename for carbon table", ex)	×
184	}
185	}
186
187	/**
188	* Returns the path of the table
189	*
190	* @param sparkSession
191	* @param dbName
192	* @param tableName
193	* @return
194	*/
195	private def getPathForTable(sparkSession: SparkSession, dbName: String,
196	tableName : String, parameters: Map[String, String]): (String, Map[String, String]) = {	2✔
197
198	if (StringUtils.isBlank(tableName)) {	2✔
199	throw new MalformedCarbonCommandException("The Specified Table Name is Blank")	×
200	}
201	if (tableName.contains(" ")) {	2✔
202	throw new MalformedCarbonCommandException("Table Name Should not have spaces ")	×
203	}
204	try {	×
205	if (parameters.contains("tablePath")) {	2✔
206	(parameters("tablePath"), parameters)	2✔
207	} else if (!sparkSession.isInstanceOf[CarbonSession]) {	2✔
208	(CarbonProperties.getStorePath + "/" + dbName + "/" + tableName, parameters)	×
209	} else {
210	(CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession), parameters)	2✔
211	}
212	} catch {
213	case ex: Exception =>
214	throw new Exception(s"Do not have $dbName and $tableName", ex)	×
215	}
216	}
217
218	/**
219	* produce a streaming `Sink` for a specific format
220	* now it will create a default sink(CarbonAppendableStreamSink) for row format
221	*/
222	override def createSink(sqlContext: SQLContext,
223	parameters: Map[String, String],
224	partitionColumns: Seq[String],
225	outputMode: OutputMode): Sink = {
226
227	// check "tablePath" option
228	val options = new CarbonOption(parameters)	2✔
229	val dbName = CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)	2✔
230	val tableName = options.tableName	2✔
231	if (tableName.contains(" ")) {	2✔
232	throw new CarbonStreamException("Table creation failed. Table name cannot contain blank " +	×
233	"space")
234	}
235	val sparkSession = sqlContext.sparkSession	2✔
236	val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)	2✔
237	if (!carbonTable.isStreamingSink) {	2✔
238	throw new CarbonStreamException(s"Table ${carbonTable.getDatabaseName}." +	2✔
239	s"${carbonTable.getTableName} is not a streaming table")	2✔
240	}
241
242	// CarbonSession has added CarbonStreamingQueryListener during the initialization.
243	// But other SparkSessions didn't, so here will add the listener once.
244	if (!"CarbonSession".equals(sparkSession.getClass.getSimpleName)) {	2✔
245	if (CarbonSource.listenerAdded.get(sparkSession.hashCode()).isEmpty) {	×
246	synchronized {	×
247	if (CarbonSource.listenerAdded.get(sparkSession.hashCode()).isEmpty) {	×
248	sparkSession.streams.addListener(new CarbonStreamingQueryListener(sparkSession))	×
249	CarbonSource.listenerAdded.put(sparkSession.hashCode(), true)	×
250	}
251	}
252	}
253	}
254
255	// create sink
256	StreamSinkFactory.createStreamTableSink(	2✔
257	sqlContext.sparkSession,	2✔
258	sqlContext.sparkSession.sessionState.newHadoopConf(),	2✔
259	carbonTable,	2✔
260	parameters)	2✔
261	}
262
263	}
264
265	object CarbonSource {
266
267	lazy val listenerAdded = new mutable.HashMap[Int, Boolean]()	×
268
269	def createTableInfoFromParams(
270	parameters: Map[String, String],
271	dataSchema: StructType,
272	identifier: AbsoluteTableIdentifier,
273	query: Option[LogicalPlan],
274	sparkSession: SparkSession): TableModel = {
275	val sqlParser = new CarbonSpark2SqlParser	2✔
276	val map = scala.collection.mutable.Map[String, String]()	2✔
277	parameters.foreach { case (key, value) => map.put(key, value.toLowerCase()) }	2✔
278	val options = new CarbonOption(parameters)	2✔
279	val fields = query match {	2✔
280	case Some(q) =>	2✔
281	// if query is provided then it is a CTAS flow
282	if (sqlParser.getFields(dataSchema).nonEmpty) {	1✔
283	throw new AnalysisException(	×
284	"Schema cannot be specified in a Create Table As Select (CTAS) statement")	×
285	}
286	sqlParser	1✔
287	.getFields(CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
288	.getSchemaFromUnresolvedRelation(sparkSession, q))	1✔
289	case None =>	2✔
290	sqlParser.getFields(dataSchema)	2✔
291	}
292	val bucketFields = sqlParser.getBucketFields(map, fields, options)	2✔
293	sqlParser.prepareTableModel(ifNotExistPresent = false, Option(identifier.getDatabaseName),	2✔
294	identifier.getTableName, fields, Nil, map, bucketFields)	2✔
295	}
296
297	/**
298	* Update spark catalog table with schema information in case of schema storage is hive metastore
299	* @param tableDesc
300	* @param sparkSession
301	* @return
302	*/
303	def updateCatalogTableWithCarbonSchema(
304	tableDesc: CatalogTable,
305	sparkSession: SparkSession,
306	query: Option[LogicalPlan] = None): CatalogTable = {	2✔
307	val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
308	val storageFormat = tableDesc.storage	2✔
309	val properties = storageFormat.properties	2✔
310	if (!properties.contains("carbonSchemaPartsNo")) {	2✔
311	val tablePath = CarbonEnv.getTablePath(	2✔
312	tableDesc.identifier.database, tableDesc.identifier.table)(sparkSession)	2✔
313	val dbName = CarbonEnv.getDatabaseName(tableDesc.identifier.database)(sparkSession)	2✔
314	val identifier = AbsoluteTableIdentifier.from(tablePath, dbName, tableDesc.identifier.table)	2✔
315	val map = updateAndCreateTable(	2✔
316	identifier,	2✔
317	tableDesc.schema,	2✔
318	sparkSession,	2✔
319	metaStore,	2✔
320	properties,	2✔
321	query)	2✔
322	// updating params
323	val updatedFormat = storageFormat.copy(properties = map)	2✔
324	tableDesc.copy(storage = updatedFormat)	2✔
325	} else {
326	val tableInfo = CarbonUtil.convertGsonToTableInfo(properties.asJava)	2✔
327	val isExternal = properties.getOrElse("isExternal", "false")	2✔
328	val isTransactionalTable = properties.getOrElse("isTransactional", "true")	2✔
329	.contains("true")	2✔
330	tableInfo.setTransactionalTable(isTransactionalTable)	2✔
331	if (isTransactionalTable && !metaStore.isReadFromHiveMetaStore) {	2✔
332	// save to disk
333	metaStore.saveToDisk(tableInfo, properties("tablePath"))	2✔
334	// remove schema string from map as we don't store carbon schema to hive metastore
335	val map = CarbonUtil.removeSchemaFromMap(properties.asJava)	2✔
336	val updatedFormat = storageFormat.copy(properties = map.asScala.toMap)	2✔
337	tableDesc.copy(storage = updatedFormat)	2✔
338	} else {
339	tableDesc	1✔
340	}
341	}
342	}
343
344	def updateAndCreateTable(
345	identifier: AbsoluteTableIdentifier,
346	dataSchema: StructType,
347	sparkSession: SparkSession,
348	metaStore: CarbonMetaStore,
349	properties: Map[String, String],
350	query: Option[LogicalPlan]): Map[String, String] = {
351	val model = createTableInfoFromParams(properties, dataSchema, identifier, query, sparkSession)	2✔
352	val tableInfo: TableInfo = TableNewProcessor(model)	2✔
353	val isExternal = properties.getOrElse("isExternal", "false")	2✔
354	val isTransactionalTable = properties.getOrElse("isTransactional", "true")	2✔
355	.contains("true")	2✔
356	val tablePath = properties.getOrElse("path", "")	2✔
357	tableInfo.setTablePath(identifier.getTablePath)	2✔
358	tableInfo.setTransactionalTable(isTransactionalTable)	2✔
359	tableInfo.setDatabaseName(identifier.getDatabaseName)	2✔
360	val schemaEvolutionEntry = new SchemaEvolutionEntry	2✔
361	schemaEvolutionEntry.setTimeStamp(tableInfo.getLastUpdatedTime)	2✔
362	tableInfo.getFactTable.getSchemaEvolution.getSchemaEvolutionEntryList.add(schemaEvolutionEntry)	2✔
363	val map = if (!metaStore.isReadFromHiveMetaStore && isTransactionalTable) {	2✔
364	metaStore.saveToDisk(tableInfo, identifier.getTablePath)	2✔
365	new java.util.HashMap[String, String]()	2✔
366	} else {
367	CarbonUtil.convertToMultiStringMap(tableInfo)	×
368	}
369	properties.foreach(e => map.put(e._1, e._2))	2✔
370	map.put("tablepath", identifier.getTablePath)	2✔
371	map.put("dbname", identifier.getDatabaseName)	2✔
372	if (map.containsKey("tableName")) {	2✔
373	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	2✔
374	LOGGER.warn("tableName is not required in options, ignoring it")	2✔
375	}
376	map.put("tableName", identifier.getTableName)	2✔
377	map.asScala.toMap	2✔
378	}
379	}
380
381	/**
382	* Code ported from Apache Spark
383	* Builds a map in which keys are case insensitive. Input map can be accessed for cases where
384	* case-sensitive information is required. The primary constructor is marked private to avoid
385	* nested case-insensitive map creation, otherwise the keys in the original map will become
386	* case-insensitive in this scenario.
387	*/
388	case class CaseInsensitiveMap[T] (originalMap: Map[String, T]) extends Map[String, T]	2✔
389	with Serializable {
390
391	val keyLowerCasedMap = originalMap.map(kv => kv.copy(_1 = kv._1.toLowerCase(Locale.ROOT)))	2✔
392
393	override def get(k: String): Option[T] = keyLowerCasedMap.get(k.toLowerCase(Locale.ROOT))	2✔
394
395	override def contains(k: String): Boolean =
396	keyLowerCasedMap.contains(k.toLowerCase(Locale.ROOT))	2✔
397
398	override def +[B1 >: T](kv: (String, B1)): Map[String, B1] = {
399	new CaseInsensitiveMap(originalMap + kv)	×
400	}
401
402	override def iterator: Iterator[(String, T)] = keyLowerCasedMap.iterator	2✔
403
404	override def -(key: String): Map[String, T] = {
405	new CaseInsensitiveMap(originalMap.filterKeys(!_.equalsIgnoreCase(key)))	×
406	}
407	}

apache / carbondata / 135

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