6644

Build # 6644

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/3ba72d942744414cd1dfb564d564412fab6d5198">3ba72d942<a href="https://github.com/apache/carbondata/commit/3ba72d942744414cd1dfb564d564412fab6d5198">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c7e5ff9f29be0d3cdc07e8618f9a194a9f6c2ddf">c7e5ff9f2</a>">c7e5ff9f2</a><a href="https://github.com/apache/carbondata/commit/3ba72d942744414cd1dfb564d564412fab6d5198"> into f66a9fdca">f66a9fdca</a>

Run Details

57189 of 70656 relevant lines covered (80.94%)

1.08 hits per line

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

90.06

/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

import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
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 (!indexSchemas.isEmpty) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for table which has index datamaps")
        }
      }
      val tableRelation = if (SPARK_VERSION.startsWith("2.1")) {
        relation
      } else if (SPARK_VERSION.startsWith("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 (!indexSchemas.isEmpty) {
            throw new UnsupportedOperationException(
              "Delete operation is not supported for table which has index datamaps")
          }
        }
        // include tuple id in subquery
        if (SPARK_VERSION.startsWith("2.1")) {
          Project(projList, relation)
        } else if (SPARK_VERSION.startsWith("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: Alias =>
            Alias(attr.child, s"col${ columnWithIndex._2 }")(attr.exprId)
          case attr: Attribute =>
            Alias(attr, s"col${ columnWithIndex._2 }")(NamedExpression.newExprId)
          case attr => attr
        }
      }
      val version = SPARK_VERSION
      val newChild: LogicalPlan = if (newChildOutput == child.output) {
        if (version.startsWith("2.1")) {
          CarbonReflectionUtils.getField("child", p).asInstanceOf[LogicalPlan]
        } else if (version.startsWith("2.2")) {
          CarbonReflectionUtils.getField("query", p).asInstanceOf[LogicalPlan]
        } else {
          throw new UnsupportedOperationException(s"Spark version $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
36
37	import org.apache.carbondata.core.constants.CarbonCommonConstants
38	import org.apache.carbondata.core.datamap.DataMapStoreManager
39	import org.apache.carbondata.core.util.CarbonUtil
40
41	case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {	2✔
42
43	private lazy val parser = sparkSession.sessionState.sqlParser	2✔
44
45	private def processUpdateQuery(
46	table: UnresolvedRelation,
47	columns: List[String],
48	selectStmt: String,
49	alias: Option[String],
50	filter: String): LogicalPlan = {
51	var includedDestColumns = false	2✔
52	var includedDestRelation = false	2✔
53	var addedTupleId = false	2✔
54
55	def prepareTargetReleation(relation: UnresolvedRelation): SubqueryAlias = {
56	val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",	2✔
57	Seq.empty, isDistinct = false), "tupleId")())	2✔
58
59	val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)	2✔
60	val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)	2✔
61	if (carbonTable != null) {	2✔
62	if (CarbonUtil.hasAggregationDataMap(carbonTable)) {	2✔
63	throw new UnsupportedOperationException(	1✔
64	"Update operation is not supported for tables which have a pre-aggregate table. " +	1✔
65	"Drop pre-aggregate tables to continue.")
66	}
67	if (carbonTable.isChildDataMap) {	2✔
68	throw new UnsupportedOperationException(	1✔
69	"Update operation is not supported for pre-aggregate table")	1✔
70	}
71	val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)	2✔
72	if (!indexSchemas.isEmpty) {	2✔
73	throw new UnsupportedOperationException(	1✔
74	"Update operation is not supported for table which has index datamaps")	1✔
75	}
76	}
77	val tableRelation = if (SPARK_VERSION.startsWith("2.1")) {	2✔
78	relation	×
79	} else if (SPARK_VERSION.startsWith("2.2")) {	2✔
80	alias match {	2✔
81	case Some(_) =>	2✔
82	CarbonReflectionUtils.getSubqueryAlias(	2✔
83	sparkSession,	2✔
84	alias,	2✔
85	relation,	2✔
86	Some(table.tableIdentifier))	2✔
87	case _ => relation	×
88	}
89	} else {
90	throw new UnsupportedOperationException("Unsupported Spark version.")	×
91	}
92
93	CarbonReflectionUtils.getSubqueryAlias(	2✔
94	sparkSession,	2✔
95	alias,	2✔
96	Project(projList, tableRelation),	2✔
97	Some(table.tableIdentifier))	2✔
98	}
99
100	// get the un-analyzed logical plan
101	val targetTable = prepareTargetReleation(table)	2✔
102	val selectPlan = parser.parsePlan(selectStmt) transform {	2✔
103	case Project(projectList, child) if !includedDestColumns =>	2✔
104	includedDestColumns = true	2✔
105	if (projectList.size != columns.size) {	2✔
106	CarbonException.analysisException(	1✔
107	"The number of columns in source table and destination table columns mismatch")	1✔
108	}
109	val renamedProjectList = projectList.zip(columns).map { case (attr, col) =>	2✔
110	attr match {	2✔
111	case UnresolvedAlias(child22, _) =>	2✔
112	UnresolvedAlias(Alias(child22, col + "-updatedColumn")())	2✔
113	case UnresolvedAttribute(_) =>	2✔
114	UnresolvedAlias(Alias(attr, col + "-updatedColumn")())	2✔
115	case _ => attr	×
116	}
117	}
118	val tableName: Option[Seq[String]] = alias match {	2✔
119	case Some(_) => Some(alias.toSeq)	2✔
120	case _ => Some(Seq(child.asInstanceOf[UnresolvedRelation].tableIdentifier.table.toString))	×
121	}
122	val list = Seq(	2✔
123	UnresolvedAlias(UnresolvedStar(tableName))) ++ renamedProjectList	2✔
124	Project(list, child)	2✔
125	case Filter(cond, child) if !includedDestRelation =>	2✔
126	includedDestRelation = true	2✔
127	Filter(cond, Join(child, targetTable, Inner, None))	2✔
128	case r@CarbonUnresolvedRelation(t) if !includedDestRelation && t != table.tableIdentifier =>	2✔
129	includedDestRelation = true	1✔
130	Join(r, targetTable, Inner, None)	1✔
131	}
132	val updatedSelectPlan: LogicalPlan = if (!includedDestRelation) {	2✔
133	// special case to handle self join queries
134	// Eg. update tableName SET (column1) = (column1+1)
135	selectPlan transform {	2✔
136	case relation: UnresolvedRelation	2✔
137	if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>	2✔
138	addedTupleId = true	2✔
139	targetTable	2✔
140	}
141	} else {
142	selectPlan	2✔
143	}
144	val finalPlan = if (filter.length > 0) {	2✔
145	var transformed: Boolean = false	2✔
146	// Create a dummy projection to include filter conditions
147	var newPlan: LogicalPlan = null	2✔
148	if (table.tableIdentifier.database.isDefined) {	2✔
149	newPlan = parser.parsePlan("select * from " +	2✔
150	table.tableIdentifier.database.getOrElse("") + "." +	2✔
151	table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)	2✔
152	}
153	else {
154	newPlan = parser.parsePlan("select * from " +	2✔
155	table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)	2✔
156	}
157	newPlan transform {	2✔
158	case CarbonUnresolvedRelation(t)	2✔
159	if !transformed && t == table.tableIdentifier =>	2✔
160	transformed = true	2✔
161
162	CarbonReflectionUtils.getSubqueryAlias(	2✔
163	sparkSession,	2✔
164	alias,	2✔
165	updatedSelectPlan,	2✔
166	Some(table.tableIdentifier))	2✔
167	}
168	} else {
169	updatedSelectPlan	2✔
170	}
171	val destinationTable = CarbonReflectionUtils.getUnresolvedRelation(table.tableIdentifier, alias)	2✔
172
173	ProjectForUpdate(destinationTable, columns, Seq(finalPlan))	2✔
174	}
175
176
177	def processDeleteRecordsQuery(selectStmt: String,
178	alias: Option[String],
179	table: UnresolvedRelation): LogicalPlan = {
180	var addedTupleId = false	2✔
181	val parsePlan = parser.parsePlan(selectStmt)	2✔
182
183	val selectPlan = parsePlan transform {	2✔
184	case relation: UnresolvedRelation	2✔
185	if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>	2✔
186	addedTupleId = true	2✔
187	val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",	2✔
188	Seq.empty, isDistinct = false), "tupleId")())	2✔
189
190	val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)	2✔
191	val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)	2✔
192	if (carbonTable != null) {	2✔
193	if (CarbonUtil.hasAggregationDataMap(carbonTable)) {	2✔
194	throw new UnsupportedOperationException(	1✔
195	"Delete operation is not supported for tables which have a pre-aggregate table. " +	1✔
196	"Drop pre-aggregate tables to continue.")
197	}
198	if (carbonTable.isChildDataMap) {	2✔
199	throw new UnsupportedOperationException(	1✔
200	"Delete operation is not supported for pre-aggregate table")	1✔
201	}
202	val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)	2✔
203	if (!indexSchemas.isEmpty) {	2✔
204	throw new UnsupportedOperationException(	1✔
205	"Delete operation is not supported for table which has index datamaps")	1✔
206	}
207	}
208	// include tuple id in subquery
209	if (SPARK_VERSION.startsWith("2.1")) {	2✔
210	Project(projList, relation)	×
211	} else if (SPARK_VERSION.startsWith("2.2")) {	2✔
212	alias match {	2✔
213	case Some(_) =>	2✔
214	val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(	1✔
215	sparkSession,	1✔
216	alias,	1✔
217	relation,	1✔
218	Some(table.tableIdentifier))	1✔
219	Project(projList, subqueryAlias)	1✔
220	case _ => Project(projList, relation)	2✔
221	}
222	} else {
223	throw new UnsupportedOperationException("Unsupported Spark version.")	×
224	}
225	}
226	CarbonProjectForDeleteCommand(	2✔
227	selectPlan,	2✔
228	table.tableIdentifier.database,	2✔
229	table.tableIdentifier.table,	2✔
230	System.currentTimeMillis().toString)	2✔
231	}
232
233	override def apply(logicalplan: LogicalPlan): LogicalPlan = {
234
235	logicalplan transform {	2✔
236	case UpdateTable(t, cols, sel, alias, where) => processUpdateQuery(t, cols, sel, alias, where)	2✔
237	case DeleteRecords(statement, alias, table) =>	2✔
238	processDeleteRecordsQuery(	2✔
239	statement,	2✔
240	alias,	2✔
241	table)	2✔
242	}
243	}
244	}
245
246	/**
247	* Insert into carbon table from other source
248	*/
249	case class CarbonPreInsertionCasts(sparkSession: SparkSession) extends Rule[LogicalPlan] {	2✔
250	def apply(plan: LogicalPlan): LogicalPlan = {
251	plan.transform {	2✔
252	// Wait until children are resolved.
253	case p: LogicalPlan if !p.childrenResolved => p	2✔
254
255	case p@InsertIntoTable(relation: LogicalRelation, _, child, _, _)	2✔
256	if relation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>	2✔
257	castChildOutput(p, relation, child)	2✔
258	}
259	}
260
261	def castChildOutput(p: InsertIntoTable,
262	relation: LogicalRelation,
263	child: LogicalPlan): LogicalPlan = {	2✔
264	val carbonDSRelation = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]	2✔
265	if (carbonDSRelation.carbonRelation.output.size > CarbonCommonConstants	2✔
266	.DEFAULT_MAX_NUMBER_OF_COLUMNS) {	2✔
267	CarbonException.analysisException(	×
268	s"Maximum number of columns supported: " +	×
269	s"${CarbonCommonConstants.DEFAULT_MAX_NUMBER_OF_COLUMNS}")	×
270	}
271	if (child.output.size >= carbonDSRelation.carbonRelation.output.size \|\|	2✔
272	carbonDSRelation.carbonTable.isHivePartitionTable) {	2✔
273	val newChildOutput = child.output.zipWithIndex.map { columnWithIndex =>	2✔
274	columnWithIndex._1 match {	2✔
275	case attr: Alias =>	2✔
276	Alias(attr.child, s"col${ columnWithIndex._2 }")(attr.exprId)	×
277	case attr: Attribute =>	2✔
278	Alias(attr, s"col${ columnWithIndex._2 }")(NamedExpression.newExprId)	2✔
279	case attr => attr	×
280	}
281	}
282	val version = SPARK_VERSION	2✔
283	val newChild: LogicalPlan = if (newChildOutput == child.output) {	2✔
284	if (version.startsWith("2.1")) {	×
285	CarbonReflectionUtils.getField("child", p).asInstanceOf[LogicalPlan]	×
286	} else if (version.startsWith("2.2")) {	×
287	CarbonReflectionUtils.getField("query", p).asInstanceOf[LogicalPlan]	×
288	} else {
289	throw new UnsupportedOperationException(s"Spark version $version is not supported")	×
290	}
291	} else {
292	Project(newChildOutput, child)	2✔
293	}
294
295	val overwrite = CarbonReflectionUtils.getOverWriteOption("overwrite", p)	2✔
296
297	InsertIntoCarbonTable(carbonDSRelation, p.partition, newChild, overwrite, true)	2✔
298	} else {
299	CarbonException.analysisException(	2✔
300	"Cannot insert into target table because number of columns mismatch")	2✔
301	}
302	}
303	}
304

apache / carbondata / 6644

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