2054

Build # 2054

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/128e0c7ca5e00c75cca62b03fe8330cb53ec69f3">128e0c7ca<a href="https://github.com/apache/carbondata/commit/128e0c7ca5e00c75cca62b03fe8330cb53ec69f3">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/a20c4b714f9dff1d2661880ad43372ec98dd511e">a20c4b714</a>">a20c4b714</a><a href="https://github.com/apache/carbondata/commit/128e0c7ca5e00c75cca62b03fe8330cb53ec69f3"> into 96cc1a6ab">96cc1a6ab</a>

Run Details

301 of 301 new or added lines in 17 files covered. (100.0%)

61506 of 77844 relevant lines covered (79.01%)

1.05 hits per line

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

76.75

/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 createLikeTable: CreateTableLikeCommand =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore
          .tableExists(createLikeTable.sourceTable)(sparkSession)
        if (isCarbonTable) {
          throw new MalformedCarbonCommandException(
            "Operation not allowed, when source table is carbon table")
        } else {
          ExecutedCommandExec(createLikeTable) :: 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 colRenameDataTypeChange@CarbonAlterTableColRenameDataTypeChangeCommand(
      alterTableColRenameAndDataTypeChangeModel) =>
        val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore
          .tableExists(TableIdentifier(alterTableColRenameAndDataTypeChangeModel.tableName,
            alterTableColRenameAndDataTypeChangeModel.databaseName))(sparkSession)
        if (isCarbonTable) {
          val carbonTable = CarbonEnv
            .getCarbonTable(alterTableColRenameAndDataTypeChangeModel.databaseName,
              alterTableColRenameAndDataTypeChangeModel.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(colRenameDataTypeChange) :: 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 createLikeTable: CreateTableLikeCommand =>	2✔
96	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
97	.tableExists(createLikeTable.sourceTable)(sparkSession)	1✔
98	if (isCarbonTable) {	1✔
99	throw new MalformedCarbonCommandException(	1✔
100	"Operation not allowed, when source table is carbon table")	1✔
101	} else {
102	ExecutedCommandExec(createLikeTable) :: Nil	×
103	}
104	case InsertIntoCarbonTable(relation: CarbonDatasourceHadoopRelation,	2✔
105	partition, child: LogicalPlan, overwrite, _) =>	2✔
106	ExecutedCommandExec(CarbonInsertIntoCommand(relation, child, overwrite, partition)) :: Nil	2✔
107	case createDb@CreateDatabaseCommand(dbName, ifNotExists, _, _, _) =>	2✔
108	val dbLocation = try {	2✔
109	CarbonEnv.getDatabaseLocation(dbName, sparkSession)	2✔
110	} catch {
111	case e: NoSuchDatabaseException =>
112	CarbonProperties.getStorePath	2✔
113	}
114	ThreadLocalSessionInfo
115	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
116	FileUtils.createDatabaseDirectory(dbName, dbLocation, sparkSession.sparkContext)	2✔
117	ExecutedCommandExec(createDb) :: Nil	2✔
118	case drop@DropDatabaseCommand(dbName, ifExists, isCascade) =>	2✔
119	ExecutedCommandExec(CarbonDropDatabaseCommand(drop)) :: Nil	2✔
120	case alterTable@CarbonAlterTableCompactionCommand(altertablemodel, _, _) =>	2✔
121	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
122	.tableExists(TableIdentifier(altertablemodel.tableName,	2✔
123	altertablemodel.dbName))(sparkSession)	2✔
124	if (isCarbonTable) {	2✔
125	ExecutedCommandExec(alterTable) :: Nil	2✔
126	} else {
127	throw new MalformedCarbonCommandException(	×
128	"Operation not allowed : " + altertablemodel.alterSql)	×
129	}
130	case colRenameDataTypeChange@CarbonAlterTableColRenameDataTypeChangeCommand(	2✔
131	alterTableColRenameAndDataTypeChangeModel) =>	2✔
132	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
133	.tableExists(TableIdentifier(alterTableColRenameAndDataTypeChangeModel.tableName,	2✔
134	alterTableColRenameAndDataTypeChangeModel.databaseName))(sparkSession)	2✔
135	if (isCarbonTable) {	2✔
136	val carbonTable = CarbonEnv	2✔
137	.getCarbonTable(alterTableColRenameAndDataTypeChangeModel.databaseName,	2✔
138	alterTableColRenameAndDataTypeChangeModel.tableName)(sparkSession)	2✔
139	if (carbonTable != null && carbonTable.isFileLevelFormat) {	2✔
140	throw new MalformedCarbonCommandException(	1✔
141	"Unsupported alter operation on Carbon external fileformat table")	1✔
142	} else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
143	throw new MalformedCarbonCommandException(	1✔
144	"Unsupported operation on non transactional table")	1✔
145	} else {
146	ExecutedCommandExec(colRenameDataTypeChange) :: Nil	2✔
147	}
148	} else {
149	throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")	×
150	}
151	case addColumn@CarbonAlterTableAddColumnCommand(alterTableAddColumnsModel) =>	2✔
152	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
153	.tableExists(TableIdentifier(alterTableAddColumnsModel.tableName,	2✔
154	alterTableAddColumnsModel.databaseName))(sparkSession)	2✔
155	if (isCarbonTable) {	2✔
156	val carbonTable = CarbonEnv.getCarbonTable(alterTableAddColumnsModel.databaseName,	2✔
157	alterTableAddColumnsModel.tableName)(sparkSession)	2✔
158	if (carbonTable != null && carbonTable.isFileLevelFormat) {	2✔
159	throw new MalformedCarbonCommandException(	×
160	"Unsupported alter operation on Carbon external fileformat table")	×
161	} else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
162	throw new MalformedCarbonCommandException(	×
163	"Unsupported operation on non transactional table")	×
164	} else {
165	ExecutedCommandExec(addColumn) :: Nil	2✔
166	}
167	// TODO: remove this else if check once the 2.1 version is unsupported by carbon
168	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	×
169	val structField = (alterTableAddColumnsModel.dimCols ++ alterTableAddColumnsModel.msrCols)	×
170	.map {	×
171	a =>	×
172	StructField(a.column,	×
173	Util.convertCarbonToSparkDataType(DataTypeUtil.valueOf(a.dataType.get)))	×
174	}
175	val identifier = TableIdentifier(	×
176	alterTableAddColumnsModel.tableName,	×
177	alterTableAddColumnsModel.databaseName)	×
178	ExecutedCommandExec(CarbonReflectionUtils	×
179	.invokeAlterTableAddColumn(identifier, structField).asInstanceOf[RunnableCommand]) ::	×
180	Nil	×
181	// TODO: remove this else check once the 2.1 version is unsupported by carbon
182	} else {
183	throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")	×
184	}
185	case dropColumn@CarbonAlterTableDropColumnCommand(alterTableDropColumnModel) =>	2✔
186	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
187	.tableExists(TableIdentifier(alterTableDropColumnModel.tableName,	2✔
188	alterTableDropColumnModel.databaseName))(sparkSession)	2✔
189	if (isCarbonTable) {	2✔
190	val carbonTable = CarbonEnv.getCarbonTable(alterTableDropColumnModel.databaseName,	2✔
191	alterTableDropColumnModel.tableName)(sparkSession)	2✔
192	if (carbonTable != null && carbonTable.isFileLevelFormat) {	2✔
193	throw new MalformedCarbonCommandException(	×
194	"Unsupported alter operation on Carbon external fileformat table")	×
195	} else if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
196	throw new MalformedCarbonCommandException(	×
197	"Unsupported operation on non transactional table")	×
198	} else {
199	ExecutedCommandExec(dropColumn) :: Nil	2✔
200	}
201	} else {
202	throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")	×
203	}
204	case desc@DescribeTableCommand(identifier, partitionSpec, isExtended) =>	2✔
205	val isFormatted: Boolean = if (sparkSession.version.startsWith("2.1")) {	2✔
206	CarbonReflectionUtils	×
207	.getDescribeTableFormattedField(desc.asInstanceOf[DescribeTableCommand])	×
208	} else {
209	false	2✔
210	}
211	if (CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
212	.tableExists(identifier)(sparkSession) && (isExtended \|\| isFormatted)) {	2✔
213	val resolvedTable =	2✔
214	sparkSession.sessionState.executePlan(UnresolvedRelation(identifier)).analyzed	2✔
215	val resultPlan = sparkSession.sessionState.executePlan(resolvedTable).executedPlan	2✔
216	ExecutedCommandExec(	2✔
217	CarbonDescribeFormattedCommand(	2✔
218	resultPlan,	2✔
219	plan.output,	2✔
220	partitionSpec,	2✔
221	identifier)) :: Nil	2✔
222	} else {
223	Nil	2✔
224	}
225	case ShowPartitionsCommand(t, cols) =>	2✔
226	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
227	.tableExists(t)(sparkSession)	1✔
228	if (isCarbonTable) {	1✔
229	val carbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
230	.lookupRelation(t)(sparkSession).asInstanceOf[CarbonRelation].carbonTable	1✔
231	if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	1✔
232	throw new MalformedCarbonCommandException(	1✔
233	"Unsupported operation on non transactional table")	1✔
234	}
235	if (!carbonTable.isHivePartitionTable) {	1✔
236	ExecutedCommandExec(CarbonShowCarbonPartitionsCommand(t)) :: Nil	1✔
237	} else {
238	ExecutedCommandExec(ShowPartitionsCommand(t, cols)) :: Nil	1✔
239	}
240	} else {
241	ExecutedCommandExec(ShowPartitionsCommand(t, cols)) :: Nil	1✔
242	}
243	case adp@AlterTableDropPartitionCommand(tableName, specs, ifExists, purge, retainData) =>	2✔
244	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
245	.tableExists(tableName)(sparkSession)	1✔
246	if (isCarbonTable) {	1✔
247	ExecutedCommandExec(	1✔
248	CarbonAlterTableDropHivePartitionCommand(	1✔
249	tableName,	1✔
250	specs,	1✔
251	ifExists,	1✔
252	purge,	1✔
253	retainData)) :: Nil	1✔
254	} else {
255	ExecutedCommandExec(adp) :: Nil	×
256	}
257	case set@SetCommand(kv) =>	2✔
258	ExecutedCommandExec(CarbonSetCommand(set)) :: Nil	2✔
259	case reset@ResetCommand =>	2✔
260	ExecutedCommandExec(CarbonResetCommand()) :: Nil	2✔
261	case org.apache.spark.sql.execution.datasources.CreateTable(tableDesc, mode, None)	2✔
262	if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER	×
263	&& (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")	×
264	\|\| tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>	×
265	val updatedCatalog =	×
266	CarbonSource.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession)	×
267	val cmd =	×
268	CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists = mode == SaveMode.Ignore)	×
269	ExecutedCommandExec(cmd) :: Nil	×
270	case MatchCreateDataSourceTable(tableDesc, mode, query)	2✔
271	if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER	2✔
272	&& (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")	2✔
273	\|\| tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>	2✔
274	val updatedCatalog = CarbonSource	1✔
275	.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession, Option(query))	1✔
276	val cmd = CreateCarbonSourceTableAsSelectCommand(updatedCatalog, SaveMode.Ignore, query)	1✔
277	ExecutedCommandExec(cmd) :: Nil	1✔
278	case cmd@org.apache.spark.sql.execution.datasources.CreateTable(tableDesc, mode, query)	×
279	if tableDesc.provider.get != DDLUtils.HIVE_PROVIDER	×
280	&& (tableDesc.provider.get.equals("org.apache.spark.sql.CarbonSource")	×
281	\|\| tableDesc.provider.get.equalsIgnoreCase("carbondata")) =>	×
282	val updatedCatalog = CarbonSource	×
283	.updateCatalogTableWithCarbonSchema(tableDesc, sparkSession, query)	×
284	val cmd = CreateCarbonSourceTableAsSelectCommand(updatedCatalog, SaveMode.Ignore, query.get)	×
285	ExecutedCommandExec(cmd) :: Nil	×
286	case CreateDataSourceTableCommand(table, ignoreIfExists)	2✔
287	if table.provider.get != DDLUtils.HIVE_PROVIDER	2✔
288	&& (table.provider.get.equals("org.apache.spark.sql.CarbonSource")	2✔
289	\|\| table.provider.get.equalsIgnoreCase("carbondata")) =>	2✔
290	val updatedCatalog = CarbonSource	2✔
291	.updateCatalogTableWithCarbonSchema(table, sparkSession)	2✔
292	val cmd = CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists)	2✔
293	ExecutedCommandExec(cmd) :: Nil	2✔
294	case AlterTableSetPropertiesCommand(tableName, properties, isView)	2✔
295	if CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
296	.tableExists(tableName)(sparkSession) => {	2✔
297
298	val carbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
299	.lookupRelation(tableName)(sparkSession).asInstanceOf[CarbonRelation].carbonTable	2✔
300	if (carbonTable != null && !carbonTable.getTableInfo.isTransactionalTable) {	2✔
301	throw new MalformedCarbonCommandException(	×
302	"Unsupported operation on non transactional table")	×
303	}
304
305	// TODO remove this limitation later
306	val property = properties.find(_._1.equalsIgnoreCase("streaming"))	2✔
307	if (property.isDefined) {	2✔
308	if (carbonTable.getTablePath.startsWith("s3") && property.get._2.equalsIgnoreCase("s3")) {	2✔
309	throw new UnsupportedOperationException("streaming is not supported with s3 store")	×
310	}
311	if (carbonTable.isStreamingSink) {	2✔
312	throw new MalformedCarbonCommandException(	2✔
313	"Streaming property can not be changed once it is 'true'")	2✔
314	} else {
315	if (!property.get._2.trim.equalsIgnoreCase("true")) {	2✔
316	throw new MalformedCarbonCommandException(	2✔
317	"Streaming property value is incorrect")	2✔
318	}
319	if (CarbonTable.hasMVDataMap(carbonTable)) {	2✔
320	throw new MalformedCarbonCommandException(	×
321	"The table which has MV datamap does not support set streaming property")	×
322	}
323	}
324	}
325	ExecutedCommandExec(CarbonAlterTableSetCommand(tableName, properties, isView)) :: Nil	2✔
326	}
327	case AlterTableUnsetPropertiesCommand(tableName, propKeys, ifExists, isView)	2✔
328	if CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
329	.tableExists(tableName)(sparkSession) => {	2✔
330	// TODO remove this limitation later
331	if (propKeys.exists(_.equalsIgnoreCase("streaming"))) {	2✔
332	throw new MalformedCarbonCommandException(	2✔
333	"Streaming property can not be removed")	2✔
334	}
335	ExecutedCommandExec(	1✔
336	CarbonAlterTableUnsetCommand(tableName, propKeys, ifExists, isView)) :: Nil	1✔
337	}
338	case rename@AlterTableRenamePartitionCommand(tableName, oldPartition, newPartition) =>	2✔
339	val dbOption = tableName.database.map(_.toLowerCase)	×
340	val tableIdentifier = TableIdentifier(tableName.table.toLowerCase(), dbOption)	×
341	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	×
342	.tableExists(tableIdentifier)(sparkSession)	×
343	if (isCarbonTable) {	×
344	throw new UnsupportedOperationException("Renaming partition on table is not supported")	×
345	} else {
346	ExecutedCommandExec(rename) :: Nil	×
347	}
348	case addPart@AlterTableAddPartitionCommand(tableName, partitionSpecsAndLocs, ifNotExists) =>	2✔
349	val dbOption = tableName.database.map(_.toLowerCase)	1✔
350	val tableIdentifier = TableIdentifier(tableName.table.toLowerCase(), dbOption)	1✔
351	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
352	.tableExists(tableIdentifier)(sparkSession)	1✔
353	if (isCarbonTable) {	1✔
354	ExecutedCommandExec(	1✔
355	CarbonAlterTableAddHivePartitionCommand(	1✔
356	tableName,	1✔
357	partitionSpecsAndLocs,	1✔
358	ifNotExists)) :: Nil	1✔
359	} else {
360	ExecutedCommandExec(addPart) :: Nil	1✔
361	}
362	case RefreshTable(tableIdentifier) =>	2✔
363	RefreshCarbonTableCommand(tableIdentifier.database,	2✔
364	tableIdentifier.table).run(sparkSession)	2✔
365	ExecutedCommandExec(RefreshTable(tableIdentifier)) :: Nil	2✔
366	case refresh@RefreshResource(path : String) =>	2✔
367	val plan = try {	×
368	new CarbonSpark2SqlParser().parse(s"REFRESH $path")	×
369	} catch {
370	case e: Exception =>
371	LOGGER.error(e.getMessage)	×
372	refresh	×
373	}
374	ExecutedCommandExec(plan.asInstanceOf[RunnableCommand]) :: Nil	×
375	case alterSetLoc@AlterTableSetLocationCommand(tableName, _, _) =>	2✔
376	val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore	1✔
377	.tableExists(tableName)(sparkSession)	1✔
378	if (isCarbonTable) {	1✔
379	throw new UnsupportedOperationException("Set partition location is not supported")	1✔
380	} else {
381	ExecutedCommandExec(alterSetLoc) :: Nil	×
382	}
383	case _ => Nil	2✔
384	}
385	}
386
387	}

apache / carbondata / 2054

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