apache / carbondata / 718

Build # 718

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/bd838a7f706e808c7bb186dcf88b657b3e5a8437">bd838a7f7<a href="https://github.com/apache/carbondata/commit/bd838a7f706e808c7bb186dcf88b657b3e5a8437">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/eb24c2e68dce926096c59d980f6bdb7af613cfca">eb24c2e68</a>">eb24c2e68</a><a href="https://github.com/apache/carbondata/commit/bd838a7f706e808c7bb186dcf88b657b3e5a8437"> into 3cd8b947c">3cd8b947c</a>

Run Details

60454 of 75051 relevant lines covered (80.55%)

1.06 hits per line

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

88.24

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

import scala.collection.mutable

import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Count}
import org.apache.spark.sql.catalyst.plans.logical.{UnaryNode, _}
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.optimizer.CarbonDecoderRelation

import org.apache.carbondata.spark.CarbonAliasDecoderRelation

case class CarbonDictionaryCatalystDecoder(
    relations: Seq[CarbonDecoderRelation],
    profile: CarbonProfile,
    aliasMap: CarbonAliasDecoderRelation,
    isOuter: Boolean,
    child: LogicalPlan) extends UnaryNode {
  // the output should be updated with converted datatype, it is need for limit+sort plan.
  override def output: Seq[Attribute] = {
    child match {
      case l: LogicalRelation =>
        // If the child is logical plan then firts update all dictionary attr with IntegerType
        val logicalOut =
          CarbonDictionaryDecoder.updateAttributes(child.output, relations, aliasMap)
        CarbonDictionaryDecoder.convertOutput(logicalOut, relations, profile, aliasMap)
      case Filter(cond, l: LogicalRelation) =>
        // If the child is logical plan then firts update all dictionary attr with IntegerType
        val logicalOut =
          CarbonDictionaryDecoder.updateAttributes(child.output, relations, aliasMap)
        CarbonDictionaryDecoder.convertOutput(logicalOut, relations, profile, aliasMap)
      case Join(l: LogicalRelation, r: LogicalRelation, _, _) =>
        val logicalOut =
          CarbonDictionaryDecoder.updateAttributes(child.output, relations, aliasMap)
        CarbonDictionaryDecoder.convertOutput(logicalOut, relations, profile, aliasMap)
      case _ => CarbonDictionaryDecoder.convertOutput(child.output, relations, profile, aliasMap)
    }
  }

}

abstract class CarbonProfile(attributes: Seq[Attribute]) extends Serializable {
  def isEmpty: Boolean = attributes.isEmpty
}

case class IncludeProfile(attributes: Seq[Attribute]) extends CarbonProfile(attributes)

case class ExcludeProfile(attributes: Seq[Attribute]) extends CarbonProfile(attributes)

case class ProjectForUpdate(
    table: UnresolvedRelation,
    columns: List[String],
    children: Seq[LogicalPlan] ) extends LogicalPlan {
  override def output: Seq[AttributeReference] = Seq.empty
}

case class UpdateTable(
    table: UnresolvedRelation,
    columns: List[String],
    selectStmt: String,
    alias: Option[String] = None,
    filer: String) extends LogicalPlan {
  override def children: Seq[LogicalPlan] = Seq.empty
  override def output: Seq[AttributeReference] = Seq.empty
}

case class DeleteRecords(
    statement: String,
    alias: Option[String] = None,
    table: UnresolvedRelation) extends LogicalPlan {
  override def children: Seq[LogicalPlan] = Seq.empty
  override def output: Seq[AttributeReference] = Seq.empty
}

/**
 * A logical plan representing insertion into Hive table
 * This plan ignores nullability of ArrayType, MapType, StructType unlike InsertIntoTable
 * because Hive Table doesn't have nullability for ARRAY, MAP,STRUCT types.
 */
case class InsertIntoCarbonTable (table: CarbonDatasourceHadoopRelation,
    partition: Map[String, Option[String]],
    child: LogicalPlan,
    overwrite: Boolean,
    ifNotExists: Boolean)
  extends Command {

    override def output: Seq[Attribute] = Seq.empty

    // This is the expected schema of the table prepared to be inserted into
    // including dynamic partition columns.
    val tableOutput = table.carbonRelation.output
}

/**
 * It checks if query is count(*) then push down to carbon
 *
 * The returned values for this match are as follows:
 * - whether its count star
 * - count star attribute
 * - child plan
 */
object CountStarPlan {
  type ReturnType = (mutable.MutableList[Attribute], LogicalPlan)

 /**
  * It fill count star query attribute.
  * 2.2.1 plan
  * Aggregate [count(1) AS count(1)#30L]
  * +- Project
  *
  *2.3.0 plan
  * Aggregate [cast(count(1) as string) AS count(1)#29]
  * +- Project
  */
 private def fillCountStarAttribute(
     expr: Expression,
     outputColumns: mutable.MutableList[Attribute]) {
   expr match {
     case par@Alias(cast: Cast, _) =>
       if (cast.child.isInstanceOf[AggregateExpression]) {
         val head = cast.child.children.head
         head match {
           case count: Count if count.children.head.isInstanceOf[Literal] =>
             outputColumns += par.toAttribute
           case _ =>
         }
       }
     case par@Alias(child, _) =>
       if (child.isInstanceOf[AggregateExpression]) {
         val head = child.children.head
         head match {
           case count: Count if count.children.head.isInstanceOf[Literal] =>
             outputColumns += par.toAttribute
           case _ =>
         }
       }
   }
 }

  def unapply(plan: LogicalPlan): Option[ReturnType] = {
    plan match {
      case Aggregate(groupingExpressions, aggregateExpressions,
      child) if strictCountStar(groupingExpressions, aggregateExpressions, child) =>
        val outputColumns = scala.collection.mutable.MutableList[Attribute]()
        fillCountStarAttribute(aggregateExpressions.head, outputColumns)
        if (outputColumns.nonEmpty) {
          Some(outputColumns, child)
        } else {
          None
        }
      case _ => None
    }
  }

  /**
   * check if child
   */
  def strictCountStar(groupingExpressions: Seq[Expression],
      partialComputation: Seq[NamedExpression],
      child: LogicalPlan): Boolean = {
    if (groupingExpressions.nonEmpty) {
      return false
    }
    if (partialComputation.size > 1 && partialComputation.nonEmpty) {
      return false
    }
    child collect {
      case cd: Filter => return false
    }
    true
  }
}

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
19
20	import scala.collection.mutable
21
22	import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
23	import org.apache.spark.sql.catalyst.expressions._
24	import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Count}
25	import org.apache.spark.sql.catalyst.plans.logical.{UnaryNode, _}
26	import org.apache.spark.sql.execution.datasources.LogicalRelation
27	import org.apache.spark.sql.optimizer.CarbonDecoderRelation
28
29	import org.apache.carbondata.spark.CarbonAliasDecoderRelation
30
31	case class CarbonDictionaryCatalystDecoder(	2✔
32	relations: Seq[CarbonDecoderRelation],	2✔
33	profile: CarbonProfile,	2✔
34	aliasMap: CarbonAliasDecoderRelation,	2✔
35	isOuter: Boolean,	2✔
36	child: LogicalPlan) extends UnaryNode {	2✔
37	// the output should be updated with converted datatype, it is need for limit+sort plan.
38	override def output: Seq[Attribute] = {
39	child match {	2✔
40	case l: LogicalRelation =>	2✔
41	// If the child is logical plan then firts update all dictionary attr with IntegerType
42	val logicalOut =	2✔
43	CarbonDictionaryDecoder.updateAttributes(child.output, relations, aliasMap)	2✔
44	CarbonDictionaryDecoder.convertOutput(logicalOut, relations, profile, aliasMap)	2✔
45	case Filter(cond, l: LogicalRelation) =>	2✔
46	// If the child is logical plan then firts update all dictionary attr with IntegerType
47	val logicalOut =	2✔
48	CarbonDictionaryDecoder.updateAttributes(child.output, relations, aliasMap)	2✔
49	CarbonDictionaryDecoder.convertOutput(logicalOut, relations, profile, aliasMap)	2✔
50	case Join(l: LogicalRelation, r: LogicalRelation, _, _) =>	2✔
51	val logicalOut =	×
52	CarbonDictionaryDecoder.updateAttributes(child.output, relations, aliasMap)	×
53	CarbonDictionaryDecoder.convertOutput(logicalOut, relations, profile, aliasMap)	×
54	case _ => CarbonDictionaryDecoder.convertOutput(child.output, relations, profile, aliasMap)	2✔
55	}
56	}
57
58	}
59
60	abstract class CarbonProfile(attributes: Seq[Attribute]) extends Serializable {	2✔
61	def isEmpty: Boolean = attributes.isEmpty	2✔
62	}
63
64	case class IncludeProfile(attributes: Seq[Attribute]) extends CarbonProfile(attributes)	2✔
65
66	case class ExcludeProfile(attributes: Seq[Attribute]) extends CarbonProfile(attributes)	2✔
67
68	case class ProjectForUpdate(	2✔
69	table: UnresolvedRelation,	2✔
70	columns: List[String],	2✔
71	children: Seq[LogicalPlan] ) extends LogicalPlan {	2✔
72	override def output: Seq[AttributeReference] = Seq.empty	2✔
73	}
74
75	case class UpdateTable(	2✔
76	table: UnresolvedRelation,	2✔
77	columns: List[String],	2✔
78	selectStmt: String,	2✔
79	alias: Option[String] = None,	2✔
80	filer: String) extends LogicalPlan {	2✔
81	override def children: Seq[LogicalPlan] = Seq.empty	2✔
82	override def output: Seq[AttributeReference] = Seq.empty	×
83	}
84
85	case class DeleteRecords(	2✔
86	statement: String,	2✔
87	alias: Option[String] = None,	2✔
88	table: UnresolvedRelation) extends LogicalPlan {	2✔
89	override def children: Seq[LogicalPlan] = Seq.empty	2✔
90	override def output: Seq[AttributeReference] = Seq.empty	×
91	}
92
93	/**
94	* A logical plan representing insertion into Hive table
95	* This plan ignores nullability of ArrayType, MapType, StructType unlike InsertIntoTable
96	* because Hive Table doesn't have nullability for ARRAY, MAP,STRUCT types.
97	*/
98	case class InsertIntoCarbonTable (table: CarbonDatasourceHadoopRelation,	2✔
99	partition: Map[String, Option[String]],	2✔
100	child: LogicalPlan,	2✔
101	overwrite: Boolean,	2✔
102	ifNotExists: Boolean)	2✔
103	extends Command {	2✔
104
105	override def output: Seq[Attribute] = Seq.empty	2✔
106
107	// This is the expected schema of the table prepared to be inserted into
108	// including dynamic partition columns.
109	val tableOutput = table.carbonRelation.output	2✔
110	}
111
112	/**
113	* It checks if query is count(*) then push down to carbon
114	*
115	* The returned values for this match are as follows:
116	* - whether its count star
117	* - count star attribute
118	* - child plan
119	*/
120	object CountStarPlan {
121	type ReturnType = (mutable.MutableList[Attribute], LogicalPlan)
122
123	/**
124	* It fill count star query attribute.
125	* 2.2.1 plan
126	* Aggregate [count(1) AS count(1)#30L]
127	* +- Project
128	*
129	*2.3.0 plan
130	* Aggregate [cast(count(1) as string) AS count(1)#29]
131	* +- Project
132	*/
133	private def fillCountStarAttribute(
134	expr: Expression,
135	outputColumns: mutable.MutableList[Attribute]) {
136	expr match {	2✔
137	case par@Alias(cast: Cast, _) =>	2✔
138	if (cast.child.isInstanceOf[AggregateExpression]) {	1✔
139	val head = cast.child.children.head	×
140	head match {	×
141	case count: Count if count.children.head.isInstanceOf[Literal] =>	×
142	outputColumns += par.toAttribute	×
143	case _ =>	×
144	}
145	}
146	case par@Alias(child, _) =>	2✔
147	if (child.isInstanceOf[AggregateExpression]) {	2✔
148	val head = child.children.head	2✔
149	head match {	2✔
150	case count: Count if count.children.head.isInstanceOf[Literal] =>	2✔
151	outputColumns += par.toAttribute	2✔
152	case _ =>	2✔
153	}
154	}
155	}
156	}
157
158	def unapply(plan: LogicalPlan): Option[ReturnType] = {
159	plan match {	2✔
160	case Aggregate(groupingExpressions, aggregateExpressions,	2✔
161	child) if strictCountStar(groupingExpressions, aggregateExpressions, child) =>	2✔
162	val outputColumns = scala.collection.mutable.MutableList[Attribute]()	2✔
163	fillCountStarAttribute(aggregateExpressions.head, outputColumns)	2✔
164	if (outputColumns.nonEmpty) {	2✔
165	Some(outputColumns, child)	2✔
166	} else {
167	None	2✔
168	}
169	case _ => None	2✔
170	}
171	}
172
173	/**
174	* check if child
175	*/
176	def strictCountStar(groupingExpressions: Seq[Expression],	2✔
177	partialComputation: Seq[NamedExpression],
178	child: LogicalPlan): Boolean = {	2✔
179	if (groupingExpressions.nonEmpty) {	2✔
180	return false	2✔
181	}
182	if (partialComputation.size > 1 && partialComputation.nonEmpty) {	2✔
183	return false	2✔
184	}
185	child collect {	2✔
186	case cd: Filter => return false	2✔
187	}
188	true	2✔
189	}
190	}	2✔

apache / carbondata / 718

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