1136

Build # 1136

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/92231af871268d9101c0bc16dd1eb144b92c0bdc">92231af87<a href="https://github.com/apache/carbondata/commit/92231af871268d9101c0bc16dd1eb144b92c0bdc">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9f752b74081806d5d4f5aece1558b968006b8c7c">9f752b740</a>">9f752b740</a><a href="https://github.com/apache/carbondata/commit/92231af871268d9101c0bc16dd1eb144b92c0bdc"> into 8c49e5b42">8c49e5b42</a>

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

60555 of 75153 relevant lines covered (80.58%)

1.06 hits per line

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

76.6

/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.catalog.CatalogTable
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.sql.types.StructField
import org.apache.spark.util.{CarbonReflectionUtils, FileUtils, SparkUtil}

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, DataTypeUtil, ThreadLocalSessionInfo}
import org.apache.carbondata.spark.util.Util

  /**
   * Carbon strategies for ddl commands
   * CreateDataSourceTableAsSelectCommand class has extra argument in
   * 2.3, so need to add wrapper to match the case
   */
object MatchCreateDataSourceTable {
  def unapply(plan: LogicalPlan): Option[(CatalogTable, SaveMode, LogicalPlan)] = plan match {
    case t: CreateDataSourceTableAsSelectCommand => Some(t.table, t.mode, t.query)
    case _ => None
  }
}

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

apache / carbondata / 1136

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