8141

Build # 8141

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/62bc35d64e5dc171edb0112e658259f189361f43">62bc35d64<a href="https://github.com/apache/carbondata/commit/62bc35d64e5dc171edb0112e658259f189361f43">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/dea293ccb59bbeecc72dd17c2ecfc24627e952b8">dea293ccb</a>">dea293ccb</a><a href="https://github.com/apache/carbondata/commit/62bc35d64e5dc171edb0112e658259f189361f43"> into 2f537b724">2f537b724</a>

Run Details

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

57233 of 70982 relevant lines covered (80.63%)

1.07 hits per line

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

75.41

/integration/spark2/src/main/scala/org/apache/spark/sql/SparkUnknownExpression.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 java.util.{ArrayList, List}

import scala.collection.JavaConverters._

import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Expression => SparkExpression, GenericInternalRow}

import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.scan.expression.{ColumnExpression, Expression, ExpressionResult, UnknownExpression}
import org.apache.carbondata.core.scan.expression.conditional.ConditionalExpression
import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException
import org.apache.carbondata.core.scan.filter.intf.{ExpressionType, RowIntf}


class SparkUnknownExpression(
    var sparkExp: SparkExpression,
    expressionType: ExpressionType = ExpressionType.UNKNOWN)
  extends UnknownExpression with ConditionalExpression {

  private var evaluateExpression: (InternalRow) => Any = sparkExp.eval
  private var isExecutor: Boolean = false
  children.addAll(getColumnList())

  override def evaluate(carbonRowInstance: RowIntf): ExpressionResult = {

    val values = carbonRowInstance.getValues.toSeq.map {
      case s: String => org.apache.spark.unsafe.types.UTF8String.fromString(s)
      case d: java.math.BigDecimal => org.apache.spark.sql.types.Decimal.apply(d)
      case b: Array[Byte] => org.apache.spark.unsafe.types.UTF8String.fromBytes(b)
      case value => value
    }
    try {
      val result = evaluateExpression(
        new GenericInternalRow(values.map(a => a.asInstanceOf[Any]).toArray))
      val sparkRes = if (isExecutor) {
        result.asInstanceOf[InternalRow].get(0, sparkExp.dataType)
      } else {
        result
      }
      new ExpressionResult(
        CarbonSparkDataSourceUtil.convertSparkToCarbonDataType(sparkExp.dataType),
        sparkRes
      )
    } catch {
      case e: Exception => throw new FilterUnsupportedException(e.getMessage)
    }
  }

  override def getFilterExpressionType: ExpressionType = {
    expressionType
  }

  override def getString: String = {
    sparkExp.toString()
  }

  override def getStatement: String = {
    sparkExp.toString()
  }

  def setEvaluateExpression(evaluateExpression: (InternalRow) => Any): Unit = {
    this.evaluateExpression = evaluateExpression
    isExecutor = true
  }

  override def findAndSetChild(oldExpr: Expression, newExpr: Expression): Unit = {}

  def getColumnList: java.util.List[ColumnExpression] = {

    val lst = new java.util.ArrayList[ColumnExpression]()
    getColumnListFromExpressionTree(sparkExp, lst)
    lst
  }

  def getLiterals: java.util.List[ExpressionResult] = {

    val lst = new java.util.ArrayList[ExpressionResult]()
    lst
  }

  def getAllColumnList: java.util.List[ColumnExpression] = {
    val lst = new java.util.ArrayList[ColumnExpression]()
    getAllColumnListFromExpressionTree(sparkExp, lst)
    lst
  }

  def isSingleColumn: Boolean = {
    val lst = new java.util.ArrayList[ColumnExpression]()
    getAllColumnListFromExpressionTree(sparkExp, lst)
    if (lst.size == 1 && lst.get(0).isDimension) {
      true
    } else {
      false
    }
  }

  def getColumnListFromExpressionTree(sparkCurrentExp: SparkExpression,
      lst: java.util.List[ColumnExpression]): Unit = {
    sparkCurrentExp match {
      case carbonBoundRef: CarbonBoundReference =>
        val foundExp = lst.asScala
          .find(p => p.getColumnName() == carbonBoundRef.colExp.getColumnName())
        if (foundExp.isEmpty) {
          carbonBoundRef.colExp.setColIndex(lst.size)
          lst.add(carbonBoundRef.colExp)
        } else {
          carbonBoundRef.colExp.setColIndex(foundExp.get.getColIndex())
        }
      case _ => sparkCurrentExp.children.foreach(getColumnListFromExpressionTree(_, lst))
    }
  }




  def getAllColumnListFromExpressionTree(sparkCurrentExp: SparkExpression,
      list: List[ColumnExpression]): List[ColumnExpression] = {
    sparkCurrentExp match {
      case carbonBoundRef: CarbonBoundReference => list.add(carbonBoundRef.colExp)
      case _ => sparkCurrentExp.children.foreach(getColumnListFromExpressionTree(_, list))
    }
    list
  }

  def isDirectDictionaryColumns: Boolean = {
    val lst = new ArrayList[ColumnExpression]()
    getAllColumnListFromExpressionTree(sparkExp, lst)
    if (lst.get(0).getCarbonColumn.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
      true
    } else {
      false
    }
  }


}

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 java.util.{ArrayList, List}
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil
25	import org.apache.spark.sql.catalyst.InternalRow
26	import org.apache.spark.sql.catalyst.expressions.{Expression => SparkExpression, GenericInternalRow}
27
28	import org.apache.carbondata.core.metadata.encoder.Encoding
29	import org.apache.carbondata.core.scan.expression.{ColumnExpression, Expression, ExpressionResult, UnknownExpression}
30	import org.apache.carbondata.core.scan.expression.conditional.ConditionalExpression
31	import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException
32	import org.apache.carbondata.core.scan.filter.intf.{ExpressionType, RowIntf}
33
34
35	class SparkUnknownExpression(	2✔
36	var sparkExp: SparkExpression,	2✔
37	expressionType: ExpressionType = ExpressionType.UNKNOWN)	2✔
38	extends UnknownExpression with ConditionalExpression {	2✔
39
40	private var evaluateExpression: (InternalRow) => Any = sparkExp.eval	2✔
41	private var isExecutor: Boolean = false	2✔
42	children.addAll(getColumnList())	2✔
43
44	override def evaluate(carbonRowInstance: RowIntf): ExpressionResult = {	2✔
45
46	val values = carbonRowInstance.getValues.toSeq.map {	2✔
47	case s: String => org.apache.spark.unsafe.types.UTF8String.fromString(s)	2✔
48	case d: java.math.BigDecimal => org.apache.spark.sql.types.Decimal.apply(d)	2✔
49	case b: Array[Byte] => org.apache.spark.unsafe.types.UTF8String.fromBytes(b)	2✔
50	case value => value	2✔
51	}
52	try {	×
53	val result = evaluateExpression(	2✔
54	new GenericInternalRow(values.map(a => a.asInstanceOf[Any]).toArray))	2✔
55	val sparkRes = if (isExecutor) {	2✔
56	result.asInstanceOf[InternalRow].get(0, sparkExp.dataType)	×
57	} else {
58	result	2✔
59	}
60	new ExpressionResult(	2✔
61	CarbonSparkDataSourceUtil.convertSparkToCarbonDataType(sparkExp.dataType),	2✔
62	sparkRes	2✔
63	)
64	} catch {
65	case e: Exception => throw new FilterUnsupportedException(e.getMessage)
66	}
67	}
68
69	override def getFilterExpressionType: ExpressionType = {
70	expressionType	2✔
71	}
72
73	override def getString: String = {
74	sparkExp.toString()	2✔
75	}
76
77	override def getStatement: String = {
78	sparkExp.toString()	2✔
79	}
80
81	def setEvaluateExpression(evaluateExpression: (InternalRow) => Any): Unit = {
82	this.evaluateExpression = evaluateExpression	×
83	isExecutor = true	×
84	}
85
86	override def findAndSetChild(oldExpr: Expression, newExpr: Expression): Unit = {}	×
87
88	def getColumnList: java.util.List[ColumnExpression] = {
89
90	val lst = new java.util.ArrayList[ColumnExpression]()	2✔
91	getColumnListFromExpressionTree(sparkExp, lst)	2✔
92	lst	2✔
93	}
94
95	def getLiterals: java.util.List[ExpressionResult] = {
96
97	val lst = new java.util.ArrayList[ExpressionResult]()	×
98	lst	×
99	}
100
101	def getAllColumnList: java.util.List[ColumnExpression] = {
102	val lst = new java.util.ArrayList[ColumnExpression]()	2✔
103	getAllColumnListFromExpressionTree(sparkExp, lst)	2✔
104	lst	2✔
105	}
106
107	def isSingleColumn: Boolean = {	2✔
108	val lst = new java.util.ArrayList[ColumnExpression]()	2✔
109	getAllColumnListFromExpressionTree(sparkExp, lst)	2✔
110	if (lst.size == 1 && lst.get(0).isDimension) {	2✔
111	true	2✔
112	} else {
113	false	×
114	}
115	}
116
117	def getColumnListFromExpressionTree(sparkCurrentExp: SparkExpression,
118	lst: java.util.List[ColumnExpression]): Unit = {
119	sparkCurrentExp match {	2✔
120	case carbonBoundRef: CarbonBoundReference =>	2✔
121	val foundExp = lst.asScala	2✔
122	.find(p => p.getColumnName() == carbonBoundRef.colExp.getColumnName())	2✔
123	if (foundExp.isEmpty) {	2✔
124	carbonBoundRef.colExp.setColIndex(lst.size)	2✔
125	lst.add(carbonBoundRef.colExp)	2✔
126	} else {
127	carbonBoundRef.colExp.setColIndex(foundExp.get.getColIndex())	×
128	}
129	case _ => sparkCurrentExp.children.foreach(getColumnListFromExpressionTree(_, lst))	2✔
130	}
131	}
132
133
134
135
136	def getAllColumnListFromExpressionTree(sparkCurrentExp: SparkExpression,
137	list: List[ColumnExpression]): List[ColumnExpression] = {
138	sparkCurrentExp match {	2✔
139	case carbonBoundRef: CarbonBoundReference => list.add(carbonBoundRef.colExp)	2✔
140	case _ => sparkCurrentExp.children.foreach(getColumnListFromExpressionTree(_, list))	2✔
141	}
142	list	2✔
143	}
144
145	def isDirectDictionaryColumns: Boolean = {	×
146	val lst = new ArrayList[ColumnExpression]()	×
147	getAllColumnListFromExpressionTree(sparkExp, lst)	×
148	if (lst.get(0).getCarbonColumn.hasEncoding(Encoding.DIRECT_DICTIONARY)) {	×
149	true	×
150	} else {
151	false	×
152	}
153	}
154
155
156	}

apache / carbondata / 8141

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