11586

Build # 11586

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/82f8ba29d3d408e01a00e1f7d0339a54e8da1c49">82f8ba29d<a href="https://github.com/apache/carbondata/commit/82f8ba29d3d408e01a00e1f7d0339a54e8da1c49">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/fd1f893c811ccf726bd5e779900071661d3d659b">fd1f893c8</a>">fd1f893c8</a><a href="https://github.com/apache/carbondata/commit/82f8ba29d3d408e01a00e1f7d0339a54e8da1c49"> into df71291ec">df71291ec</a>

Run Details

18 of 18 new or added lines in 2 files covered. (100.0%)

64108 of 81901 relevant lines covered (78.27%)

1.04 hits per line

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

83.78

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

import scala.collection.JavaConverters._
import scala.collection.mutable

import org.apache.spark.SPARK_VERSION
import org.apache.spark.sql._
import org.apache.spark.sql.CarbonExpressions.CarbonUnresolvedRelation
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedStar}
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, NamedExpression}
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.execution.command.mutation.CarbonProjectForDeleteCommand
import org.apache.spark.sql.execution.datasources.{CatalogFileIndex, FileFormat, HadoopFsRelation, LogicalRelation, SparkCarbonTableFormat}
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CarbonException
import org.apache.spark.util.{CarbonReflectionUtils, SparkUtil}

import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.{DataMapStoreManager, DataMapUtil}
import org.apache.carbondata.core.datamap.status.DataMapStatusManager
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonUtil

case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {

  private lazy val parser = sparkSession.sessionState.sqlParser

  private def processUpdateQuery(
      table: UnresolvedRelation,
      columns: List[String],
      selectStmt: String,
      alias: Option[String],
      filter: String): LogicalPlan = {
    var includedDestColumns = false
    var includedDestRelation = false
    var addedTupleId = false

    def prepareTargetReleation(relation: UnresolvedRelation): SubqueryAlias = {
      val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",
        Seq.empty, isDistinct = false), "tupleId")())

      val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
      val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)
      if (carbonTable != null) {
        if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for tables which have a pre-aggregate table. " +
            "Drop pre-aggregate tables to continue.")
        }
        if (carbonTable.isChildDataMap) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for pre-aggregate table")
        }
        val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
        if (CarbonTable.hasMVDataMap(carbonTable)) {
          val allDataMapSchemas = DataMapStoreManager.getInstance
            .getDataMapSchemasOfTable(carbonTable).asScala
            .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
                                     !dataMapSchema.isIndexDataMap).asJava
          allDataMapSchemas.asScala.foreach { dataMapSchema =>
            DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)
          }
        } else if (!indexSchemas.isEmpty) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for table which has index datamaps")
        }
        if (carbonTable.isChildTable) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for mv datamap table")
        }
      }
      val tableRelation = if (SparkUtil.isSparkVersionEqualTo("2.1")) {
        relation
      } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
        alias match {
          case Some(_) =>
            CarbonReflectionUtils.getSubqueryAlias(
              sparkSession,
              alias,
              relation,
              Some(table.tableIdentifier))
          case _ => relation
        }
      } else {
        throw new UnsupportedOperationException("Unsupported Spark version.")
      }

      CarbonReflectionUtils.getSubqueryAlias(
        sparkSession,
        alias,
        Project(projList, tableRelation),
        Some(table.tableIdentifier))
    }

    // get the un-analyzed logical plan
    val targetTable = prepareTargetReleation(table)
    val selectPlan = parser.parsePlan(selectStmt) transform {
      case Project(projectList, child) if !includedDestColumns =>
        includedDestColumns = true
        if (projectList.size != columns.size) {
          CarbonException.analysisException(
            "The number of columns in source table and destination table columns mismatch")
        }
        val renamedProjectList = projectList.zip(columns).map { case (attr, col) =>
          attr match {
            case UnresolvedAlias(child22, _) =>
              UnresolvedAlias(Alias(child22, col + "-updatedColumn")())
            case UnresolvedAttribute(_) =>
              UnresolvedAlias(Alias(attr, col + "-updatedColumn")())
            case _ => attr
          }
        }
        val tableName: Option[Seq[String]] = alias match {
          case Some(_) => Some(alias.toSeq)
          case _ => Some(Seq(child.asInstanceOf[UnresolvedRelation].tableIdentifier.table.toString))
        }
        val list = Seq(
          UnresolvedAlias(UnresolvedStar(tableName))) ++ renamedProjectList
        Project(list, child)
      case Filter(cond, child) if !includedDestRelation =>
        includedDestRelation = true
        Filter(cond, Join(child, targetTable, Inner, None))
      case r@CarbonUnresolvedRelation(t) if !includedDestRelation && t != table.tableIdentifier =>
        includedDestRelation = true
        Join(r, targetTable, Inner, None)
    }
    val updatedSelectPlan: LogicalPlan = if (!includedDestRelation) {
      // special case to handle self join queries
      // Eg. update tableName  SET (column1) = (column1+1)
      selectPlan transform {
        case relation: UnresolvedRelation
          if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>
          addedTupleId = true
          targetTable
      }
    } else {
      selectPlan
    }
    val finalPlan = if (filter.length > 0) {
      var transformed: Boolean = false
      // Create a dummy projection to include filter conditions
      var newPlan: LogicalPlan = null
      if (table.tableIdentifier.database.isDefined) {
        newPlan = parser.parsePlan("select * from  " +
           table.tableIdentifier.database.getOrElse("") + "." +
           table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)
      }
      else {
        newPlan = parser.parsePlan("select * from  " +
           table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)
      }
      newPlan transform {
        case CarbonUnresolvedRelation(t)
          if !transformed && t == table.tableIdentifier =>
          transformed = true

          CarbonReflectionUtils.getSubqueryAlias(
            sparkSession,
            alias,
            updatedSelectPlan,
            Some(table.tableIdentifier))
      }
    } else {
      updatedSelectPlan
    }
    val destinationTable = CarbonReflectionUtils.getUnresolvedRelation(table.tableIdentifier, alias)

    ProjectForUpdate(destinationTable, columns, Seq(finalPlan))
  }


  def processDeleteRecordsQuery(selectStmt: String,
      alias: Option[String],
      table: UnresolvedRelation): LogicalPlan = {
    var addedTupleId = false
    val parsePlan = parser.parsePlan(selectStmt)

    val selectPlan = parsePlan transform {
      case relation: UnresolvedRelation
        if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>
        addedTupleId = true
        val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",
          Seq.empty, isDistinct = false), "tupleId")())

        val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
        val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)
        if (carbonTable != null) {
          if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
            throw new UnsupportedOperationException(
              "Delete operation is not supported for tables which have a pre-aggregate table. " +
              "Drop pre-aggregate tables to continue.")
          }
          if (carbonTable.isChildDataMap) {
            throw new UnsupportedOperationException(
              "Delete operation is not supported for pre-aggregate table")
          }
          val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
          if (CarbonTable.hasMVDataMap(carbonTable)) {
            val allDataMapSchemas = DataMapStoreManager.getInstance
              .getDataMapSchemasOfTable(carbonTable).asScala
              .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
                                       !dataMapSchema.isIndexDataMap).asJava
            allDataMapSchemas.asScala.foreach { dataMapSchema =>
              DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)
            }
          } else if (!indexSchemas.isEmpty) {
            throw new UnsupportedOperationException(
              "Delete operation is not supported for table which has index datamaps")
          }
        }
        // include tuple id in subquery
        if (SparkUtil.isSparkVersionEqualTo("2.1")) {
          Project(projList, relation)
        } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
          alias match {
            case Some(_) =>
              val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(
                sparkSession,
                alias,
                relation,
                Some(table.tableIdentifier))
              Project(projList, subqueryAlias)
            case _ => Project(projList, relation)
          }
        } else {
          throw new UnsupportedOperationException("Unsupported Spark version.")
        }
    }
    CarbonProjectForDeleteCommand(
      selectPlan,
      table.tableIdentifier.database,
      table.tableIdentifier.table,
      System.currentTimeMillis().toString)
  }

  override def apply(logicalplan: LogicalPlan): LogicalPlan = {

    logicalplan transform {
      case UpdateTable(t, cols, sel, alias, where) => processUpdateQuery(t, cols, sel, alias, where)
      case DeleteRecords(statement, alias, table) =>
        processDeleteRecordsQuery(
          statement,
          alias,
          table)
    }
  }
}

/**
 * Insert into carbon table from other source
 */
case class CarbonPreInsertionCasts(sparkSession: SparkSession) extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = {
    plan.transform {
      // Wait until children are resolved.
      case p: LogicalPlan if !p.childrenResolved => p

      case p@InsertIntoTable(relation: LogicalRelation, _, child, _, _)
        if relation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
        castChildOutput(p, relation, child)
    }
  }

  def castChildOutput(p: InsertIntoTable,
      relation: LogicalRelation,
      child: LogicalPlan): LogicalPlan = {
    val carbonDSRelation = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    if (carbonDSRelation.carbonRelation.output.size > CarbonCommonConstants
      .DEFAULT_MAX_NUMBER_OF_COLUMNS) {
      CarbonException.analysisException(
        s"Maximum number of columns supported: " +
          s"${CarbonCommonConstants.DEFAULT_MAX_NUMBER_OF_COLUMNS}")
    }
    if (child.output.size >= carbonDSRelation.carbonRelation.output.size ||
        carbonDSRelation.carbonTable.isHivePartitionTable) {
      val newChildOutput = child.output.zipWithIndex.map { columnWithIndex =>
        columnWithIndex._1 match {
          case attr: Attribute =>
            Alias(attr, s"col${ columnWithIndex._2 }")(NamedExpression.newExprId)
          case attr => attr
        }
      }
      val newChild: LogicalPlan = if (newChildOutput == child.output) {
        if (SparkUtil.isSparkVersionEqualTo("2.1")) {
          CarbonReflectionUtils.getField("child", p).asInstanceOf[LogicalPlan]
        } else if (SparkUtil.isSparkVersionEqualTo("2.2")) {
          CarbonReflectionUtils.getField("query", p).asInstanceOf[LogicalPlan]
        } else {
          throw new UnsupportedOperationException(s"Spark version $SPARK_VERSION is not supported")
        }
      } else {
        Project(newChildOutput, child)
      }

      val overwrite = CarbonReflectionUtils.getOverWriteOption("overwrite", p)

      InsertIntoCarbonTable(carbonDSRelation, p.partition, newChild, overwrite, true)
    } else {
      CarbonException.analysisException(
        "Cannot insert into target table because number of columns mismatch")
    }
  }
}


1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.hive
19
20	import scala.collection.JavaConverters._
21	import scala.collection.mutable
22
23	import org.apache.spark.SPARK_VERSION
24	import org.apache.spark.sql._
25	import org.apache.spark.sql.CarbonExpressions.CarbonUnresolvedRelation
26	import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedStar}
27	import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, NamedExpression}
28	import org.apache.spark.sql.catalyst.plans.Inner
29	import org.apache.spark.sql.catalyst.plans.logical._
30	import org.apache.spark.sql.catalyst.rules._
31	import org.apache.spark.sql.execution.command.mutation.CarbonProjectForDeleteCommand
32	import org.apache.spark.sql.execution.datasources.{CatalogFileIndex, FileFormat, HadoopFsRelation, LogicalRelation, SparkCarbonTableFormat}
33	import org.apache.spark.sql.types.StructType
34	import org.apache.spark.sql.util.CarbonException
35	import org.apache.spark.util.{CarbonReflectionUtils, SparkUtil}
36
37	import org.apache.carbondata.core.constants.CarbonCommonConstants
38	import org.apache.carbondata.core.datamap.{DataMapStoreManager, DataMapUtil}
39	import org.apache.carbondata.core.datamap.status.DataMapStatusManager
40	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
41	import org.apache.carbondata.core.util.CarbonUtil
42
43	case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {	2✔
44
45	private lazy val parser = sparkSession.sessionState.sqlParser	2✔
46
47	private def processUpdateQuery(
48	table: UnresolvedRelation,
49	columns: List[String],
50	selectStmt: String,
51	alias: Option[String],
52	filter: String): LogicalPlan = {
53	var includedDestColumns = false	2✔
54	var includedDestRelation = false	2✔
55	var addedTupleId = false	2✔
56
57	def prepareTargetReleation(relation: UnresolvedRelation): SubqueryAlias = {
58	val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",	2✔
59	Seq.empty, isDistinct = false), "tupleId")())	2✔
60
61	val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)	2✔
62	val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)	2✔
63	if (carbonTable != null) {	2✔
64	if (CarbonUtil.hasAggregationDataMap(carbonTable)) {	2✔
65	throw new UnsupportedOperationException(	1✔
66	"Update operation is not supported for tables which have a pre-aggregate table. " +	1✔
67	"Drop pre-aggregate tables to continue.")
68	}
69	if (carbonTable.isChildDataMap) {	2✔
70	throw new UnsupportedOperationException(	1✔
71	"Update operation is not supported for pre-aggregate table")	1✔
72	}
73	val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)	2✔
74	if (CarbonTable.hasMVDataMap(carbonTable)) {	2✔
75	val allDataMapSchemas = DataMapStoreManager.getInstance	×
76	.getDataMapSchemasOfTable(carbonTable).asScala	×
77	.filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&	×
78	!dataMapSchema.isIndexDataMap).asJava	×
79	allDataMapSchemas.asScala.foreach { dataMapSchema =>	×
80	DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)	×
81	}
82	} else if (!indexSchemas.isEmpty) {	2✔
83	throw new UnsupportedOperationException(	1✔
84	"Update operation is not supported for table which has index datamaps")	1✔
85	}
86	if (carbonTable.isChildTable) {	2✔
87	throw new UnsupportedOperationException(	×
88	"Update operation is not supported for mv datamap table")	×
89	}
90	}
91	val tableRelation = if (SparkUtil.isSparkVersionEqualTo("2.1")) {	2✔
92	relation	×
93	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	2✔
94	alias match {	2✔
95	case Some(_) =>	2✔
96	CarbonReflectionUtils.getSubqueryAlias(	2✔
97	sparkSession,	2✔
98	alias,	2✔
99	relation,	2✔
100	Some(table.tableIdentifier))	2✔
101	case _ => relation	×
102	}
103	} else {
104	throw new UnsupportedOperationException("Unsupported Spark version.")	×
105	}
106
107	CarbonReflectionUtils.getSubqueryAlias(	2✔
108	sparkSession,	2✔
109	alias,	2✔
110	Project(projList, tableRelation),	2✔
111	Some(table.tableIdentifier))	2✔
112	}
113
114	// get the un-analyzed logical plan
115	val targetTable = prepareTargetReleation(table)	2✔
116	val selectPlan = parser.parsePlan(selectStmt) transform {	2✔
117	case Project(projectList, child) if !includedDestColumns =>	2✔
118	includedDestColumns = true	2✔
119	if (projectList.size != columns.size) {	2✔
120	CarbonException.analysisException(	1✔
121	"The number of columns in source table and destination table columns mismatch")	1✔
122	}
123	val renamedProjectList = projectList.zip(columns).map { case (attr, col) =>	2✔
124	attr match {	2✔
125	case UnresolvedAlias(child22, _) =>	2✔
126	UnresolvedAlias(Alias(child22, col + "-updatedColumn")())	2✔
127	case UnresolvedAttribute(_) =>	2✔
128	UnresolvedAlias(Alias(attr, col + "-updatedColumn")())	2✔
129	case _ => attr	×
130	}
131	}
132	val tableName: Option[Seq[String]] = alias match {	2✔
133	case Some(_) => Some(alias.toSeq)	2✔
134	case _ => Some(Seq(child.asInstanceOf[UnresolvedRelation].tableIdentifier.table.toString))	×
135	}
136	val list = Seq(	2✔
137	UnresolvedAlias(UnresolvedStar(tableName))) ++ renamedProjectList	2✔
138	Project(list, child)	2✔
139	case Filter(cond, child) if !includedDestRelation =>	2✔
140	includedDestRelation = true	2✔
141	Filter(cond, Join(child, targetTable, Inner, None))	2✔
142	case r@CarbonUnresolvedRelation(t) if !includedDestRelation && t != table.tableIdentifier =>	2✔
143	includedDestRelation = true	1✔
144	Join(r, targetTable, Inner, None)	1✔
145	}
146	val updatedSelectPlan: LogicalPlan = if (!includedDestRelation) {	2✔
147	// special case to handle self join queries
148	// Eg. update tableName SET (column1) = (column1+1)
149	selectPlan transform {	2✔
150	case relation: UnresolvedRelation	2✔
151	if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>	2✔
152	addedTupleId = true	2✔
153	targetTable	2✔
154	}
155	} else {
156	selectPlan	2✔
157	}
158	val finalPlan = if (filter.length > 0) {	2✔
159	var transformed: Boolean = false	2✔
160	// Create a dummy projection to include filter conditions
161	var newPlan: LogicalPlan = null	2✔
162	if (table.tableIdentifier.database.isDefined) {	2✔
163	newPlan = parser.parsePlan("select * from " +	2✔
164	table.tableIdentifier.database.getOrElse("") + "." +	2✔
165	table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)	2✔
166	}
167	else {
168	newPlan = parser.parsePlan("select * from " +	2✔
169	table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)	2✔
170	}
171	newPlan transform {	2✔
172	case CarbonUnresolvedRelation(t)	2✔
173	if !transformed && t == table.tableIdentifier =>	2✔
174	transformed = true	2✔
175
176	CarbonReflectionUtils.getSubqueryAlias(	2✔
177	sparkSession,	2✔
178	alias,	2✔
179	updatedSelectPlan,	2✔
180	Some(table.tableIdentifier))	2✔
181	}
182	} else {
183	updatedSelectPlan	2✔
184	}
185	val destinationTable = CarbonReflectionUtils.getUnresolvedRelation(table.tableIdentifier, alias)	2✔
186
187	ProjectForUpdate(destinationTable, columns, Seq(finalPlan))	2✔
188	}
189
190
191	def processDeleteRecordsQuery(selectStmt: String,
192	alias: Option[String],
193	table: UnresolvedRelation): LogicalPlan = {
194	var addedTupleId = false	2✔
195	val parsePlan = parser.parsePlan(selectStmt)	2✔
196
197	val selectPlan = parsePlan transform {	2✔
198	case relation: UnresolvedRelation	2✔
199	if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>	2✔
200	addedTupleId = true	2✔
201	val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",	2✔
202	Seq.empty, isDistinct = false), "tupleId")())	2✔
203
204	val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)	2✔
205	val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)	2✔
206	if (carbonTable != null) {	2✔
207	if (CarbonUtil.hasAggregationDataMap(carbonTable)) {	2✔
208	throw new UnsupportedOperationException(	1✔
209	"Delete operation is not supported for tables which have a pre-aggregate table. " +	1✔
210	"Drop pre-aggregate tables to continue.")
211	}
212	if (carbonTable.isChildDataMap) {	2✔
213	throw new UnsupportedOperationException(	1✔
214	"Delete operation is not supported for pre-aggregate table")	1✔
215	}
216	val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)	2✔
217	if (CarbonTable.hasMVDataMap(carbonTable)) {	2✔
218	val allDataMapSchemas = DataMapStoreManager.getInstance	×
219	.getDataMapSchemasOfTable(carbonTable).asScala	×
220	.filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&	×
221	!dataMapSchema.isIndexDataMap).asJava	×
222	allDataMapSchemas.asScala.foreach { dataMapSchema =>	×
223	DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)	×
224	}
225	} else if (!indexSchemas.isEmpty) {	2✔
226	throw new UnsupportedOperationException(	1✔
227	"Delete operation is not supported for table which has index datamaps")	1✔
228	}
229	}
230	// include tuple id in subquery
231	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	2✔
232	Project(projList, relation)	×
233	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	2✔
234	alias match {	2✔
235	case Some(_) =>	2✔
236	val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(	1✔
237	sparkSession,	1✔
238	alias,	1✔
239	relation,	1✔
240	Some(table.tableIdentifier))	1✔
241	Project(projList, subqueryAlias)	1✔
242	case _ => Project(projList, relation)	2✔
243	}
244	} else {
245	throw new UnsupportedOperationException("Unsupported Spark version.")	×
246	}
247	}
248	CarbonProjectForDeleteCommand(	2✔
249	selectPlan,	2✔
250	table.tableIdentifier.database,	2✔
251	table.tableIdentifier.table,	2✔
252	System.currentTimeMillis().toString)	2✔
253	}
254
255	override def apply(logicalplan: LogicalPlan): LogicalPlan = {
256
257	logicalplan transform {	2✔
258	case UpdateTable(t, cols, sel, alias, where) => processUpdateQuery(t, cols, sel, alias, where)	2✔
259	case DeleteRecords(statement, alias, table) =>	2✔
260	processDeleteRecordsQuery(	2✔
261	statement,	2✔
262	alias,	2✔
263	table)	2✔
264	}
265	}
266	}
267
268	/**
269	* Insert into carbon table from other source
270	*/
271	case class CarbonPreInsertionCasts(sparkSession: SparkSession) extends Rule[LogicalPlan] {	2✔
272	def apply(plan: LogicalPlan): LogicalPlan = {
273	plan.transform {	2✔
274	// Wait until children are resolved.
275	case p: LogicalPlan if !p.childrenResolved => p	2✔
276
277	case p@InsertIntoTable(relation: LogicalRelation, _, child, _, _)	2✔
278	if relation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>	2✔
279	castChildOutput(p, relation, child)	2✔
280	}
281	}
282
283	def castChildOutput(p: InsertIntoTable,
284	relation: LogicalRelation,
285	child: LogicalPlan): LogicalPlan = {	2✔
286	val carbonDSRelation = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]	2✔
287	if (carbonDSRelation.carbonRelation.output.size > CarbonCommonConstants	2✔
288	.DEFAULT_MAX_NUMBER_OF_COLUMNS) {	2✔
289	CarbonException.analysisException(	×
290	s"Maximum number of columns supported: " +	×
291	s"${CarbonCommonConstants.DEFAULT_MAX_NUMBER_OF_COLUMNS}")	×
292	}
293	if (child.output.size >= carbonDSRelation.carbonRelation.output.size \|\|	2✔
294	carbonDSRelation.carbonTable.isHivePartitionTable) {	2✔
295	val newChildOutput = child.output.zipWithIndex.map { columnWithIndex =>	2✔
296	columnWithIndex._1 match {	2✔
297	case attr: Attribute =>	2✔
298	Alias(attr, s"col${ columnWithIndex._2 }")(NamedExpression.newExprId)	2✔
299	case attr => attr	×
300	}
301	}
302	val newChild: LogicalPlan = if (newChildOutput == child.output) {	2✔
303	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	×
304	CarbonReflectionUtils.getField("child", p).asInstanceOf[LogicalPlan]	×
305	} else if (SparkUtil.isSparkVersionEqualTo("2.2")) {	×
306	CarbonReflectionUtils.getField("query", p).asInstanceOf[LogicalPlan]	×
307	} else {
308	throw new UnsupportedOperationException(s"Spark version $SPARK_VERSION is not supported")	×
309	}
310	} else {
311	Project(newChildOutput, child)	2✔
312	}
313
314	val overwrite = CarbonReflectionUtils.getOverWriteOption("overwrite", p)	2✔
315
316	InsertIntoCarbonTable(carbonDSRelation, p.partition, newChild, overwrite, true)	2✔
317	} else {
318	CarbonException.analysisException(	2✔
319	"Cannot insert into target table because number of columns mismatch")	2✔
320	}
321	}
322	}
323

apache / carbondata / 11586

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