11863

Build # 11863

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a">3c0c1970f<a href="https://github.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/4790f8073addb42583888aa181f0cb9331d672c6">4790f8073</a>">4790f8073</a><a href="https://github.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a"> into 14f170c5d">14f170c5d</a>

Run Details

64031 of 82738 relevant lines covered (77.39%)

1.01 hits per line

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

89.17

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.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.execution.command.table

import scala.collection.JavaConverters._

import org.apache.spark.sql.{CarbonEnv, Row, SparkSession, _}
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.SQLExecution.EXECUTION_ID_KEY
import org.apache.spark.sql.execution.command.MetadataCommand

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.compression.CompressorFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.exception.InvalidConfigurationException
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.partition.PartitionType
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo}
import org.apache.carbondata.core.util.{CarbonUtil, ThreadLocalSessionInfo}
import org.apache.carbondata.events.{CreateTablePostExecutionEvent, CreateTablePreExecutionEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.spark.util.CarbonSparkUtil

case class CarbonCreateTableCommand(
    tableInfo: TableInfo,
    ifNotExistsSet: Boolean = false,
    tableLocation: Option[String] = None,
    isExternal : Boolean = false,
    createDSTable: Boolean = true,
    isVisible: Boolean = true)
  extends MetadataCommand {

  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
    val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
    val tableName = tableInfo.getFactTable.getTableName
    var databaseOpt : Option[String] = None
    ThreadLocalSessionInfo
      .setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())
    if (tableInfo.getDatabaseName != null) {
      databaseOpt = Some(tableInfo.getDatabaseName)
    }
    val dbName = CarbonEnv.getDatabaseName(databaseOpt)(sparkSession)
    setAuditTable(dbName, tableName)
    setAuditInfo(tableInfo.getFactTable.getTableProperties.asScala.toMap
                 ++ Map("external" -> isExternal.toString))
    // set dbName and tableUnique Name in the table info
    tableInfo.setDatabaseName(dbName)
    tableInfo.setTableUniqueName(CarbonTable.buildUniqueName(dbName, tableName))
    val isTransactionalTable = tableInfo.isTransactionalTable
    if (sparkSession.sessionState.catalog.listTables(dbName)
      .exists(_.table.equalsIgnoreCase(tableName))) {
      if (!ifNotExistsSet) {
        throw new TableAlreadyExistsException(dbName, tableName)
      }
    } else {
      val path = tableLocation.getOrElse(
        CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession))
      val tablePath = if (FileFactory.getCarbonFile(path).exists() && !isExternal &&
                          isTransactionalTable && tableLocation.isEmpty) {
        path + "_" + tableInfo.getFactTable.getTableId
      } else {
        path
      }
      val streaming = tableInfo.getFactTable.getTableProperties.get("streaming")
      if (streaming != null && streaming.equalsIgnoreCase("true") && path.startsWith("s3")) {
        throw new UnsupportedOperationException("streaming is not supported with s3 store")
      }
      tableInfo.setTablePath(tablePath)
      val tableIdentifier = AbsoluteTableIdentifier
        .from(tablePath, dbName, tableName, tableInfo.getFactTable.getTableId)

      // Add validation for sort scope when create table
      val sortScope = tableInfo.getFactTable.getTableProperties.asScala
        .getOrElse("sort_scope", CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT)
      if (!CarbonUtil.isValidSortOption(sortScope)) {
        throw new InvalidConfigurationException(
          s"Passing invalid SORT_SCOPE '$sortScope', valid SORT_SCOPE are 'NO_SORT'," +
          s" 'BATCH_SORT', 'LOCAL_SORT' and 'GLOBAL_SORT' ")
      }

      if (tableInfo.getFactTable.getListOfColumns.size <= 0) {
        throwMetadataException(dbName, tableName, "Table should have at least one column.")
      }

      // Add validatation for column compressor when create table
      val columnCompressor = tableInfo.getFactTable.getTableProperties.get(
        CarbonCommonConstants.COMPRESSOR)
      try {
        if (null != columnCompressor) {
          CompressorFactory.getInstance().getCompressor(columnCompressor)
        }
      } catch {
        case ex : UnsupportedOperationException =>
          throw new InvalidConfigurationException(ex.getMessage)
      }

      val operationContext = new OperationContext
      val createTablePreExecutionEvent: CreateTablePreExecutionEvent =
        CreateTablePreExecutionEvent(sparkSession, tableIdentifier, Some(tableInfo))
      OperationListenerBus.getInstance.fireEvent(createTablePreExecutionEvent, operationContext)
      val catalog = CarbonEnv.getInstance(sparkSession).carbonMetaStore
      val carbonSchemaString = catalog.generateTableSchemaString(tableInfo, tableIdentifier)
      if (createDSTable) {
        try {
          val tablePath = tableIdentifier.getTablePath
          val carbonRelation = CarbonSparkUtil.createCarbonRelation(tableInfo, tablePath)
          val rawSchema = CarbonSparkUtil.getRawSchema(carbonRelation)
          sparkSession.sparkContext.setLocalProperty(EXECUTION_ID_KEY, null)
          val partitionInfo = tableInfo.getFactTable.getPartitionInfo
          val partitionString =
            if (partitionInfo != null &&
                partitionInfo.getPartitionType == PartitionType.NATIVE_HIVE) {
              // Restrict dictionary encoding on partition columns.
              // TODO Need to decide whether it is required
              val dictionaryOnPartitionColumn =
              partitionInfo.getColumnSchemaList.asScala.exists{p =>
                p.hasEncoding(Encoding.DICTIONARY) && !p.hasEncoding(Encoding.DIRECT_DICTIONARY)
              }
              if (dictionaryOnPartitionColumn) {
                throwMetadataException(
                  dbName,
                  tableName,
                  s"Dictionary include cannot be applied on partition columns")
              }
              s" PARTITIONED BY (${partitionInfo.getColumnSchemaList.asScala.map(
                _.getColumnName).mkString(",")})"
            } else {
              ""
            }

          // synchronized to prevent concurrently creation of table with same name
          CarbonCreateTableCommand.synchronized {
            // isVisible property is added to hive table properties to differentiate between main
            // table and datamaps(like preaggregate). It is false only for datamaps. This is added
            // to improve the show tables performance when filtering the datamaps from main tables
            sparkSession.sql(
              s"""CREATE TABLE $dbName.$tableName
                 |(${ rawSchema })
                 |USING org.apache.spark.sql.CarbonSource
                 |OPTIONS (
                 |  tableName "$tableName",
                 |  dbName "$dbName",
                 |  tablePath "$tablePath",
                 |  path "${FileFactory.addSchemeIfNotExists(tablePath)}",
                 |  isExternal "$isExternal",
                 |  isTransactional "$isTransactionalTable",
                 |  isVisible "$isVisible"
                 |  $carbonSchemaString)
                 |  $partitionString
             """.stripMargin)
          }
        } catch {
          case e: AnalysisException =>
            // AnalysisException thrown with table already exists msg incase of conurrent drivers
            if (e.getMessage().contains("already exists")) {

              // Clear the cache first
              CarbonEnv.getInstance(sparkSession).carbonMetaStore
                .removeTableFromMetadata(dbName, tableName)

              // Delete the folders created by this call if the actual path is different
              val actualPath = CarbonEnv
                .getCarbonTable(TableIdentifier(tableName, Option(dbName)))(sparkSession)
                .getTablePath

              if (!actualPath.equalsIgnoreCase(tablePath)) {
                LOGGER
                  .error(
                    "TableAlreadyExists with path : " + actualPath + " So, deleting " + tablePath)
                FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(tablePath))
              }

              // No need to throw for create if not exists
              if (ifNotExistsSet) {
                LOGGER.error(e, e)
              } else {
                LOGGER.error(e)
                throw e
              }
            } else {
              LOGGER.error(e)
              throw e
            }

          case e: Exception =>
            // call the drop table to delete the created table.
            try {
              CarbonEnv.getInstance(sparkSession).carbonMetaStore
                .dropTable(tableIdentifier)(sparkSession)
            } catch {
              case _: Exception => // No operation
            }
            val msg = s"Create table'$tableName' in database '$dbName' failed"
            throwMetadataException(dbName, tableName, msg.concat(", ").concat(e.getMessage))
        }
      }
      val createTablePostExecutionEvent: CreateTablePostExecutionEvent =
        CreateTablePostExecutionEvent(sparkSession, tableIdentifier)
      OperationListenerBus.getInstance.fireEvent(createTablePostExecutionEvent, operationContext)
    }
    Seq.empty
  }

  override protected def opName: String = "CREATE TABLE"
}

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.execution.command.table
19
20	import scala.collection.JavaConverters._
21
22	import org.apache.spark.sql.{CarbonEnv, Row, SparkSession, _}
23	import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
24	import org.apache.spark.sql.catalyst.TableIdentifier
25	import org.apache.spark.sql.execution.SQLExecution.EXECUTION_ID_KEY
26	import org.apache.spark.sql.execution.command.MetadataCommand
27
28	import org.apache.carbondata.common.logging.LogServiceFactory
29	import org.apache.carbondata.core.constants.CarbonCommonConstants
30	import org.apache.carbondata.core.datastore.compression.CompressorFactory
31	import org.apache.carbondata.core.datastore.impl.FileFactory
32	import org.apache.carbondata.core.exception.InvalidConfigurationException
33	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
34	import org.apache.carbondata.core.metadata.encoder.Encoding
35	import org.apache.carbondata.core.metadata.schema.partition.PartitionType
36	import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo}
37	import org.apache.carbondata.core.util.{CarbonUtil, ThreadLocalSessionInfo}
38	import org.apache.carbondata.events.{CreateTablePostExecutionEvent, CreateTablePreExecutionEvent, OperationContext, OperationListenerBus}
39	import org.apache.carbondata.spark.util.CarbonSparkUtil
40
41	case class CarbonCreateTableCommand(	2✔
42	tableInfo: TableInfo,	2✔
43	ifNotExistsSet: Boolean = false,	2✔
44	tableLocation: Option[String] = None,	2✔
45	isExternal : Boolean = false,	2✔
46	createDSTable: Boolean = true,	2✔
47	isVisible: Boolean = true)	2✔
48	extends MetadataCommand {	2✔
49
50	override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
51	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	2✔
52	val tableName = tableInfo.getFactTable.getTableName	2✔
53	var databaseOpt : Option[String] = None	2✔
54	ThreadLocalSessionInfo
55	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
56	if (tableInfo.getDatabaseName != null) {	2✔
57	databaseOpt = Some(tableInfo.getDatabaseName)	2✔
58	}
59	val dbName = CarbonEnv.getDatabaseName(databaseOpt)(sparkSession)	2✔
60	setAuditTable(dbName, tableName)	2✔
61	setAuditInfo(tableInfo.getFactTable.getTableProperties.asScala.toMap	2✔
62	++ Map("external" -> isExternal.toString))	2✔
63	// set dbName and tableUnique Name in the table info
64	tableInfo.setDatabaseName(dbName)	2✔
65	tableInfo.setTableUniqueName(CarbonTable.buildUniqueName(dbName, tableName))	2✔
66	val isTransactionalTable = tableInfo.isTransactionalTable	2✔
67	if (sparkSession.sessionState.catalog.listTables(dbName)	2✔
68	.exists(_.table.equalsIgnoreCase(tableName))) {	2✔
69	if (!ifNotExistsSet) {	1✔
70	throw new TableAlreadyExistsException(dbName, tableName)	×
71	}
72	} else {
73	val path = tableLocation.getOrElse(	2✔
74	CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession))	2✔
75	val tablePath = if (FileFactory.getCarbonFile(path).exists() && !isExternal &&	2✔
76	isTransactionalTable && tableLocation.isEmpty) {	2✔
77	path + "_" + tableInfo.getFactTable.getTableId	2✔
78	} else {
79	path	2✔
80	}
81	val streaming = tableInfo.getFactTable.getTableProperties.get("streaming")	2✔
82	if (streaming != null && streaming.equalsIgnoreCase("true") && path.startsWith("s3")) {	2✔
83	throw new UnsupportedOperationException("streaming is not supported with s3 store")	×
84	}
85	tableInfo.setTablePath(tablePath)	2✔
86	val tableIdentifier = AbsoluteTableIdentifier	2✔
87	.from(tablePath, dbName, tableName, tableInfo.getFactTable.getTableId)	2✔
88
89	// Add validation for sort scope when create table
90	val sortScope = tableInfo.getFactTable.getTableProperties.asScala	2✔
91	.getOrElse("sort_scope", CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT)	2✔
92	if (!CarbonUtil.isValidSortOption(sortScope)) {	2✔
93	throw new InvalidConfigurationException(	1✔
94	s"Passing invalid SORT_SCOPE '$sortScope', valid SORT_SCOPE are 'NO_SORT'," +	1✔
95	s" 'BATCH_SORT', 'LOCAL_SORT' and 'GLOBAL_SORT' ")	1✔
96	}
97
98	if (tableInfo.getFactTable.getListOfColumns.size <= 0) {	2✔
99	throwMetadataException(dbName, tableName, "Table should have at least one column.")	×
100	}
101
102	// Add validatation for column compressor when create table
103	val columnCompressor = tableInfo.getFactTable.getTableProperties.get(	2✔
104	CarbonCommonConstants.COMPRESSOR)	2✔
105	try {	2✔
106	if (null != columnCompressor) {	2✔
107	CompressorFactory.getInstance().getCompressor(columnCompressor)	2✔
108	}
109	} catch {
110	case ex : UnsupportedOperationException =>
111	throw new InvalidConfigurationException(ex.getMessage)	×
112	}
113
114	val operationContext = new OperationContext	2✔
115	val createTablePreExecutionEvent: CreateTablePreExecutionEvent =	2✔
116	CreateTablePreExecutionEvent(sparkSession, tableIdentifier, Some(tableInfo))	2✔
117	OperationListenerBus.getInstance.fireEvent(createTablePreExecutionEvent, operationContext)	2✔
118	val catalog = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
119	val carbonSchemaString = catalog.generateTableSchemaString(tableInfo, tableIdentifier)	2✔
120	if (createDSTable) {	2✔
121	try {	1✔
122	val tablePath = tableIdentifier.getTablePath	2✔
123	val carbonRelation = CarbonSparkUtil.createCarbonRelation(tableInfo, tablePath)	2✔
124	val rawSchema = CarbonSparkUtil.getRawSchema(carbonRelation)	2✔
125	sparkSession.sparkContext.setLocalProperty(EXECUTION_ID_KEY, null)	2✔
126	val partitionInfo = tableInfo.getFactTable.getPartitionInfo	2✔
127	val partitionString =	2✔
128	if (partitionInfo != null &&	2✔
129	partitionInfo.getPartitionType == PartitionType.NATIVE_HIVE) {	2✔
130	// Restrict dictionary encoding on partition columns.
131	// TODO Need to decide whether it is required
132	val dictionaryOnPartitionColumn =	2✔
133	partitionInfo.getColumnSchemaList.asScala.exists{p =>	2✔
134	p.hasEncoding(Encoding.DICTIONARY) && !p.hasEncoding(Encoding.DIRECT_DICTIONARY)	2✔
135	}
136	if (dictionaryOnPartitionColumn) {	2✔
137	throwMetadataException(	1✔
138	dbName,	1✔
139	tableName,	1✔
140	s"Dictionary include cannot be applied on partition columns")	1✔
141	}
142	s" PARTITIONED BY (${partitionInfo.getColumnSchemaList.asScala.map(	2✔
143	_.getColumnName).mkString(",")})"	2✔
144	} else {
145	""	2✔
146	}
147
148	// synchronized to prevent concurrently creation of table with same name
149	CarbonCreateTableCommand.synchronized {	2✔
150	// isVisible property is added to hive table properties to differentiate between main
151	// table and datamaps(like preaggregate). It is false only for datamaps. This is added
152	// to improve the show tables performance when filtering the datamaps from main tables
153	sparkSession.sql(	2✔
154	s"""CREATE TABLE $dbName.$tableName	2✔
155	\|(${ rawSchema })	2✔
156	\|USING org.apache.spark.sql.CarbonSource
157	\|OPTIONS (
158	\| tableName "$tableName",	2✔
159	\| dbName "$dbName",	2✔
160	\| tablePath "$tablePath",	2✔
161	\| path "${FileFactory.addSchemeIfNotExists(tablePath)}",	2✔
162	\| isExternal "$isExternal",	2✔
163	\| isTransactional "$isTransactionalTable",	2✔
164	\| isVisible "$isVisible"	2✔
165	\| $carbonSchemaString)	2✔
166	\| $partitionString	2✔
167	""".stripMargin)	2✔
168	}
169	} catch {
170	case e: AnalysisException =>
171	// AnalysisException thrown with table already exists msg incase of conurrent drivers
172	if (e.getMessage().contains("already exists")) {	1✔
173
174	// Clear the cache first
175	CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
176	.removeTableFromMetadata(dbName, tableName)	1✔
177
178	// Delete the folders created by this call if the actual path is different
179	val actualPath = CarbonEnv	1✔
180	.getCarbonTable(TableIdentifier(tableName, Option(dbName)))(sparkSession)	1✔
181	.getTablePath	1✔
182
183	if (!actualPath.equalsIgnoreCase(tablePath)) {	1✔
184	LOGGER	×
185	.error(	×
186	"TableAlreadyExists with path : " + actualPath + " So, deleting " + tablePath)	×
187	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(tablePath))	×
188	}
189
190	// No need to throw for create if not exists
191	if (ifNotExistsSet) {	1✔
192	LOGGER.error(e, e)	1✔
193	} else {
194	LOGGER.error(e)	×
195	throw e	×
196	}
197	} else {
198	LOGGER.error(e)	1✔
199	throw e	1✔
200	}
201
202	case e: Exception =>	×
203	// call the drop table to delete the created table.
204	try {	×
205	CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
206	.dropTable(tableIdentifier)(sparkSession)	1✔
207	} catch {
208	case _: Exception => // No operation
209	}
210	val msg = s"Create table'$tableName' in database '$dbName' failed"	1✔
211	throwMetadataException(dbName, tableName, msg.concat(", ").concat(e.getMessage))	×
212	}
213	}
214	val createTablePostExecutionEvent: CreateTablePostExecutionEvent =	2✔
215	CreateTablePostExecutionEvent(sparkSession, tableIdentifier)	2✔
216	OperationListenerBus.getInstance.fireEvent(createTablePostExecutionEvent, operationContext)	2✔
217	}
218	Seq.empty	2✔
219	}
220
221	override protected def opName: String = "CREATE TABLE"	2✔
222	}

apache / carbondata / 11863

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