apache / carbondata / 260

Build # 260

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/a3b2694c7556b4e09fc2de3f12189b2244b05f53">a3b2694c7">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/a3b2694c7556b4e09fc2de3f12189b2244b05f53">a3b2694c7</a><a href="https://github.com/apache/carbondata/commit/a3b2694c7556b4e09fc2de3f12189b2244b05f53">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/a3b2694c7556b4e09fc2de3f12189b2244b05f53&quot;&gt;a3b2694c7&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/a3b2694c7556b4e09fc2de3f12189b2244b05f53&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

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

57778 of 71585 relevant lines covered (80.71%)

1.07 hits per line

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

80.65

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

import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, UnresolvedRelation}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.command.management.{CarbonAlterTableCompactionCommand, CarbonInsertIntoCommand, CarbonLoadDataCommand, RefreshCarbonTableCommand}
import org.apache.spark.sql.execution.command.partition.{CarbonAlterTableAddHivePartitionCommand, CarbonAlterTableDropHivePartitionCommand, CarbonShowCarbonPartitionsCommand}
import org.apache.spark.sql.execution.command.schema._
import org.apache.spark.sql.execution.command.table.{CarbonDescribeFormattedCommand, CarbonDropTableCommand}
import org.apache.spark.sql.hive.execution.command.{CarbonDropDatabaseCommand, CarbonResetCommand, CarbonSetCommand}
import org.apache.spark.sql.CarbonExpressions.{CarbonDescribeTable => DescribeTableCommand}
import org.apache.spark.sql.execution.datasources.{RefreshResource, RefreshTable}
import org.apache.spark.sql.hive.{CarbonRelation, CreateCarbonSourceTableAsSelectCommand}
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
import org.apache.spark.util.{CarbonReflectionUtils, FileUtils}

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.{CarbonProperties, ThreadLocalSessionInfo}

/**
 * Carbon strategies for ddl commands
 */

class DDLStrategy(sparkSession: SparkSession) extends SparkStrategy {
  val LOGGER: LogService =
    LogServiceFactory.getLogService(this.getClass.getName)
  def apply(plan: LogicalPlan): Seq[SparkPlan] = {
    plan match {
      case LoadDataCommand(identifier, path, isLocal, isOverwrite, partition)
        if CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(identifier)(sparkSession) =>
        ExecutedCommandExec(
          CarbonLoadDataCommand(
            databaseNameOp = identifier.database,
            tableName = identifier.table.toLowerCase,
            factPathFromUser = path,
            dimFilesPath = Seq(),
            options = Map(),
            isOverwriteTable = isOverwrite,
            inputSqlString = null,
            dataFrame = None,
            updateModel = None,
            tableInfoOp = None,
            internalOptions = Map.empty,
            partition = partition.getOrElse(Map.empty).map { case (col, value) =>
              (col, Some(value))})) :: Nil
      case alter@AlterTableRenameCommand(oldTableIdentifier, newTableIdentifier, _) =>
        val dbOption = oldTableIdentifier.database.map(_.toLowerCase)
        val tableIdentifier = TableIdentifier(oldTableIdentifier.table.toLowerCase(), dbOption)
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableIdentifier)(sparkSession)
        if (isCarbonTable) {
          val renameModel = AlterTableRenameModel(tableIdentifier, newTableIdentifier)
          ExecutedCommandExec(CarbonAlterTableRenameCommand(renameModel)) :: Nil
        } else {
          ExecutedCommandExec(alter) :: Nil
        }
      case DropTableCommand(identifier, ifNotExists, isView, _)
        if CarbonEnv.getInstance(sparkSession).carbonMetastore
          .isTablePathExists(identifier)(sparkSession) =>
        ExecutedCommandExec(
          CarbonDropTableCommand(ifNotExists, identifier.database,
            identifier.table.toLowerCase)) :: Nil
      case InsertIntoCarbonTable(relation: CarbonDatasourceHadoopRelation,
      partition, child: LogicalPlan, overwrite, _) =>
        ExecutedCommandExec(CarbonInsertIntoCommand(relation, child, overwrite, partition)) :: Nil
      case createDb@CreateDatabaseCommand(dbName, ifNotExists, _, _, _) =>
        val dbLocation = try {
          CarbonEnv.getDatabaseLocation(dbName, sparkSession)
        } catch {
          case e: NoSuchDatabaseException =>
            CarbonProperties.getStorePath
        }
        ThreadLocalSessionInfo
          .setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())
        FileUtils.createDatabaseDirectory(dbName, dbLocation, sparkSession.sparkContext)
        ExecutedCommandExec(createDb) :: Nil
      case drop@DropDatabaseCommand(dbName, ifExists, isCascade) =>
        ExecutedCommandExec(CarbonDropDatabaseCommand(drop)) :: Nil
      case alterTable@CarbonAlterTableCompactionCommand(altertablemodel, _, _) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(TableIdentifier(altertablemodel.tableName,
            altertablemodel.dbName))(sparkSession)
        if (isCarbonTable) {
            ExecutedCommandExec(alterTable) :: Nil
        } else {
          throw new MalformedCarbonCommandException(
            "Operation not allowed : " + altertablemodel.alterSql)
        }
      case dataTypeChange@CarbonAlterTableDataTypeChangeCommand(alterTableChangeDataTypeModel) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(TableIdentifier(alterTableChangeDataTypeModel.tableName,
            alterTableChangeDataTypeModel.databaseName))(sparkSession)
        if (isCarbonTable) {
          val carbonTable = CarbonEnv.getCarbonTable(alterTableChangeDataTypeModel.databaseName,
            alterTableChangeDataTypeModel.tableName)(sparkSession)
          if (carbonTable != null && carbonTable.isFileLevelFormat) {
            throw new MalformedCarbonCommandException(
              "Unsupported alter operation on Carbon external fileformat table")
          } else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {
            throw new MalformedCarbonCommandException(
              "Unsupported operation on non transactional table")
          } else {
            ExecutedCommandExec(dataTypeChange) :: Nil
          }
        } else {
          throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")
        }
      case addColumn@CarbonAlterTableAddColumnCommand(alterTableAddColumnsModel) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(TableIdentifier(alterTableAddColumnsModel.tableName,
            alterTableAddColumnsModel.databaseName))(sparkSession)
        if (isCarbonTable) {
          val carbonTable = CarbonEnv.getCarbonTable(alterTableAddColumnsModel.databaseName,
            alterTableAddColumnsModel.tableName)(sparkSession)
          if (carbonTable != null && carbonTable.isFileLevelFormat) {
            throw new MalformedCarbonCommandException(
              "Unsupported alter operation on Carbon external fileformat table")
          } else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {
            throw new MalformedCarbonCommandException(
              "Unsupported operation on non transactional table")
          } else {
            ExecutedCommandExec(addColumn) :: Nil
          }
        } else {
          throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")
        }
      case dropColumn@CarbonAlterTableDropColumnCommand(alterTableDropColumnModel) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(TableIdentifier(alterTableDropColumnModel.tableName,
            alterTableDropColumnModel.databaseName))(sparkSession)
        if (isCarbonTable) {
          val carbonTable = CarbonEnv.getCarbonTable(alterTableDropColumnModel.databaseName,
            alterTableDropColumnModel.tableName)(sparkSession)
          if (carbonTable != null && carbonTable.isFileLevelFormat) {
            throw new MalformedCarbonCommandException(
              "Unsupported alter operation on Carbon external fileformat table")
          } else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {
            throw new MalformedCarbonCommandException(
              "Unsupported operation on non transactional table")
          } else {
            ExecutedCommandExec(dropColumn) :: Nil
          }
        } else {
          throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")
        }
      case desc@DescribeTableCommand(identifier, partitionSpec, isExtended) =>
        val isFormatted: Boolean = if (sparkSession.version.startsWith("2.1")) {
          CarbonReflectionUtils
            .getDescribeTableFormattedField(desc.asInstanceOf[DescribeTableCommand])
        } else {
          false
        }
        if (CarbonEnv.getInstance(sparkSession).carbonMetastore
              .tableExists(identifier)(sparkSession) && (isExtended || isFormatted)) {
          val resolvedTable =
            sparkSession.sessionState.executePlan(UnresolvedRelation(identifier)).analyzed
          val resultPlan = sparkSession.sessionState.executePlan(resolvedTable).executedPlan
          ExecutedCommandExec(
            CarbonDescribeFormattedCommand(
              resultPlan,
              plan.output,
              partitionSpec,
              identifier)) :: Nil
        } else {
          Nil
        }
      case ShowPartitionsCommand(t, cols) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(t)(sparkSession)
        if (isCarbonTable) {
          val carbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
            .lookupRelation(t)(sparkSession).asInstanceOf[CarbonRelation].carbonTable
          if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {
            throw new MalformedCarbonCommandException(
              "Unsupported operation on non transactional table")
          }
          if (!carbonTable.isHivePartitionTable) {
            ExecutedCommandExec(CarbonShowCarbonPartitionsCommand(t)) :: Nil
          } else {
            ExecutedCommandExec(ShowPartitionsCommand(t, cols)) :: Nil
          }
        } else {
          ExecutedCommandExec(ShowPartitionsCommand(t, cols)) :: Nil
        }
      case adp@AlterTableDropPartitionCommand(tableName, specs, ifExists, purge, retainData) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableName)(sparkSession)
        if (isCarbonTable) {
          ExecutedCommandExec(
            CarbonAlterTableDropHivePartitionCommand(
              tableName,
              specs,
              ifExists,
              purge,
              retainData)) :: Nil
        } else {
          ExecutedCommandExec(adp) :: Nil
        }
      case set@SetCommand(kv) =>
        ExecutedCommandExec(CarbonSetCommand(set)) :: Nil
      case reset@ResetCommand =>
        ExecutedCommandExec(CarbonResetCommand()) :: Nil
      case org.apache.spark.sql.execution.datasources.CreateTable(tableDesc, mode, None)
        if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER
          && (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")
          || tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>
        val updatedCatalog =
          CarbonSource.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession)
        val cmd =
          CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists = mode == SaveMode.Ignore)
        ExecutedCommandExec(cmd) :: Nil
      case cmd@CreateDataSourceTableAsSelectCommand(tableDesc, mode, query)
        if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER
           && (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")
               || tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>
        val updatedCatalog = CarbonSource
          .updateCatalogTableWithCarbonSchema(tableDesc, sparkSession, Option(query))
        val cmd = CreateCarbonSourceTableAsSelectCommand(updatedCatalog, SaveMode.Ignore, query)
        ExecutedCommandExec(cmd) :: Nil
      case cmd@org.apache.spark.sql.execution.datasources.CreateTable(tableDesc, mode, query)
        if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER
           && (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")
               || tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>
        val updatedCatalog = CarbonSource
          .updateCatalogTableWithCarbonSchema(tableDesc, sparkSession, query)
        val cmd = CreateCarbonSourceTableAsSelectCommand(updatedCatalog, SaveMode.Ignore, query.get)
        ExecutedCommandExec(cmd) :: Nil
      case CreateDataSourceTableCommand(table, ignoreIfExists)
        if table.provider.get != DDLUtils.HIVE_PROVIDER
          && (table.provider.get.equals("org.apache.spark.sql.CarbonSource")
          || table.provider.get.equalsIgnoreCase("carbondata")) =>
        val updatedCatalog = CarbonSource
          .updateCatalogTableWithCarbonSchema(table, sparkSession)
        val cmd = CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists)
        ExecutedCommandExec(cmd) :: Nil
      case AlterTableSetPropertiesCommand(tableName, properties, isView)
        if CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableName)(sparkSession) => {

        val carbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .lookupRelation(tableName)(sparkSession).asInstanceOf[CarbonRelation].carbonTable
        if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {
          throw new MalformedCarbonCommandException(
            "Unsupported operation on non transactional table")
        }

        // TODO remove this limitation later
        val property = properties.find(_._1.equalsIgnoreCase("streaming"))
        if (property.isDefined) {
          if (carbonTable.getTablePath.startsWith("s3") && property.get._2.equalsIgnoreCase("s3")) {
            throw new UnsupportedOperationException("streaming is not supported with s3 store")
          }
          if (carbonTable.isStreamingSink) {
            throw new MalformedCarbonCommandException(
              "Streaming property can not be changed once it is 'true'")
          } else {
            if (!property.get._2.trim.equalsIgnoreCase("true")) {
              throw new MalformedCarbonCommandException(
                "Streaming property value is incorrect")
            }
            if (CarbonTable.hasMVDataMap(carbonTable)) {
              throw new MalformedCarbonCommandException(
                "The table which has MV datamap does not support set streaming property")
            }
          }
        }
        ExecutedCommandExec(CarbonAlterTableSetCommand(tableName, properties, isView)) :: Nil
      }
      case AlterTableUnsetPropertiesCommand(tableName, propKeys, ifExists, isView)
        if CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableName)(sparkSession) => {
        // TODO remove this limitation later
        if (propKeys.exists(_.equalsIgnoreCase("streaming"))) {
          throw new MalformedCarbonCommandException(
            "Streaming property can not be removed")
        }
        ExecutedCommandExec(
          CarbonAlterTableUnsetCommand(tableName, propKeys, ifExists, isView)) :: Nil
      }
      case rename@AlterTableRenamePartitionCommand(tableName, oldPartition, newPartition) =>
        val dbOption = tableName.database.map(_.toLowerCase)
        val tableIdentifier = TableIdentifier(tableName.table.toLowerCase(), dbOption)
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableIdentifier)(sparkSession)
        if (isCarbonTable) {
          throw new UnsupportedOperationException("Renaming partition on table is not supported")
        } else {
          ExecutedCommandExec(rename) :: Nil
        }
      case addPart@AlterTableAddPartitionCommand(tableName, partitionSpecsAndLocs, ifNotExists) =>
        val dbOption = tableName.database.map(_.toLowerCase)
        val tableIdentifier = TableIdentifier(tableName.table.toLowerCase(), dbOption)
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableIdentifier)(sparkSession)
        if (isCarbonTable) {
          ExecutedCommandExec(
            CarbonAlterTableAddHivePartitionCommand(
              tableName,
              partitionSpecsAndLocs,
              ifNotExists)) :: Nil
        } else {
          ExecutedCommandExec(addPart) :: Nil
        }
      case RefreshTable(tableIdentifier) =>
        RefreshCarbonTableCommand(tableIdentifier.database,
          tableIdentifier.table).run(sparkSession)
        ExecutedCommandExec(RefreshTable(tableIdentifier)) :: Nil
      case refresh@RefreshResource(path : String) =>
        val plan = try {
          new CarbonSpark2SqlParser().parse(s"REFRESH $path")
        } catch {
          case e: Exception =>
            LOGGER.error(e.getMessage)
            refresh
        }
        ExecutedCommandExec(plan.asInstanceOf[RunnableCommand]) :: Nil
      case alterSetLoc@AlterTableSetLocationCommand(tableName, _, _) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore
          .tableExists(tableName)(sparkSession)
        if (isCarbonTable) {
          throw new UnsupportedOperationException("Set partition location is not supported")
        } else {
          ExecutedCommandExec(alterSetLoc) :: Nil
        }
      case _ => Nil
    }
  }

}

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	package org.apache.spark.sql.execution.strategy
18
19	import org.apache.spark.sql._
20	import org.apache.spark.sql.catalyst.TableIdentifier
21	import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, UnresolvedRelation}
22	import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
23	import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
24	import org.apache.spark.sql.execution.command._
25	import org.apache.spark.sql.execution.command.management.{CarbonAlterTableCompactionCommand, CarbonInsertIntoCommand, CarbonLoadDataCommand, RefreshCarbonTableCommand}
26	import org.apache.spark.sql.execution.command.partition.{CarbonAlterTableAddHivePartitionCommand, CarbonAlterTableDropHivePartitionCommand, CarbonShowCarbonPartitionsCommand}
27	import org.apache.spark.sql.execution.command.schema._
28	import org.apache.spark.sql.execution.command.table.{CarbonDescribeFormattedCommand, CarbonDropTableCommand}
29	import org.apache.spark.sql.hive.execution.command.{CarbonDropDatabaseCommand, CarbonResetCommand, CarbonSetCommand}
30	import org.apache.spark.sql.CarbonExpressions.{CarbonDescribeTable => DescribeTableCommand}
31	import org.apache.spark.sql.execution.datasources.{RefreshResource, RefreshTable}
32	import org.apache.spark.sql.hive.{CarbonRelation, CreateCarbonSourceTableAsSelectCommand}
33	import org.apache.spark.sql.parser.CarbonSpark2SqlParser
34	import org.apache.spark.util.{CarbonReflectionUtils, FileUtils}
35
36	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
37	import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
38	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
39	import org.apache.carbondata.core.util.{CarbonProperties, ThreadLocalSessionInfo}
40
41	/**
42	* Carbon strategies for ddl commands
43	*/
44
45	class DDLStrategy(sparkSession: SparkSession) extends SparkStrategy {	2✔
46	val LOGGER: LogService =	2✔
47	LogServiceFactory.getLogService(this.getClass.getName)	2✔
48	def apply(plan: LogicalPlan): Seq[SparkPlan] = {
49	plan match {	2✔
50	case LoadDataCommand(identifier, path, isLocal, isOverwrite, partition)	2✔
51	if CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
52	.tableExists(identifier)(sparkSession) =>	2✔
53	ExecutedCommandExec(	2✔
54	CarbonLoadDataCommand(	2✔
55	databaseNameOp = identifier.database,	2✔
56	tableName = identifier.table.toLowerCase,	2✔
57	factPathFromUser = path,	2✔
58	dimFilesPath = Seq(),	2✔
59	options = Map(),	2✔
60	isOverwriteTable = isOverwrite,	2✔
61	inputSqlString = null,	2✔
62	dataFrame = None,	2✔
63	updateModel = None,	2✔
64	tableInfoOp = None,	2✔
65	internalOptions = Map.empty,	2✔
66	partition = partition.getOrElse(Map.empty).map { case (col, value) =>	2✔
67	(col, Some(value))})) :: Nil	2✔
68	case alter@AlterTableRenameCommand(oldTableIdentifier, newTableIdentifier, _) =>	2✔
69	val dbOption = oldTableIdentifier.database.map(_.toLowerCase)	2✔
70	val tableIdentifier = TableIdentifier(oldTableIdentifier.table.toLowerCase(), dbOption)	2✔
71	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
72	.tableExists(tableIdentifier)(sparkSession)	2✔
73	if (isCarbonTable) {	2✔
74	val renameModel = AlterTableRenameModel(tableIdentifier, newTableIdentifier)	2✔
75	ExecutedCommandExec(CarbonAlterTableRenameCommand(renameModel)) :: Nil	2✔
76	} else {
77	ExecutedCommandExec(alter) :: Nil	2✔
78	}
79	case DropTableCommand(identifier, ifNotExists, isView, _)	2✔
80	if CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
81	.isTablePathExists(identifier)(sparkSession) =>	2✔
82	ExecutedCommandExec(	2✔
83	CarbonDropTableCommand(ifNotExists, identifier.database,	2✔
84	identifier.table.toLowerCase)) :: Nil	2✔
85	case InsertIntoCarbonTable(relation: CarbonDatasourceHadoopRelation,	2✔
86	partition, child: LogicalPlan, overwrite, _) =>	2✔
87	ExecutedCommandExec(CarbonInsertIntoCommand(relation, child, overwrite, partition)) :: Nil	2✔
88	case createDb@CreateDatabaseCommand(dbName, ifNotExists, _, _, _) =>	2✔
89	val dbLocation = try {	2✔
90	CarbonEnv.getDatabaseLocation(dbName, sparkSession)	2✔
91	} catch {
92	case e: NoSuchDatabaseException =>
93	CarbonProperties.getStorePath	2✔
94	}
95	ThreadLocalSessionInfo
96	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
97	FileUtils.createDatabaseDirectory(dbName, dbLocation, sparkSession.sparkContext)	2✔
98	ExecutedCommandExec(createDb) :: Nil	2✔
99	case drop@DropDatabaseCommand(dbName, ifExists, isCascade) =>	2✔
100	ExecutedCommandExec(CarbonDropDatabaseCommand(drop)) :: Nil	2✔
101	case alterTable@CarbonAlterTableCompactionCommand(altertablemodel, _, _) =>	2✔
102	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
103	.tableExists(TableIdentifier(altertablemodel.tableName,	2✔
104	altertablemodel.dbName))(sparkSession)	2✔
105	if (isCarbonTable) {	2✔
106	ExecutedCommandExec(alterTable) :: Nil	2✔
107	} else {
108	throw new MalformedCarbonCommandException(	×
109	"Operation not allowed : " + altertablemodel.alterSql)	×
110	}
111	case dataTypeChange@CarbonAlterTableDataTypeChangeCommand(alterTableChangeDataTypeModel) =>	2✔
112	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
113	.tableExists(TableIdentifier(alterTableChangeDataTypeModel.tableName,	2✔
114	alterTableChangeDataTypeModel.databaseName))(sparkSession)	2✔
115	if (isCarbonTable) {	2✔
116	val carbonTable = CarbonEnv.getCarbonTable(alterTableChangeDataTypeModel.databaseName,	2✔
117	alterTableChangeDataTypeModel.tableName)(sparkSession)	2✔
118	if (carbonTable != null && carbonTable.isFileLevelFormat) {	2✔
119	throw new MalformedCarbonCommandException(	1✔
120	"Unsupported alter operation on Carbon external fileformat table")	1✔
121	} else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
122	throw new MalformedCarbonCommandException(	1✔
123	"Unsupported operation on non transactional table")	1✔
124	} else {
125	ExecutedCommandExec(dataTypeChange) :: Nil	2✔
126	}
127	} else {
128	throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")	×
129	}
130	case addColumn@CarbonAlterTableAddColumnCommand(alterTableAddColumnsModel) =>	2✔
131	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
132	.tableExists(TableIdentifier(alterTableAddColumnsModel.tableName,	2✔
133	alterTableAddColumnsModel.databaseName))(sparkSession)	2✔
134	if (isCarbonTable) {	2✔
135	val carbonTable = CarbonEnv.getCarbonTable(alterTableAddColumnsModel.databaseName,	2✔
136	alterTableAddColumnsModel.tableName)(sparkSession)	2✔
137	if (carbonTable != null && carbonTable.isFileLevelFormat) {	2✔
138	throw new MalformedCarbonCommandException(	×
139	"Unsupported alter operation on Carbon external fileformat table")	×
140	} else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
141	throw new MalformedCarbonCommandException(	×
142	"Unsupported operation on non transactional table")	×
143	} else {
144	ExecutedCommandExec(addColumn) :: Nil	2✔
145	}
146	} else {
147	throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")	1✔
148	}
149	case dropColumn@CarbonAlterTableDropColumnCommand(alterTableDropColumnModel) =>	2✔
150	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
151	.tableExists(TableIdentifier(alterTableDropColumnModel.tableName,	2✔
152	alterTableDropColumnModel.databaseName))(sparkSession)	2✔
153	if (isCarbonTable) {	2✔
154	val carbonTable = CarbonEnv.getCarbonTable(alterTableDropColumnModel.databaseName,	2✔
155	alterTableDropColumnModel.tableName)(sparkSession)	2✔
156	if (carbonTable != null && carbonTable.isFileLevelFormat) {	2✔
157	throw new MalformedCarbonCommandException(	×
158	"Unsupported alter operation on Carbon external fileformat table")	×
159	} else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
160	throw new MalformedCarbonCommandException(	×
161	"Unsupported operation on non transactional table")	×
162	} else {
163	ExecutedCommandExec(dropColumn) :: Nil	2✔
164	}
165	} else {
166	throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")	1✔
167	}
168	case desc@DescribeTableCommand(identifier, partitionSpec, isExtended) =>	2✔
169	val isFormatted: Boolean = if (sparkSession.version.startsWith("2.1")) {	2✔
170	CarbonReflectionUtils	×
171	.getDescribeTableFormattedField(desc.asInstanceOf[DescribeTableCommand])	×
172	} else {
173	false	2✔
174	}
175	if (CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
176	.tableExists(identifier)(sparkSession) && (isExtended \|\| isFormatted)) {	2✔
177	val resolvedTable =	2✔
178	sparkSession.sessionState.executePlan(UnresolvedRelation(identifier)).analyzed	2✔
179	val resultPlan = sparkSession.sessionState.executePlan(resolvedTable).executedPlan	2✔
180	ExecutedCommandExec(	2✔
181	CarbonDescribeFormattedCommand(	2✔
182	resultPlan,	2✔
183	plan.output,	2✔
184	partitionSpec,	2✔
185	identifier)) :: Nil	2✔
186	} else {
187	Nil	2✔
188	}
189	case ShowPartitionsCommand(t, cols) =>	2✔
190	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
191	.tableExists(t)(sparkSession)	1✔
192	if (isCarbonTable) {	1✔
193	val carbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
194	.lookupRelation(t)(sparkSession).asInstanceOf[CarbonRelation].carbonTable	1✔
195	if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	1✔
196	throw new MalformedCarbonCommandException(	1✔
197	"Unsupported operation on non transactional table")	1✔
198	}
199	if (!carbonTable.isHivePartitionTable) {	1✔
200	ExecutedCommandExec(CarbonShowCarbonPartitionsCommand(t)) :: Nil	1✔
201	} else {
202	ExecutedCommandExec(ShowPartitionsCommand(t, cols)) :: Nil	1✔
203	}
204	} else {
205	ExecutedCommandExec(ShowPartitionsCommand(t, cols)) :: Nil	1✔
206	}
207	case adp@AlterTableDropPartitionCommand(tableName, specs, ifExists, purge, retainData) =>	2✔
208	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
209	.tableExists(tableName)(sparkSession)	1✔
210	if (isCarbonTable) {	1✔
211	ExecutedCommandExec(	1✔
212	CarbonAlterTableDropHivePartitionCommand(	1✔
213	tableName,	1✔
214	specs,	1✔
215	ifExists,	1✔
216	purge,	1✔
217	retainData)) :: Nil	1✔
218	} else {
219	ExecutedCommandExec(adp) :: Nil	×
220	}
221	case set@SetCommand(kv) =>	2✔
222	ExecutedCommandExec(CarbonSetCommand(set)) :: Nil	2✔
223	case reset@ResetCommand =>	2✔
224	ExecutedCommandExec(CarbonResetCommand()) :: Nil	2✔
225	case org.apache.spark.sql.execution.datasources.CreateTable(tableDesc, mode, None)	2✔
226	if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER	×
227	&& (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")	×
228	\|\| tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>	×
229	val updatedCatalog =	×
230	CarbonSource.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession)	×
231	val cmd =	×
232	CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists = mode == SaveMode.Ignore)	×
233	ExecutedCommandExec(cmd) :: Nil	×
234	case cmd@CreateDataSourceTableAsSelectCommand(tableDesc, mode, query)	2✔
235	if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER	2✔
236	&& (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")	2✔
237	\|\| tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>	2✔
238	val updatedCatalog = CarbonSource	1✔
239	.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession, Option(query))	1✔
240	val cmd = CreateCarbonSourceTableAsSelectCommand(updatedCatalog, SaveMode.Ignore, query)	1✔
241	ExecutedCommandExec(cmd) :: Nil	1✔
242	case cmd@org.apache.spark.sql.execution.datasources.CreateTable(tableDesc, mode, query)	×
243	if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER	×
244	&& (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")	×
245	\|\| tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>	×
246	val updatedCatalog = CarbonSource	×
247	.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession, query)	×
248	val cmd = CreateCarbonSourceTableAsSelectCommand(updatedCatalog, SaveMode.Ignore, query.get)	×
249	ExecutedCommandExec(cmd) :: Nil	×
250	case CreateDataSourceTableCommand(table, ignoreIfExists)	2✔
251	if table.provider.get != DDLUtils.HIVE_PROVIDER	2✔
252	&& (table.provider.get.equals("org.apache.spark.sql.CarbonSource")	2✔
253	\|\| table.provider.get.equalsIgnoreCase("carbondata")) =>	2✔
254	val updatedCatalog = CarbonSource	2✔
255	.updateCatalogTableWithCarbonSchema(table, sparkSession)	2✔
256	val cmd = CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists)	2✔
257	ExecutedCommandExec(cmd) :: Nil	2✔
258	case AlterTableSetPropertiesCommand(tableName, properties, isView)	2✔
259	if CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
260	.tableExists(tableName)(sparkSession) => {	2✔
261
262	val carbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
263	.lookupRelation(tableName)(sparkSession).asInstanceOf[CarbonRelation].carbonTable	2✔
264	if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
265	throw new MalformedCarbonCommandException(	×
266	"Unsupported operation on non transactional table")	×
267	}
268
269	// TODO remove this limitation later
270	val property = properties.find(_._1.equalsIgnoreCase("streaming"))	2✔
271	if (property.isDefined) {	2✔
272	if (carbonTable.getTablePath.startsWith("s3") && property.get._2.equalsIgnoreCase("s3")) {	2✔
273	throw new UnsupportedOperationException("streaming is not supported with s3 store")	×
274	}
275	if (carbonTable.isStreamingSink) {	2✔
276	throw new MalformedCarbonCommandException(	2✔
277	"Streaming property can not be changed once it is 'true'")	2✔
278	} else {
279	if (!property.get._2.trim.equalsIgnoreCase("true")) {	2✔
280	throw new MalformedCarbonCommandException(	2✔
281	"Streaming property value is incorrect")	2✔
282	}
283	if (CarbonTable.hasMVDataMap(carbonTable)) {	2✔
284	throw new MalformedCarbonCommandException(	×
285	"The table which has MV datamap does not support set streaming property")	×
286	}
287	}
288	}
289	ExecutedCommandExec(CarbonAlterTableSetCommand(tableName, properties, isView)) :: Nil	2✔
290	}
291	case AlterTableUnsetPropertiesCommand(tableName, propKeys, ifExists, isView)	2✔
292	if CarbonEnv.getInstance(sparkSession).carbonMetastore	2✔
293	.tableExists(tableName)(sparkSession) => {	2✔
294	// TODO remove this limitation later
295	if (propKeys.exists(_.equalsIgnoreCase("streaming"))) {	2✔
296	throw new MalformedCarbonCommandException(	2✔
297	"Streaming property can not be removed")	2✔
298	}
299	ExecutedCommandExec(	1✔
300	CarbonAlterTableUnsetCommand(tableName, propKeys, ifExists, isView)) :: Nil	1✔
301	}
302	case rename@AlterTableRenamePartitionCommand(tableName, oldPartition, newPartition) =>	2✔
303	val dbOption = tableName.database.map(_.toLowerCase)	×
304	val tableIdentifier = TableIdentifier(tableName.table.toLowerCase(), dbOption)	×
305	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	×
306	.tableExists(tableIdentifier)(sparkSession)	×
307	if (isCarbonTable) {	×
308	throw new UnsupportedOperationException("Renaming partition on table is not supported")	×
309	} else {
310	ExecutedCommandExec(rename) :: Nil	×
311	}
312	case addPart@AlterTableAddPartitionCommand(tableName, partitionSpecsAndLocs, ifNotExists) =>	2✔
313	val dbOption = tableName.database.map(_.toLowerCase)	1✔
314	val tableIdentifier = TableIdentifier(tableName.table.toLowerCase(), dbOption)	1✔
315	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
316	.tableExists(tableIdentifier)(sparkSession)	1✔
317	if (isCarbonTable) {	1✔
318	ExecutedCommandExec(	1✔
319	CarbonAlterTableAddHivePartitionCommand(	1✔
320	tableName,	1✔
321	partitionSpecsAndLocs,	1✔
322	ifNotExists)) :: Nil	1✔
323	} else {
324	ExecutedCommandExec(addPart) :: Nil	1✔
325	}
326	case RefreshTable(tableIdentifier) =>	2✔
327	RefreshCarbonTableCommand(tableIdentifier.database,	2✔
328	tableIdentifier.table).run(sparkSession)	2✔
329	ExecutedCommandExec(RefreshTable(tableIdentifier)) :: Nil	2✔
330	case refresh@RefreshResource(path : String) =>	2✔
331	val plan = try {	×
332	new CarbonSpark2SqlParser().parse(s"REFRESH $path")	×
333	} catch {
334	case e: Exception =>
335	LOGGER.error(e.getMessage)	×
336	refresh	×
337	}
338	ExecutedCommandExec(plan.asInstanceOf[RunnableCommand]) :: Nil	×
339	case alterSetLoc@AlterTableSetLocationCommand(tableName, _, _) =>	2✔
340	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetastore	1✔
341	.tableExists(tableName)(sparkSession)	1✔
342	if (isCarbonTable) {	1✔
343	throw new UnsupportedOperationException("Set partition location is not supported")	1✔
344	} else {
345	ExecutedCommandExec(alterSetLoc) :: Nil	×
346	}
347	case _ => Nil	2✔
348	}
349	}
350
351	}

apache / carbondata / 260

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