8253

Build # 8253

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/431cf922f9ad4b9f2053e5aaf1c0b3b4a09fa7b7">431cf922f<a href="https://github.com/apache/carbondata/commit/431cf922f9ad4b9f2053e5aaf1c0b3b4a09fa7b7">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/3070a16a4d396cd77a8ad2f458cc7ddc7fa241cf">3070a16a4</a>">3070a16a4</a><a href="https://github.com/apache/carbondata/commit/431cf922f9ad4b9f2053e5aaf1c0b3b4a09fa7b7"> into 884ac915b">884ac915b</a>

Run Details

71 of 71 new or added lines in 6 files covered. (100.0%)

57556 of 71358 relevant lines covered (80.66%)

1.07 hits per line

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

59.7

/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.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.util

import java.lang.reflect.Field

import scala.reflect.runtime._
import scala.reflect.runtime.universe._

import org.apache.spark.SPARK_VERSION
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.catalyst.parser.AstBuilder
import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, LogicalPlan, SubqueryAlias}
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.sources.BaseRelation

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants

/**
 * Reflection APIs
 */

object CarbonReflectionUtils {
  private val rm = universe.runtimeMirror(getClass.getClassLoader)

  /**
   * Returns the field val from a object through reflection.
   * @param name - name of the field being retrieved.
   * @param obj - Object from which the field has to be retrieved.
   * @tparam T
   * @return
   */
  def getField[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
    val im = rm.reflect(obj)
    im.symbol.typeSignature.members.find(_.name.toString.equals(name))
      .map(l => im.reflectField(l.asTerm).get).getOrElse(null)
  }

  def getUnresolvedRelation(
      tableIdentifier: TableIdentifier,
      tableAlias: Option[String] = None): UnresolvedRelation = {
    val className = "org.apache.spark.sql.catalyst.analysis.UnresolvedRelation"
    if (SPARK_VERSION.startsWith("2.1")) {
      createObject(
        className,
        tableIdentifier,
        tableAlias)._1.asInstanceOf[UnresolvedRelation]
    } else if (SPARK_VERSION.startsWith("2.2")) {
      createObject(
        className,
        tableIdentifier)._1.asInstanceOf[UnresolvedRelation]
    } else {
      throw new UnsupportedOperationException(s"Unsupported Spark version $SPARK_VERSION")
    }
  }

  def getSubqueryAlias(sparkSession: SparkSession, alias: Option[String],
      relation: LogicalPlan,
      view: Option[TableIdentifier]): SubqueryAlias = {
    val className = "org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias"
    if (SPARK_VERSION.startsWith("2.1")) {
      createObject(
        className,
        alias.getOrElse(""),
        relation,
        Option(view))._1.asInstanceOf[SubqueryAlias]
    } else if (SPARK_VERSION.startsWith("2.2")) {
      createObject(
        className,
        alias.getOrElse(""),
        relation)._1.asInstanceOf[SubqueryAlias]
    } else {
      throw new UnsupportedOperationException("Unsupported Spark version")
    }
  }

  def getInsertIntoCommand(table: LogicalPlan,
      partition: Map[String, Option[String]],
      query: LogicalPlan,
      overwrite: Boolean,
      ifPartitionNotExists: Boolean): InsertIntoTable = {
    val className = "org.apache.spark.sql.catalyst.plans.logical.InsertIntoTable"
    if (SPARK_VERSION.startsWith("2.1")) {
      val overwriteOptions = createObject(
        "org.apache.spark.sql.catalyst.plans.logical.OverwriteOptions",
        overwrite.asInstanceOf[Object], Map.empty.asInstanceOf[Object])._1.asInstanceOf[Object]
      createObject(
        className,
        table,
        partition,
        query,
        overwriteOptions,
        ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]
    } else if (SPARK_VERSION.startsWith("2.2")) {
      createObject(
        className,
        table,
        partition,
        query,
        overwrite.asInstanceOf[Object],
        ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]
    } else {
      throw new UnsupportedOperationException("Unsupported Spark version")
    }
  }

  def getLogicalRelation(relation: BaseRelation,
      expectedOutputAttributes: Seq[Attribute],
      catalogTable: Option[CatalogTable]): LogicalRelation = {
    val className = "org.apache.spark.sql.execution.datasources.LogicalRelation"
    if (SPARK_VERSION.startsWith("2.1")) {
      createObject(
        className,
        relation,
        Some(expectedOutputAttributes),
        catalogTable)._1.asInstanceOf[LogicalRelation]
    } else if (SPARK_VERSION.startsWith("2.2")) {
      createObject(
        className,
        relation,
        expectedOutputAttributes,
        catalogTable)._1.asInstanceOf[LogicalRelation]
    } else {
      throw new UnsupportedOperationException("Unsupported Spark version")
    }
  }


  def getOverWriteOption[T: TypeTag : reflect.ClassTag](name: String, obj: T): Boolean = {
    var overwriteboolean: Boolean = false
    val im = rm.reflect(obj)
    for (m <- typeOf[T].members.filter(!_.isMethod)) {
      if (m.toString.contains("overwrite")) {
        val typ = m.typeSignature
        if (typ.toString.contains("Boolean")) {
          // Spark2.2
          overwriteboolean = im.reflectField(m.asTerm).get.asInstanceOf[Boolean]
        } else {
          overwriteboolean = getOverWrite("enabled", im.reflectField(m.asTerm).get)
        }
      }
    }
    overwriteboolean
  }

  private def getOverWrite[T: TypeTag : reflect.ClassTag](name: String, obj: T): Boolean = {
    var overwriteboolean: Boolean = false
    val im = rm.reflect(obj)
    for (l <- im.symbol.typeSignature.members.filter(_.name.toString.contains("enabled"))) {
      overwriteboolean = im.reflectField(l.asTerm).get.asInstanceOf[Boolean]
    }
    overwriteboolean
  }

  def getFieldOfCatalogTable[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
    val im = rm.reflect(obj)
    val sym = im.symbol.typeSignature.member(TermName(name))
    val tableMeta = im.reflectMethod(sym.asMethod).apply()
    tableMeta
  }

  def getAstBuilder(conf: Object,
      sqlParser: Object,
      sparkSession: SparkSession): AstBuilder = {
    if (SPARK_VERSION.startsWith("2.1") || SPARK_VERSION.startsWith("2.2")) {
      val className = sparkSession.sparkContext.conf.get(
        CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME,
        "org.apache.spark.sql.hive.CarbonSqlAstBuilder")
      createObject(className,
        conf,
        sqlParser, sparkSession)._1.asInstanceOf[AstBuilder]
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def getSessionState(sparkContext: SparkContext,
      carbonSession: Object,
      useHiveMetaStore: Boolean): Any = {
    if (SPARK_VERSION.startsWith("2.1")) {
      val className = sparkContext.conf.get(
        CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
        "org.apache.spark.sql.hive.CarbonSessionState")
      createObject(className, carbonSession)._1
    } else if (SPARK_VERSION.startsWith("2.2")) {
      if (useHiveMetaStore) {
        val className = sparkContext.conf.get(
          CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
          "org.apache.spark.sql.hive.CarbonSessionStateBuilder")
        val tuple = createObject(className, carbonSession, None)
        val method = tuple._2.getMethod("build")
        method.invoke(tuple._1)
      } else {
        val className = sparkContext.conf.get(
          CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
          "org.apache.spark.sql.hive.CarbonInMemorySessionStateBuilder")
        val tuple = createObject(className, carbonSession, None)
        val method = tuple._2.getMethod("build")
        method.invoke(tuple._1)
      }
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def hasPredicateSubquery(filterExp: Expression) : Boolean = {
    if (SPARK_VERSION.startsWith("2.1")) {
      val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.PredicateSubquery")
      val method = tuple.getMethod("hasPredicateSubquery", classOf[Expression])
      val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]
      hasSubquery
    } else if (SPARK_VERSION.startsWith("2.2")) {
      val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.SubqueryExpression")
      val method = tuple.getMethod("hasInOrExistsSubquery", classOf[Expression])
      val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]
      hasSubquery
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def getDescribeTableFormattedField[T: TypeTag : reflect.ClassTag](obj: T): Boolean = {
    val im = rm.reflect(obj)
    val isFormatted = im.symbol.typeSignature.members
      .find(_.name.toString.equalsIgnoreCase("isFormatted"))
      .map(l => im.reflectField(l.asTerm).get).getOrElse("false").asInstanceOf[Boolean]
    isFormatted
  }

  def createObject(className: String, conArgs: Object*): (Any, Class[_]) = {
    val clazz = Utils.classForName(className)
    val ctor = clazz.getConstructors.head
    ctor.setAccessible(true)
    (ctor.newInstance(conArgs: _*), clazz)
  }

}

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.util
19
20	import java.lang.reflect.Field
21
22	import scala.reflect.runtime._
23	import scala.reflect.runtime.universe._
24
25	import org.apache.spark.SPARK_VERSION
26	import org.apache.spark.SparkContext
27	import org.apache.spark.sql.SparkSession
28	import org.apache.spark.sql.catalyst.TableIdentifier
29	import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
30	import org.apache.spark.sql.catalyst.catalog.CatalogTable
31	import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
32	import org.apache.spark.sql.catalyst.parser.AstBuilder
33	import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, LogicalPlan, SubqueryAlias}
34	import org.apache.spark.sql.execution.datasources.LogicalRelation
35	import org.apache.spark.sql.sources.BaseRelation
36
37	import org.apache.carbondata.common.logging.LogServiceFactory
38	import org.apache.carbondata.core.constants.CarbonCommonConstants
39
40	/**
41	* Reflection APIs
42	*/
43
44	object CarbonReflectionUtils {
45	private val rm = universe.runtimeMirror(getClass.getClassLoader)	1✔
46
47	/**
48	* Returns the field val from a object through reflection.
49	* @param name - name of the field being retrieved.
50	* @param obj - Object from which the field has to be retrieved.
51	* @tparam T
52	* @return
53	*/
54	def getField[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
55	val im = rm.reflect(obj)	1✔
56	im.symbol.typeSignature.members.find(_.name.toString.equals(name))	1✔
57	.map(l => im.reflectField(l.asTerm).get).getOrElse(null)	1✔
58	}
59
60	def getUnresolvedRelation(
61	tableIdentifier: TableIdentifier,
62	tableAlias: Option[String] = None): UnresolvedRelation = {	1✔
63	val className = "org.apache.spark.sql.catalyst.analysis.UnresolvedRelation"	1✔
64	if (SPARK_VERSION.startsWith("2.1")) {	1✔
65	createObject(	1✔
66	className,	1✔
67	tableIdentifier,	1✔
68	tableAlias)._1.asInstanceOf[UnresolvedRelation]	1✔
69	} else if (SPARK_VERSION.startsWith("2.2")) {	×
70	createObject(	×
71	className,	×
72	tableIdentifier)._1.asInstanceOf[UnresolvedRelation]	×
73	} else {
74	throw new UnsupportedOperationException(s"Unsupported Spark version $SPARK_VERSION")	×
75	}
76	}
77
78	def getSubqueryAlias(sparkSession: SparkSession, alias: Option[String],
79	relation: LogicalPlan,
80	view: Option[TableIdentifier]): SubqueryAlias = {	1✔
81	val className = "org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias"	1✔
82	if (SPARK_VERSION.startsWith("2.1")) {	1✔
83	createObject(	1✔
84	className,	1✔
85	alias.getOrElse(""),	1✔
86	relation,	1✔
87	Option(view))._1.asInstanceOf[SubqueryAlias]	1✔
88	} else if (SPARK_VERSION.startsWith("2.2")) {	×
89	createObject(	×
90	className,	×
91	alias.getOrElse(""),	×
92	relation)._1.asInstanceOf[SubqueryAlias]	×
93	} else {
94	throw new UnsupportedOperationException("Unsupported Spark version")	×
95	}
96	}
97
98	def getInsertIntoCommand(table: LogicalPlan,
99	partition: Map[String, Option[String]],
100	query: LogicalPlan,
101	overwrite: Boolean,
102	ifPartitionNotExists: Boolean): InsertIntoTable = {	1✔
103	val className = "org.apache.spark.sql.catalyst.plans.logical.InsertIntoTable"	1✔
104	if (SPARK_VERSION.startsWith("2.1")) {	1✔
105	val overwriteOptions = createObject(	1✔
106	"org.apache.spark.sql.catalyst.plans.logical.OverwriteOptions",	1✔
107	overwrite.asInstanceOf[Object], Map.empty.asInstanceOf[Object])._1.asInstanceOf[Object]	1✔
108	createObject(	1✔
109	className,	1✔
110	table,	1✔
111	partition,	1✔
112	query,	1✔
113	overwriteOptions,	1✔
114	ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]	1✔
115	} else if (SPARK_VERSION.startsWith("2.2")) {	×
116	createObject(	×
117	className,	×
118	table,	×
119	partition,	×
120	query,	×
121	overwrite.asInstanceOf[Object],	×
122	ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]	×
123	} else {
124	throw new UnsupportedOperationException("Unsupported Spark version")	×
125	}
126	}
127
128	def getLogicalRelation(relation: BaseRelation,
129	expectedOutputAttributes: Seq[Attribute],
130	catalogTable: Option[CatalogTable]): LogicalRelation = {	1✔
131	val className = "org.apache.spark.sql.execution.datasources.LogicalRelation"	1✔
132	if (SPARK_VERSION.startsWith("2.1")) {	1✔
133	createObject(	1✔
134	className,	1✔
135	relation,	1✔
136	Some(expectedOutputAttributes),	1✔
137	catalogTable)._1.asInstanceOf[LogicalRelation]	1✔
138	} else if (SPARK_VERSION.startsWith("2.2")) {	×
139	createObject(	×
140	className,	×
141	relation,	×
142	expectedOutputAttributes,	×
143	catalogTable)._1.asInstanceOf[LogicalRelation]	×
144	} else {
145	throw new UnsupportedOperationException("Unsupported Spark version")	×
146	}
147	}
148
149
150	def getOverWriteOption[T: TypeTag : reflect.ClassTag](name: String, obj: T): Boolean = {
151	var overwriteboolean: Boolean = false	1✔
152	val im = rm.reflect(obj)	1✔
153	for (m <- typeOf[T].members.filter(!_.isMethod)) {	1✔
154	if (m.toString.contains("overwrite")) {	1✔
155	val typ = m.typeSignature	1✔
156	if (typ.toString.contains("Boolean")) {	1✔
157	// Spark2.2
158	overwriteboolean = im.reflectField(m.asTerm).get.asInstanceOf[Boolean]	×
159	} else {
160	overwriteboolean = getOverWrite("enabled", im.reflectField(m.asTerm).get)	1✔
161	}
162	}
163	}
164	overwriteboolean	1✔
165	}
166
167	private def getOverWrite[T: TypeTag : reflect.ClassTag](name: String, obj: T): Boolean = {
168	var overwriteboolean: Boolean = false	1✔
169	val im = rm.reflect(obj)	1✔
170	for (l <- im.symbol.typeSignature.members.filter(_.name.toString.contains("enabled"))) {	1✔
171	overwriteboolean = im.reflectField(l.asTerm).get.asInstanceOf[Boolean]	1✔
172	}
173	overwriteboolean	1✔
174	}
175
176	def getFieldOfCatalogTable[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
177	val im = rm.reflect(obj)	×
178	val sym = im.symbol.typeSignature.member(TermName(name))	×
179	val tableMeta = im.reflectMethod(sym.asMethod).apply()	×
180	tableMeta	×
181	}
182
183	def getAstBuilder(conf: Object,
184	sqlParser: Object,
185	sparkSession: SparkSession): AstBuilder = {
186	if (SPARK_VERSION.startsWith("2.1") \|\| SPARK_VERSION.startsWith("2.2")) {	1✔
187	val className = sparkSession.sparkContext.conf.get(	1✔
188	CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME,	1✔
189	"org.apache.spark.sql.hive.CarbonSqlAstBuilder")	1✔
190	createObject(className,	1✔
191	conf,	1✔
192	sqlParser, sparkSession)._1.asInstanceOf[AstBuilder]	1✔
193	} else {
194	throw new UnsupportedOperationException("Spark version not supported")	×
195	}
196	}
197
198	def getSessionState(sparkContext: SparkContext,
199	carbonSession: Object,
200	useHiveMetaStore: Boolean): Any = {
201	if (SPARK_VERSION.startsWith("2.1")) {	1✔
202	val className = sparkContext.conf.get(	1✔
203	CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,	1✔
204	"org.apache.spark.sql.hive.CarbonSessionState")	1✔
205	createObject(className, carbonSession)._1	1✔
206	} else if (SPARK_VERSION.startsWith("2.2")) {	×
207	if (useHiveMetaStore) {	×
208	val className = sparkContext.conf.get(	×
209	CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,	×
210	"org.apache.spark.sql.hive.CarbonSessionStateBuilder")	×
211	val tuple = createObject(className, carbonSession, None)	×
212	val method = tuple._2.getMethod("build")	×
213	method.invoke(tuple._1)	×
214	} else {
215	val className = sparkContext.conf.get(	×
216	CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,	×
217	"org.apache.spark.sql.hive.CarbonInMemorySessionStateBuilder")	×
218	val tuple = createObject(className, carbonSession, None)	×
219	val method = tuple._2.getMethod("build")	×
220	method.invoke(tuple._1)	×
221	}
222	} else {
223	throw new UnsupportedOperationException("Spark version not supported")	×
224	}
225	}
226
227	def hasPredicateSubquery(filterExp: Expression) : Boolean = {
228	if (SPARK_VERSION.startsWith("2.1")) {	1✔
229	val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.PredicateSubquery")	1✔
230	val method = tuple.getMethod("hasPredicateSubquery", classOf[Expression])	1✔
231	val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]	1✔
232	hasSubquery	1✔
233	} else if (SPARK_VERSION.startsWith("2.2")) {	×
234	val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.SubqueryExpression")	×
235	val method = tuple.getMethod("hasInOrExistsSubquery", classOf[Expression])	×
236	val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]	×
237	hasSubquery	×
238	} else {
239	throw new UnsupportedOperationException("Spark version not supported")	×
240	}
241	}
242
243	def getDescribeTableFormattedField[T: TypeTag : reflect.ClassTag](obj: T): Boolean = {
244	val im = rm.reflect(obj)	1✔
245	val isFormatted = im.symbol.typeSignature.members	1✔
246	.find(_.name.toString.equalsIgnoreCase("isFormatted"))	1✔
247	.map(l => im.reflectField(l.asTerm).get).getOrElse("false").asInstanceOf[Boolean]	1✔
248	isFormatted	1✔
249	}
250
251	def createObject(className: String, conArgs: Object*): (Any, Class[_]) = {
252	val clazz = Utils.classForName(className)	1✔
253	val ctor = clazz.getConstructors.head	1✔
254	ctor.setAccessible(true)	1✔
255	(ctor.newInstance(conArgs: _*), clazz)	1✔
256	}
257
258	}	1✔

apache / carbondata / 8253

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