10738

Build # 10738

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/1016a7a7fff946224b638a5ad00f46482f1ba9b6">1016a7a7f">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/1016a7a7fff946224b638a5ad00f46482f1ba9b6">1016a7a7f</a><a href="https://github.com/apache/carbondata/commit/1016a7a7fff946224b638a5ad00f46482f1ba9b6">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/1016a7a7fff946224b638a5ad00f46482f1ba9b6&quot;&gt;1016a7a7f&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/1016a7a7fff946224b638a5ad00f46482f1ba9b6&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

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

62424 of 78800 relevant lines covered (79.22%)

1.05 hits per line

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

59.49

/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.Method

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

import org.apache.spark.{SPARK_VERSION, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.{InternalRow, 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.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.execution.{RowDataSourceScanExec, SparkPlan}
import org.apache.spark.sql.execution.command.RunnableCommand
import org.apache.spark.sql.execution.datasources.{DataSource, LogicalRelation}
import org.apache.spark.sql.internal.HiveSerDe
import org.apache.spark.sql.sources.{BaseRelation, Filter}
import org.apache.spark.sql.types.StructField

import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.hadoop.api.{CarbonTableInputFormat, CarbonTableOutputFormat}

/**
 * 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 (SparkUtil.isSparkVersionEqualTo("2.1")) {
      createObject(
        className,
        tableIdentifier,
        tableAlias)._1.asInstanceOf[UnresolvedRelation]
    } else if (SparkUtil.isSparkVersionXandAbove("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 (SparkUtil.isSparkVersionEqualTo("2.1")) {
      createObject(
        className,
        alias.getOrElse(""),
        relation,
        Option(view))._1.asInstanceOf[SubqueryAlias]
    } else if (SparkUtil.isSparkVersionXandAbove("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 (SparkUtil.isSparkVersionEqualTo("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 (SparkUtil.isSparkVersionXandAbove("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],
      isStreaming: Boolean): LogicalRelation = {
    val className = "org.apache.spark.sql.execution.datasources.LogicalRelation"
    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
      createObject(
        className,
        relation,
        Some(expectedOutputAttributes),
        catalogTable)._1.asInstanceOf[LogicalRelation]
    } else if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      createObject(
        className,
        relation,
        expectedOutputAttributes,
        catalogTable)._1.asInstanceOf[LogicalRelation]
    } else if (SparkUtil.isSparkVersionEqualTo("2.3")) {
      createObject(
        className,
        relation,
        expectedOutputAttributes,
        catalogTable,
        isStreaming.asInstanceOf[Object])._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 = {
    val className = sparkSession.sparkContext.conf.get(
      CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME,
      CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME_DEFAULT)
    createObject(className,
      conf,
      sqlParser, sparkSession)._1.asInstanceOf[AstBuilder]
  }

  def getSessionState(sparkContext: SparkContext,
      carbonSession: Object,
      useHiveMetaStore: Boolean): Any = {
    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
      val className = sparkContext.conf.get(
        CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
        "org.apache.spark.sql.hive.CarbonSessionState")
      createObject(className, carbonSession)._1
    } else if (SparkUtil.isSparkVersionXandAbove("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 (SparkUtil.isSparkVersionEqualTo("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 (SparkUtil.isSparkVersionXandAbove("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 getRowDataSourceScanExecObj(relation: LogicalRelation,
      output: Seq[Attribute],
      pushedFilters: Seq[Filter],
      handledFilters: Seq[Filter],
      rdd: RDD[InternalRow],
      partition: Partitioning,
      metadata: Map[String, String]): RowDataSourceScanExec = {
    val className = "org.apache.spark.sql.execution.RowDataSourceScanExec"
    if (SparkUtil.isSparkVersionEqualTo("2.1") || SparkUtil.isSparkVersionEqualTo("2.2")) {
      createObject(className, output, rdd, relation.relation,
        partition, metadata,
        relation.catalogTable.map(_.identifier))._1.asInstanceOf[RowDataSourceScanExec]
    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
      createObject(className, output, output.map(output.indexOf),
        pushedFilters.toSet, handledFilters.toSet, rdd,
        relation.relation,
        relation.catalogTable.map(_.identifier))._1.asInstanceOf[RowDataSourceScanExec]
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def invokewriteAndReadMethod(dataSourceObj: DataSource,
      dataFrame: DataFrame,
      data: LogicalPlan,
      session: SparkSession,
      mode: SaveMode,
      query: LogicalPlan,
      physicalPlan: SparkPlan): BaseRelation = {
    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      val method: Method = dataSourceObj.getClass
        .getMethod("writeAndRead", classOf[SaveMode], classOf[DataFrame])
      method.invoke(dataSourceObj, mode, dataFrame)
        .asInstanceOf[BaseRelation]
    } else if (SparkUtil.isSparkVersionEqualTo("2.3")) {
      val method: Method = dataSourceObj.getClass
        .getMethod("writeAndRead",
          classOf[SaveMode],
          classOf[LogicalPlan],
          classOf[Seq[String]],
          classOf[SparkPlan])
      // since spark 2.3.2 version (SPARK-PR#22346),
      // change 'query.output' to 'query.output.map(_.name)'
      method.invoke(dataSourceObj, mode, query, query.output.map(_.name), physicalPlan)
        .asInstanceOf[BaseRelation]
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  /**
   * method to invoke alter table add columns for hive table from carbon session
   * @param table
   * @param colsToAdd
   * @return
   */
  def invokeAlterTableAddColumn(table: TableIdentifier,
      colsToAdd: Seq[StructField]): Object = {
    val caseClassName = "org.apache.spark.sql.execution.command.AlterTableAddColumnsCommand"
    CarbonReflectionUtils.createObject(caseClassName, table, colsToAdd)
      ._1.asInstanceOf[RunnableCommand]
  }

  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)
  }

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

  /**
   * It is a hack to update the carbon input, output format information to #HiveSerDe
   */
  def updateCarbonSerdeInfo(): Unit = {
    val currentMirror = scala.reflect.runtime.currentMirror
    val instanceMirror = currentMirror.reflect(HiveSerDe)
    currentMirror.staticClass(HiveSerDe.getClass.getName).
      toType.members.find { p =>
      !p.isMethod && p.name.toString.equals("serdeMap")
    } match {
      case Some(field) =>
        val serdeMap =
          instanceMirror.reflectField(field.asTerm).get.asInstanceOf[Map[String, HiveSerDe]]
        val updatedSerdeMap =
          serdeMap ++ Map[String, HiveSerDe](
            ("org.apache.spark.sql.carbonsource", HiveSerDe(Some(
              classOf[CarbonTableInputFormat[_]].getName),
              Some(classOf[CarbonTableOutputFormat].getName))),
            ("carbon", HiveSerDe(Some(
              classOf[CarbonTableInputFormat[_]].getName),
              Some(classOf[CarbonTableOutputFormat].getName))),
            ("carbondata", HiveSerDe(Some(
              classOf[CarbonTableInputFormat[_]].getName),
              Some(classOf[CarbonTableOutputFormat].getName))))
        instanceMirror.reflectField(field.asTerm).set(updatedSerdeMap)
      case _ =>
    }
  }

  /**
   * This method updates the field of case class through reflection.
   */
  def setFieldToCaseClass(caseObj: Object, fieldName: String, objToSet: Object): Unit = {
    val nameField = caseObj.getClass.getDeclaredField(fieldName)
    nameField.setAccessible(true)
    nameField.set(caseObj, objToSet)
  }
}

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.Method
21
22	import scala.reflect.runtime._
23	import scala.reflect.runtime.universe._
24
25	import org.apache.spark.{SPARK_VERSION, SparkContext}
26	import org.apache.spark.rdd.RDD
27	import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
28	import org.apache.spark.sql.catalyst.{InternalRow, 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.catalyst.plans.physical.Partitioning
35	import org.apache.spark.sql.execution.{RowDataSourceScanExec, SparkPlan}
36	import org.apache.spark.sql.execution.command.RunnableCommand
37	import org.apache.spark.sql.execution.datasources.{DataSource, LogicalRelation}
38	import org.apache.spark.sql.internal.HiveSerDe
39	import org.apache.spark.sql.sources.{BaseRelation, Filter}
40	import org.apache.spark.sql.types.StructField
41
42	import org.apache.carbondata.core.constants.CarbonCommonConstants
43	import org.apache.carbondata.hadoop.api.{CarbonTableInputFormat, CarbonTableOutputFormat}
44
45	/**
46	* Reflection APIs
47	*/
48
49	object CarbonReflectionUtils {
50	private val rm = universe.runtimeMirror(getClass.getClassLoader)	1✔
51
52	/**
53	* Returns the field val from a object through reflection.
54	* @param name - name of the field being retrieved.
55	* @param obj - Object from which the field has to be retrieved.
56	* @tparam T
57	* @return
58	*/
59	def getField[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
60	val im = rm.reflect(obj)	×
61	im.symbol.typeSignature.members.find(_.name.toString.equals(name))	×
62	.map(l => im.reflectField(l.asTerm).get).getOrElse(null)	×
63	}
64
65	def getUnresolvedRelation(
66	tableIdentifier: TableIdentifier,
67	tableAlias: Option[String] = None): UnresolvedRelation = {	1✔
68	val className = "org.apache.spark.sql.catalyst.analysis.UnresolvedRelation"	1✔
69	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	1✔
70	createObject(	×
71	className,	×
72	tableIdentifier,	×
73	tableAlias)._1.asInstanceOf[UnresolvedRelation]	×
74	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	1✔
75	createObject(	1✔
76	className,	1✔
77	tableIdentifier)._1.asInstanceOf[UnresolvedRelation]	1✔
78	} else {
79	throw new UnsupportedOperationException(s"Unsupported Spark version $SPARK_VERSION")	×
80	}
81	}
82
83	def getSubqueryAlias(sparkSession: SparkSession, alias: Option[String],
84	relation: LogicalPlan,
85	view: Option[TableIdentifier]): SubqueryAlias = {	1✔
86	val className = "org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias"	1✔
87	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	1✔
88	createObject(	×
89	className,	×
90	alias.getOrElse(""),	×
91	relation,	×
92	Option(view))._1.asInstanceOf[SubqueryAlias]	×
93	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	1✔
94	createObject(	1✔
95	className,	1✔
96	alias.getOrElse(""),	1✔
97	relation)._1.asInstanceOf[SubqueryAlias]	1✔
98	} else {
99	throw new UnsupportedOperationException("Unsupported Spark version")	×
100	}
101	}
102
103	def getInsertIntoCommand(table: LogicalPlan,
104	partition: Map[String, Option[String]],
105	query: LogicalPlan,
106	overwrite: Boolean,
107	ifPartitionNotExists: Boolean): InsertIntoTable = {	1✔
108	val className = "org.apache.spark.sql.catalyst.plans.logical.InsertIntoTable"	1✔
109	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	1✔
110	val overwriteOptions = createObject(	×
111	"org.apache.spark.sql.catalyst.plans.logical.OverwriteOptions",	×
112	overwrite.asInstanceOf[Object], Map.empty.asInstanceOf[Object])._1.asInstanceOf[Object]	×
113	createObject(	×
114	className,	×
115	table,	×
116	partition,	×
117	query,	×
118	overwriteOptions,	×
119	ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]	×
120	} else if (SparkUtil.isSparkVersionXandAbove("2.2") ) {	1✔
121	createObject(	1✔
122	className,	1✔
123	table,	1✔
124	partition,	1✔
125	query,	1✔
126	overwrite.asInstanceOf[Object],	1✔
127	ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]	1✔
128	} else {
129	throw new UnsupportedOperationException("Unsupported Spark version")	×
130	}
131	}
132
133	def getLogicalRelation(relation: BaseRelation,
134	expectedOutputAttributes: Seq[Attribute],
135	catalogTable: Option[CatalogTable],
136	isStreaming: Boolean): LogicalRelation = {	1✔
137	val className = "org.apache.spark.sql.execution.datasources.LogicalRelation"	1✔
138	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	1✔
139	createObject(	×
140	className,	×
141	relation,	×
142	Some(expectedOutputAttributes),	×
143	catalogTable)._1.asInstanceOf[LogicalRelation]	×
144	} else if (SparkUtil.isSparkVersionEqualTo("2.2")) {	1✔
145	createObject(	×
146	className,	×
147	relation,	×
148	expectedOutputAttributes,	×
149	catalogTable)._1.asInstanceOf[LogicalRelation]	×
150	} else if (SparkUtil.isSparkVersionEqualTo("2.3")) {	1✔
151	createObject(	1✔
152	className,	1✔
153	relation,	1✔
154	expectedOutputAttributes,	1✔
155	catalogTable,	1✔
156	isStreaming.asInstanceOf[Object])._1.asInstanceOf[LogicalRelation]	1✔
157	} else {
158	throw new UnsupportedOperationException("Unsupported Spark version")	×
159	}
160	}
161
162
163	def getOverWriteOption[T: TypeTag : reflect.ClassTag](name: String, obj: T): Boolean = {
164	var overwriteboolean: Boolean = false	1✔
165	val im = rm.reflect(obj)	1✔
166	for (m <- typeOf[T].members.filter(!_.isMethod)) {	1✔
167	if (m.toString.contains("overwrite")) {	1✔
168	val typ = m.typeSignature	1✔
169	if (typ.toString.contains("Boolean")) {	1✔
170	// Spark2.2
171	overwriteboolean = im.reflectField(m.asTerm).get.asInstanceOf[Boolean]	1✔
172	} else {
173	overwriteboolean = getOverWrite("enabled", im.reflectField(m.asTerm).get)	×
174	}
175	}
176	}
177	overwriteboolean	1✔
178	}
179
180	private def getOverWrite[T: TypeTag : reflect.ClassTag](name: String, obj: T): Boolean = {
181	var overwriteboolean: Boolean = false	×
182	val im = rm.reflect(obj)	×
183	for (l <- im.symbol.typeSignature.members.filter(_.name.toString.contains("enabled"))) {	×
184	overwriteboolean = im.reflectField(l.asTerm).get.asInstanceOf[Boolean]	×
185	}
186	overwriteboolean	×
187	}
188
189	def getFieldOfCatalogTable[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
190	val im = rm.reflect(obj)	1✔
191	val sym = im.symbol.typeSignature.member(TermName(name))	1✔
192	val tableMeta = im.reflectMethod(sym.asMethod).apply()	1✔
193	tableMeta	1✔
194	}
195
196	def getAstBuilder(conf: Object,
197	sqlParser: Object,
198	sparkSession: SparkSession): AstBuilder = {
199	val className = sparkSession.sparkContext.conf.get(	1✔
200	CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME,	1✔
201	CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME_DEFAULT)	1✔
202	createObject(className,	1✔
203	conf,	1✔
204	sqlParser, sparkSession)._1.asInstanceOf[AstBuilder]	1✔
205	}
206
207	def getSessionState(sparkContext: SparkContext,
208	carbonSession: Object,
209	useHiveMetaStore: Boolean): Any = {
210	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	1✔
211	val className = sparkContext.conf.get(	×
212	CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,	×
213	"org.apache.spark.sql.hive.CarbonSessionState")	×
214	createObject(className, carbonSession)._1	×
215	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	1✔
216	if (useHiveMetaStore) {	1✔
217	val className = sparkContext.conf.get(	1✔
218	CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,	1✔
219	"org.apache.spark.sql.hive.CarbonSessionStateBuilder")	1✔
220	val tuple = createObject(className, carbonSession, None)	1✔
221	val method = tuple._2.getMethod("build")	1✔
222	method.invoke(tuple._1)	1✔
223	} else {
224	val className = sparkContext.conf.get(	×
225	CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,	×
226	"org.apache.spark.sql.hive.CarbonInMemorySessionStateBuilder")	×
227	val tuple = createObject(className, carbonSession, None)	×
228	val method = tuple._2.getMethod("build")	×
229	method.invoke(tuple._1)	×
230	}
231	} else {
232	throw new UnsupportedOperationException("Spark version not supported")	×
233	}
234	}
235
236	def hasPredicateSubquery(filterExp: Expression) : Boolean = {
237	if (SparkUtil.isSparkVersionEqualTo("2.1")) {	1✔
238	val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.PredicateSubquery")	×
239	val method = tuple.getMethod("hasPredicateSubquery", classOf[Expression])	×
240	val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]	×
241	hasSubquery	×
242	} else if (SparkUtil.isSparkVersionXandAbove("2.2")) {	1✔
243	val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.SubqueryExpression")	1✔
244	val method = tuple.getMethod("hasInOrExistsSubquery", classOf[Expression])	1✔
245	val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]	1✔
246	hasSubquery	1✔
247	} else {
248	throw new UnsupportedOperationException("Spark version not supported")	×
249	}
250	}
251
252	def getDescribeTableFormattedField[T: TypeTag : reflect.ClassTag](obj: T): Boolean = {
253	val im = rm.reflect(obj)	×
254	val isFormatted = im.symbol.typeSignature.members	×
255	.find(_.name.toString.equalsIgnoreCase("isFormatted"))	×
256	.map(l => im.reflectField(l.asTerm).get).getOrElse("false").asInstanceOf[Boolean]	×
257	isFormatted	×
258	}
259
260	def getRowDataSourceScanExecObj(relation: LogicalRelation,
261	output: Seq[Attribute],
262	pushedFilters: Seq[Filter],
263	handledFilters: Seq[Filter],
264	rdd: RDD[InternalRow],
265	partition: Partitioning,
266	metadata: Map[String, String]): RowDataSourceScanExec = {	1✔
267	val className = "org.apache.spark.sql.execution.RowDataSourceScanExec"	1✔
268	if (SparkUtil.isSparkVersionEqualTo("2.1") \|\| SparkUtil.isSparkVersionEqualTo("2.2")) {	1✔
269	createObject(className, output, rdd, relation.relation,	×
270	partition, metadata,	×
271	relation.catalogTable.map(_.identifier))._1.asInstanceOf[RowDataSourceScanExec]	×
272	} else if (SparkUtil.isSparkVersionXandAbove("2.3")) {	1✔
273	createObject(className, output, output.map(output.indexOf),	1✔
274	pushedFilters.toSet, handledFilters.toSet, rdd,	1✔
275	relation.relation,	1✔
276	relation.catalogTable.map(_.identifier))._1.asInstanceOf[RowDataSourceScanExec]	1✔
277	} else {
278	throw new UnsupportedOperationException("Spark version not supported")	×
279	}
280	}
281
282	def invokewriteAndReadMethod(dataSourceObj: DataSource,
283	dataFrame: DataFrame,
284	data: LogicalPlan,
285	session: SparkSession,
286	mode: SaveMode,
287	query: LogicalPlan,
288	physicalPlan: SparkPlan): BaseRelation = {
289	if (SparkUtil.isSparkVersionEqualTo("2.2")) {	1✔
290	val method: Method = dataSourceObj.getClass	×
291	.getMethod("writeAndRead", classOf[SaveMode], classOf[DataFrame])	×
292	method.invoke(dataSourceObj, mode, dataFrame)	×
293	.asInstanceOf[BaseRelation]
294	} else if (SparkUtil.isSparkVersionEqualTo("2.3")) {	1✔
295	val method: Method = dataSourceObj.getClass	1✔
296	.getMethod("writeAndRead",	1✔
297	classOf[SaveMode],	1✔
298	classOf[LogicalPlan],	1✔
299	classOf[Seq[String]],	1✔
300	classOf[SparkPlan])	1✔
301	// since spark 2.3.2 version (SPARK-PR#22346),
302	// change 'query.output' to 'query.output.map(_.name)'
303	method.invoke(dataSourceObj, mode, query, query.output.map(_.name), physicalPlan)	1✔
304	.asInstanceOf[BaseRelation]
305	} else {
306	throw new UnsupportedOperationException("Spark version not supported")	×
307	}
308	}
309
310	/**
311	* method to invoke alter table add columns for hive table from carbon session
312	* @param table
313	* @param colsToAdd
314	* @return
315	*/
316	def invokeAlterTableAddColumn(table: TableIdentifier,
317	colsToAdd: Seq[StructField]): Object = {
318	val caseClassName = "org.apache.spark.sql.execution.command.AlterTableAddColumnsCommand"	×
319	CarbonReflectionUtils.createObject(caseClassName, table, colsToAdd)	×
320	._1.asInstanceOf[RunnableCommand]	×
321	}
322
323	def createObject(className: String, conArgs: Object*): (Any, Class[_]) = {
324	val clazz = Utils.classForName(className)	1✔
325	val ctor = clazz.getConstructors.head	1✔
326	ctor.setAccessible(true)	1✔
327	(ctor.newInstance(conArgs: _*), clazz)	1✔
328	}
329
330	def createObjectOfPrivateConstructor(className: String, conArgs: Object*): (Any, Class[_]) = {
331	val clazz = Utils.classForName(className)	×
332	val ctor = clazz.getDeclaredConstructors.head	×
333	ctor.setAccessible(true)	×
334	(ctor.newInstance(conArgs: _*), clazz)	×
335	}
336
337	/**
338	* It is a hack to update the carbon input, output format information to #HiveSerDe
339	*/
340	def updateCarbonSerdeInfo(): Unit = {	1✔
341	val currentMirror = scala.reflect.runtime.currentMirror	1✔
342	val instanceMirror = currentMirror.reflect(HiveSerDe)	1✔
343	currentMirror.staticClass(HiveSerDe.getClass.getName).	1✔
344	toType.members.find { p =>	1✔
345	!p.isMethod && p.name.toString.equals("serdeMap")	1✔
346	} match {
347	case Some(field) =>	1✔
348	val serdeMap =	1✔
349	instanceMirror.reflectField(field.asTerm).get.asInstanceOf[Map[String, HiveSerDe]]	1✔
350	val updatedSerdeMap =
351	serdeMap ++ Map[String, HiveSerDe](	1✔
352	("org.apache.spark.sql.carbonsource", HiveSerDe(Some(	1✔
353	classOf[CarbonTableInputFormat[_]].getName),	1✔
354	Some(classOf[CarbonTableOutputFormat].getName))),	1✔
355	("carbon", HiveSerDe(Some(	1✔
356	classOf[CarbonTableInputFormat[_]].getName),	1✔
357	Some(classOf[CarbonTableOutputFormat].getName))),	1✔
358	("carbondata", HiveSerDe(Some(	1✔
359	classOf[CarbonTableInputFormat[_]].getName),	1✔
360	Some(classOf[CarbonTableOutputFormat].getName))))	1✔
361	instanceMirror.reflectField(field.asTerm).set(updatedSerdeMap)	1✔
362	case _ =>	×
363	}
364	}
365
366	/**
367	* This method updates the field of case class through reflection.
368	*/
369	def setFieldToCaseClass(caseObj: Object, fieldName: String, objToSet: Object): Unit = {
370	val nameField = caseObj.getClass.getDeclaredField(fieldName)	1✔
371	nameField.setAccessible(true)	1✔
372	nameField.set(caseObj, objToSet)	1✔
373	}
374	}	1✔

apache / carbondata / 10738

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