3929

Build # 3929

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/1d6484f0d77cbabe713cbd738043fd2749888305">1d6484f0d<a href="https://github.com/apache/carbondata/commit/1d6484f0d77cbabe713cbd738043fd2749888305">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1b568e29aab49b1578a2f07a2fcdbaaf9b9a2bf1">1b568e29a</a>">1b568e29a</a><a href="https://github.com/apache/carbondata/commit/1d6484f0d77cbabe713cbd738043fd2749888305"> into 188e7e4d4">188e7e4d4</a>

Run Details

26 of 26 new or added lines in 10 files covered. (100.0%)

64950 of 83531 relevant lines covered (77.76%)

1.02 hits per line

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

4.55

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

import java.lang.reflect.Method

import org.apache.hadoop.conf.Configuration
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.catalyst.analysis.EmptyRule
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeSeq, Cast, Expression, NamedExpression}
import org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LogicalPlan, Statistics}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.internal.{SessionState, SQLConf}
import org.apache.spark.util.{CarbonReflectionUtils, SerializableConfiguration, SparkUtil, Utils}

object SparkSQLUtil {
  def sessionState(sparkSession: SparkSession): SessionState = sparkSession.sessionState

  def execute(logicalPlan: LogicalPlan, sparkSession: SparkSession): DataFrame = {
    Dataset.ofRows(sparkSession, logicalPlan)
  }

  def getSparkSession: SparkSession = {
    SparkSession.getDefaultSession.get
  }

  def invokeStatsMethod(logicalPlanObj: LogicalPlan, conf: SQLConf): Statistics = {
    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      val method: Method = logicalPlanObj.getClass.getMethod("stats", classOf[SQLConf])
      method.invoke(logicalPlanObj, conf).asInstanceOf[Statistics]
    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
      val method: Method = logicalPlanObj.getClass.getMethod("stats")
      method.invoke(logicalPlanObj).asInstanceOf[Statistics]
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def invokeQueryPlannormalizeExprId(r: NamedExpression, input: AttributeSeq)
      : NamedExpression = {
    if (SparkUtil.isSparkVersionXandAbove("2.2")) {
      val clazz = Utils.classForName("org.apache.spark.sql.catalyst.plans.QueryPlan")
      clazz.getDeclaredMethod("normalizeExprId", classOf[Any], classOf[AttributeSeq]).
        invoke(null, r, input).asInstanceOf[NamedExpression]
    } else {
      r
    }
  }

  def getStatisticsObj(outputList: Seq[NamedExpression],
                       plan: LogicalPlan, stats: Statistics,
                       aliasMap: Option[AttributeMap[Attribute]] = None)
  : Statistics = {
    val className = "org.apache.spark.sql.catalyst.plans.logical.Statistics"
    if (SparkUtil.isSparkVersionXandAbove("2.2")) {
      val output = outputList.map(_.toAttribute)
      val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {
        table => AttributeMap(table.output.zip(output))
      }
      val rewrites = mapSeq.head
      val attributes : AttributeMap[ColumnStat] = CarbonReflectionUtils.
        getField("attributeStats", stats).asInstanceOf[AttributeMap[ColumnStat]]
      var attributeStats = AttributeMap(attributes.iterator
        .map { pair => (rewrites(pair._1), pair._2) }.toSeq)
      if (aliasMap.isDefined) {
        attributeStats = AttributeMap(
          attributeStats.map(pair => (aliasMap.get(pair._1), pair._2)).toSeq)
      }
      val hints = CarbonReflectionUtils.getField("hints", stats).asInstanceOf[Object]
      CarbonReflectionUtils.createObject(className, stats.sizeInBytes,
        stats.rowCount, attributeStats, hints).asInstanceOf[Statistics]
    } else {
      val output = outputList.map(_.name)
      val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {
        table => table.output.map(_.name).zip(output).toMap
      }
      val rewrites = mapSeq.head
      val colStats = CarbonReflectionUtils.getField("colStats", stats)
        .asInstanceOf[Map[String, ColumnStat]]
      var attributeStats = colStats.iterator
        .map { pair => (rewrites(pair._1), pair._2) }.toMap
      if (aliasMap.isDefined) {
        val aliasMapName = aliasMap.get.map(x => (x._1.name, x._2.name))
        attributeStats =
          attributeStats.map(pair => (aliasMapName.getOrElse(pair._1, pair._1)
            , pair._2))
      }
      CarbonReflectionUtils.createObject(className, stats.sizeInBytes,
        stats.rowCount, attributeStats).asInstanceOf[Statistics]
    }
  }

  def getEliminateViewObj(): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionXandAbove("2.2")) {
      val className = "org.apache.spark.sql.catalyst.analysis.EliminateView"
      CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]
    } else {
      EmptyRule
    }
  }

  def getPullupCorrelatedPredicatesObj(): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionXandAbove("2.2")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.PullupCorrelatedPredicates"
      CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]
    } else {
      EmptyRule
    }
  }

  def getRemoveRedundantAliasesObj(): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionXandAbove("2.2")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases"
      CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]
    } else {
      EmptyRule
    }
  }

  def getReorderJoinObj(conf: SQLConf): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin";
      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionEqualTo("2.1")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    }
    else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def getEliminateOuterJoinObj(conf: SQLConf): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin";
      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionEqualTo("2.1")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    }
    else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def getNullPropagationObj(conf: SQLConf): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation";
      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionEqualTo("2.1")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    } else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  def getCheckCartesianProductsObj(conf: SQLConf): Rule[LogicalPlan] = {
    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts";
      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts$";
      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
        .asInstanceOf[Rule[LogicalPlan]]
    } else if (SparkUtil.isSparkVersionEqualTo("2.1")) {
      val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts";
      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
    }
    else {
      throw new UnsupportedOperationException("Spark version not supported")
    }
  }

  /**
   * Method to broadcast a variable using spark SerializableConfiguration class
   *
   * @param sparkContext
   * @param hadoopConf
   * @return
   */
  def broadCastHadoopConf(sparkContext: SparkContext,
      hadoopConf: Configuration): Broadcast[SerializableConfiguration] = {
    sparkContext.broadcast(getSerializableConfigurableInstance(hadoopConf))
  }

  def getSerializableConfigurableInstance(hadoopConf: Configuration): SerializableConfiguration = {
    new SerializableConfiguration(hadoopConf)
  }

  /**
   * Get the task group id
   *
   * @param sparkSession
   * @return
   */
  def getTaskGroupId(sparkSession: SparkSession): String = {
    val taskGroupId = sparkSession.sparkContext.getLocalProperty("spark.jobGroup.id") match {
      case null => ""
      case _ => sparkSession.sparkContext.getLocalProperty("spark.jobGroup.id")
    }
    taskGroupId
  }

  /**
   * Get the task group description
   *
   * @param sparkSession
   * @return
   */
  def getTaskGroupDesc(sparkSession: SparkSession): String = {
    val taskGroupDesc = sparkSession.sparkContext.getLocalProperty("spark.job.description") match {
      case null => ""
      case _ => sparkSession.sparkContext.getLocalProperty("spark.job.description")
    }
    taskGroupDesc
  }

}

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.util
19
20	import java.lang.reflect.Method
21
22	import org.apache.hadoop.conf.Configuration
23	import org.apache.spark.SparkContext
24	import org.apache.spark.broadcast.Broadcast
25	import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
26	import org.apache.spark.sql.catalyst.analysis.EmptyRule
27	import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeSeq, Cast, Expression, NamedExpression}
28	import org.apache.spark.sql.catalyst.plans.logical
29	import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LogicalPlan, Statistics}
30	import org.apache.spark.sql.catalyst.rules.Rule
31	import org.apache.spark.sql.internal.{SessionState, SQLConf}
32	import org.apache.spark.util.{CarbonReflectionUtils, SerializableConfiguration, SparkUtil, Utils}
33
34	object SparkSQLUtil {
35	def sessionState(sparkSession: SparkSession): SessionState = sparkSession.sessionState	1✔
36
37	def execute(logicalPlan: LogicalPlan, sparkSession: SparkSession): DataFrame = {
38	Dataset.ofRows(sparkSession, logicalPlan)	×
39	}
40
41	def getSparkSession: SparkSession = {
42	SparkSession.getDefaultSession.get	1✔
43	}
44
45	def invokeStatsMethod(logicalPlanObj: LogicalPlan, conf: SQLConf): Statistics = {
46	if (SparkUtil.isSparkVersionEqualTo("2.2")) {	×
47	val method: Method = logicalPlanObj.getClass.getMethod("stats", classOf[SQLConf])	×
48	method.invoke(logicalPlanObj, conf).asInstanceOf[Statistics]	×
49	} else if (SparkUtil.isSparkVersionXandAbove("2.3")) {	×
50	val method: Method = logicalPlanObj.getClass.getMethod("stats")	×
51	method.invoke(logicalPlanObj).asInstanceOf[Statistics]	×
52	} else {
53	throw new UnsupportedOperationException("Spark version not supported")	×
54	}
55	}
56
57	def invokeQueryPlannormalizeExprId(r: NamedExpression, input: AttributeSeq)
58	: NamedExpression = {
59	if (SparkUtil.isSparkVersionXandAbove("2.2")) {	×
60	val clazz = Utils.classForName("org.apache.spark.sql.catalyst.plans.QueryPlan")	×
61	clazz.getDeclaredMethod("normalizeExprId", classOf[Any], classOf[AttributeSeq]).	×
62	invoke(null, r, input).asInstanceOf[NamedExpression]	×
63	} else {
64	r	×
65	}
66	}
67
68	def getStatisticsObj(outputList: Seq[NamedExpression],
69	plan: LogicalPlan, stats: Statistics,
70	aliasMap: Option[AttributeMap[Attribute]] = None)	×
71	: Statistics = {	×
72	val className = "org.apache.spark.sql.catalyst.plans.logical.Statistics"	×
73	if (SparkUtil.isSparkVersionXandAbove("2.2")) {	×
74	val output = outputList.map(_.toAttribute)	×
75	val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {	×
76	table => AttributeMap(table.output.zip(output))	×
77	}
78	val rewrites = mapSeq.head	×
79	val attributes : AttributeMap[ColumnStat] = CarbonReflectionUtils.	×
80	getField("attributeStats", stats).asInstanceOf[AttributeMap[ColumnStat]]	×
81	var attributeStats = AttributeMap(attributes.iterator	×
82	.map { pair => (rewrites(pair._1), pair._2) }.toSeq)	×
83	if (aliasMap.isDefined) {	×
84	attributeStats = AttributeMap(	×
85	attributeStats.map(pair => (aliasMap.get(pair._1), pair._2)).toSeq)	×
86	}
87	val hints = CarbonReflectionUtils.getField("hints", stats).asInstanceOf[Object]	×
88	CarbonReflectionUtils.createObject(className, stats.sizeInBytes,	×
89	stats.rowCount, attributeStats, hints).asInstanceOf[Statistics]	×
90	} else {
91	val output = outputList.map(_.name)	×
92	val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {	×
93	table => table.output.map(_.name).zip(output).toMap	×
94	}
95	val rewrites = mapSeq.head	×
96	val colStats = CarbonReflectionUtils.getField("colStats", stats)	×
97	.asInstanceOf[Map[String, ColumnStat]]
98	var attributeStats = colStats.iterator	×
99	.map { pair => (rewrites(pair._1), pair._2) }.toMap	×
100	if (aliasMap.isDefined) {	×
101	val aliasMapName = aliasMap.get.map(x => (x._1.name, x._2.name))	×
102	attributeStats =	×
103	attributeStats.map(pair => (aliasMapName.getOrElse(pair._1, pair._1)	×
104	, pair._2))	×
105	}
106	CarbonReflectionUtils.createObject(className, stats.sizeInBytes,	×
107	stats.rowCount, attributeStats).asInstanceOf[Statistics]	×
108	}
109	}
110
111	def getEliminateViewObj(): Rule[LogicalPlan] = {
112	if (SparkUtil.isSparkVersionXandAbove("2.2")) {	×
113	val className = "org.apache.spark.sql.catalyst.analysis.EliminateView"	×
114	CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]	×
115	} else {
116	EmptyRule	×
117	}
118	}
119
120	def getPullupCorrelatedPredicatesObj(): Rule[LogicalPlan] = {
121	if (SparkUtil.isSparkVersionXandAbove("2.2")) {	×
122	val className = "org.apache.spark.sql.catalyst.optimizer.PullupCorrelatedPredicates"	×
123	CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]	×
124	} else {
125	EmptyRule	×
126	}
127	}
128
129	def getRemoveRedundantAliasesObj(): Rule[LogicalPlan] = {
130	if (SparkUtil.isSparkVersionXandAbove("2.2")) {	×
131	val className = "org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases"	×
132	CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]	×
133	} else {
134	EmptyRule	×
135	}
136	}
137
138	def getReorderJoinObj(conf: SQLConf): Rule[LogicalPlan] = {
139	if (SparkUtil.isSparkVersionEqualTo("2.2")) {	×
140	val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin";	×
141	CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]	×
142	} else if (SparkUtil.isSparkVersionXandAbove("2.3")) {	×
143	val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin$";	×
144	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
145	.asInstanceOf[Rule[LogicalPlan]]
146	} else if (SparkUtil.isSparkVersionEqualTo("2.1")) {	×
147	val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin$";	×
148	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
149	.asInstanceOf[Rule[LogicalPlan]]
150	}
151	else {
152	throw new UnsupportedOperationException("Spark version not supported")	×
153	}
154	}
155
156	def getEliminateOuterJoinObj(conf: SQLConf): Rule[LogicalPlan] = {
157	if (SparkUtil.isSparkVersionEqualTo("2.2")) {	×
158	val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin";	×
159	CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]	×
160	} else if (SparkUtil.isSparkVersionXandAbove("2.3")) {	×
161	val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$";	×
162	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
163	.asInstanceOf[Rule[LogicalPlan]]
164	} else if (SparkUtil.isSparkVersionEqualTo("2.1")) {	×
165	val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$";	×
166	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
167	.asInstanceOf[Rule[LogicalPlan]]
168	}
169	else {
170	throw new UnsupportedOperationException("Spark version not supported")	×
171	}
172	}
173
174	def getNullPropagationObj(conf: SQLConf): Rule[LogicalPlan] = {
175	if (SparkUtil.isSparkVersionEqualTo("2.2")) {	×
176	val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation";	×
177	CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]	×
178	} else if (SparkUtil.isSparkVersionXandAbove("2.3")) {	×
179	val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation$";	×
180	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
181	.asInstanceOf[Rule[LogicalPlan]]
182	} else if (SparkUtil.isSparkVersionEqualTo("2.1")) {	×
183	val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation$";	×
184	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
185	.asInstanceOf[Rule[LogicalPlan]]
186	} else {
187	throw new UnsupportedOperationException("Spark version not supported")	×
188	}
189	}
190
191	def getCheckCartesianProductsObj(conf: SQLConf): Rule[LogicalPlan] = {
192	if (SparkUtil.isSparkVersionEqualTo("2.2")) {	×
193	val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts";	×
194	CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]	×
195	} else if (SparkUtil.isSparkVersionXandAbove("2.3")) {	×
196	val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts$";	×
197	CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1	×
198	.asInstanceOf[Rule[LogicalPlan]]
199	} else if (SparkUtil.isSparkVersionEqualTo("2.1")) {	×
200	val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts";	×
201	CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]	×
202	}
203	else {
204	throw new UnsupportedOperationException("Spark version not supported")	×
205	}
206	}
207
208	/**
209	* Method to broadcast a variable using spark SerializableConfiguration class
210	*
211	* @param sparkContext
212	* @param hadoopConf
213	* @return
214	*/
215	def broadCastHadoopConf(sparkContext: SparkContext,
216	hadoopConf: Configuration): Broadcast[SerializableConfiguration] = {
217	sparkContext.broadcast(getSerializableConfigurableInstance(hadoopConf))	1✔
218	}
219
220	def getSerializableConfigurableInstance(hadoopConf: Configuration): SerializableConfiguration = {
221	new SerializableConfiguration(hadoopConf)	1✔
222	}
223
224	/**
225	* Get the task group id
226	*
227	* @param sparkSession
228	* @return
229	*/
230	def getTaskGroupId(sparkSession: SparkSession): String = {
231	val taskGroupId = sparkSession.sparkContext.getLocalProperty("spark.jobGroup.id") match {	×
232	case null => ""	×
233	case _ => sparkSession.sparkContext.getLocalProperty("spark.jobGroup.id")	×
234	}
235	taskGroupId	×
236	}
237
238	/**
239	* Get the task group description
240	*
241	* @param sparkSession
242	* @return
243	*/
244	def getTaskGroupDesc(sparkSession: SparkSession): String = {
245	val taskGroupDesc = sparkSession.sparkContext.getLocalProperty("spark.job.description") match {	×
246	case null => ""	×
247	case _ => sparkSession.sparkContext.getLocalProperty("spark.job.description")	×
248	}
249	taskGroupDesc	×
250	}
251
252	}	1✔

apache / carbondata / 3929

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