9933

Build # 9933

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/52357f7de5221d27e5eb41ebac8acff7be269f67">52357f7de<a href="https://github.com/apache/carbondata/commit/52357f7de5221d27e5eb41ebac8acff7be269f67">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/2f003fdebe83149440c665b13537c01b567424a3">2f003fdeb</a>">2f003fdeb</a><a href="https://github.com/apache/carbondata/commit/52357f7de5221d27e5eb41ebac8acff7be269f67"> into d9f1a8115">d9f1a8115</a>

Run Details

61577 of 77403 relevant lines covered (79.55%)

1.05 hits per line

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

81.17

/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.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.io.File
import java.util.concurrent.atomic.AtomicLong

import scala.collection.JavaConverters._

import org.apache.commons.lang.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession.Builder
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.hive.execution.command.CarbonSetCommand
import org.apache.spark.sql.internal.{SessionState, SharedState}
import org.apache.spark.sql.profiler.{Profiler, SQLStart}
import org.apache.spark.util.{CarbonReflectionUtils, Utils}

import org.apache.carbondata.common.annotations.InterfaceAudience
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.{CarbonProperties, CarbonSessionInfo, ThreadLocalSessionInfo}
import org.apache.carbondata.streaming.CarbonStreamingQueryListener

/**
 * Session implementation for {org.apache.spark.sql.SparkSession}
 * Implemented this class only to use our own SQL DDL commands.
 * User needs to use {CarbonSession.getOrCreateCarbon} to create Carbon session.
 */
class CarbonSession(@transient val sc: SparkContext,
    @transient private val existingSharedState: Option[SharedState],
    @transient private val useHiveMetaStore: Boolean = true
) extends SparkSession(sc) { self =>

  def this(sc: SparkContext) {
    this(sc, None)
  }

  @transient
  override lazy val sessionState: SessionState = {
    CarbonReflectionUtils.getSessionState(sparkContext, this, useHiveMetaStore)
      .asInstanceOf[SessionState]
  }

  /**
   * State shared across sessions, including the `SparkContext`, cached data, listener,
   * and a catalog that interacts with external systems.
   */
  @transient
  override lazy val sharedState: SharedState = {
    existingSharedState match {
      case Some(_) =>
        val ss = existingSharedState.get
        if (ss == null) {
          new SharedState(sparkContext)
        } else {
          ss
        }
      case None =>
        new SharedState(sparkContext)
    }
  }

  override def newSession(): SparkSession = {
    new CarbonSession(sparkContext, Some(sharedState), useHiveMetaStore)
  }

  /**
   * Run search mode if enabled, otherwise run SparkSQL
   */
  override def sql(sqlText: String): DataFrame = {
    withProfiler(
      sqlText,
      (qe, sse) => {
          new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))
      }
    )
  }

  /**
   * Return true if the specified sql statement will hit the datamap
   * This API is for test purpose only
   */
  @InterfaceAudience.Developer(Array("DataMap"))
  def isDataMapHit(sqlStatement: String, dataMapName: String): Boolean = {
    // explain command will output the dataMap information only if enable.query.statistics = true
    val message = sql(s"EXPLAIN $sqlStatement").collect()
    message(0).getString(0).contains(dataMapName)
  }

  /**
   * Run SparkSQL directly
   */
  def sparkSql(sqlText: String): DataFrame = {
    withProfiler(
      sqlText,
      (qe, sse) => new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))
    )
  }

  private def withProfiler(
      sqlText: String,
      generateDF: (QueryExecution, SQLStart) => DataFrame): DataFrame = {
    val sse = SQLStart(sqlText, CarbonSession.statementId.getAndIncrement())
    CarbonSession.threadStatementId.set(sse.statementId)
    sse.startTime = System.currentTimeMillis()

    try {
      val logicalPlan = sessionState.sqlParser.parsePlan(sqlText)
      sse.parseEnd = System.currentTimeMillis()

      val qe = sessionState.executePlan(logicalPlan)
      qe.assertAnalyzed()
      sse.isCommand = qe.analyzed match {
        case c: Command => true
        case u @ Union(children) if children.forall(_.isInstanceOf[Command]) => true
        case _ => false
      }
      sse.analyzerEnd = System.currentTimeMillis()
      generateDF(qe, sse)
    } finally {
      Profiler.invokeIfEnable {
        if (sse.isCommand) {
          sse.endTime = System.currentTimeMillis()
          Profiler.send(sse)
        } else {
          Profiler.addStatementMessage(sse.statementId, sse)
        }
      }
    }
  }
}

object CarbonSession {

  private val statementId = new AtomicLong(0)

  private var enableInMemCatlog: Boolean = false

  private[sql] val threadStatementId = new ThreadLocal[Long]()

  implicit class CarbonBuilder(builder: Builder) {

    def enableInMemoryCatalog(): Builder = {
      enableInMemCatlog = true
      builder
    }
    def getOrCreateCarbonSession(): SparkSession = {
      getOrCreateCarbonSession(null, null)
    }

    def getOrCreateCarbonSession(storePath: String): SparkSession = {
      getOrCreateCarbonSession(
        storePath,
        new File(CarbonCommonConstants.METASTORE_LOCATION_DEFAULT_VAL).getCanonicalPath)
    }

    def getOrCreateCarbonSession(storePath: String,
        metaStorePath: String): SparkSession = synchronized {
      if (!enableInMemCatlog) {
        builder.enableHiveSupport()
      }
      val options =
        getValue("options", builder).asInstanceOf[scala.collection.mutable.HashMap[String, String]]
      val userSuppliedContext: Option[SparkContext] =
        getValue("userSuppliedContext", builder).asInstanceOf[Option[SparkContext]]

      if (StringUtils.isNotBlank(metaStorePath)) {
        val hadoopConf = new Configuration()
        val configFile = Utils.getContextOrSparkClassLoader.getResource("hive-site.xml")
        if (configFile != null) {
          hadoopConf.addResource(configFile)
        }
        if (options.get(CarbonCommonConstants.HIVE_CONNECTION_URL).isEmpty &&
            hadoopConf.get(CarbonCommonConstants.HIVE_CONNECTION_URL) == null) {
          val metaStorePathAbsolute = new File(metaStorePath).getCanonicalPath
          val hiveMetaStoreDB = metaStorePathAbsolute + "/metastore_db"
          options ++= Map[String, String]((CarbonCommonConstants.HIVE_CONNECTION_URL,
            s"jdbc:derby:;databaseName=$hiveMetaStoreDB;create=true"))
        }
      }

      // Get the session from current thread's active session.
      var session: SparkSession = SparkSession.getActiveSession match {
        case Some(sparkSession: CarbonSession) =>
          if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {
            options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }
            sparkSession
          } else {
            null
          }
        case _ => null
      }
      if (session ne null) {
        return session
      }

      // Global synchronization so we will only set the default session once.
      SparkSession.synchronized {
        // If the current thread does not have an active session, get it from the global session.
        session = SparkSession.getDefaultSession match {
          case Some(sparkSession: CarbonSession) =>
            if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {
              options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }
              sparkSession
            } else {
              null
            }
          case _ => null
        }
        if (session ne null) {
          return session
        }

        // No active nor global default session. Create a new one.
        val sparkContext = userSuppliedContext.getOrElse {
          // set app name if not given
          val randomAppName = java.util.UUID.randomUUID().toString
          val sparkConf = new SparkConf()
          options.foreach { case (k, v) => sparkConf.set(k, v) }
          if (!sparkConf.contains("spark.app.name")) {
            sparkConf.setAppName(randomAppName)
          }
          val sc = SparkContext.getOrCreate(sparkConf)
          // maybe this is an existing SparkContext, update its SparkConf which maybe used
          // by SparkSession
          options.foreach { case (k, v) => sc.conf.set(k, v) }
          if (!sc.conf.contains("spark.app.name")) {
            sc.conf.setAppName(randomAppName)
          }
          sc
        }

        session = new CarbonSession(sparkContext, None, !enableInMemCatlog)
        val carbonProperties = CarbonProperties.getInstance()
        if (StringUtils.isNotBlank(storePath)) {
          carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)
          // In case if it is in carbon.properties for backward compatible
        } else if (carbonProperties.getProperty(CarbonCommonConstants.STORE_LOCATION) == null) {
          carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION,
            session.sessionState.conf.warehousePath)
        }
        options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) }
        SparkSession.setDefaultSession(session)
        // Setup monitor end point and register CarbonMonitorListener
        Profiler.initialize(sparkContext)
        // Register a successfully instantiated context to the singleton. This should be at the
        // end of the class definition so that the singleton is updated only if there is no
        // exception in the construction of the instance.
        CarbonToSparkAdapater.addSparkListener(sparkContext)
        session.streams.addListener(new CarbonStreamingQueryListener(session))
      }

      session
    }

    /**
     * It is a hack to get the private field from class.
     */
    def getValue(name: String, builder: Builder): Any = {
      val currentMirror = scala.reflect.runtime.currentMirror
      val instanceMirror = currentMirror.reflect(builder)
      val m = currentMirror.classSymbol(builder.getClass).
        toType.members.find { p =>
        p.name.toString.equals(name)
      }.get.asTerm
      instanceMirror.reflectField(m).get
    }
  }

  def threadSet(key: String, value: String): Unit = {
    var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (currentThreadSessionInfo == null) {
      currentThreadSessionInfo = new CarbonSessionInfo()
    }
    else {
      currentThreadSessionInfo = currentThreadSessionInfo.clone()
    }
    val threadParams = currentThreadSessionInfo.getThreadParams
    CarbonSetCommand.validateAndSetValue(threadParams, key, value)
    ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)
  }


  def threadSet(key: String, value: Object): Unit = {
    var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (currentThreadSessionInfo == null) {
      currentThreadSessionInfo = new CarbonSessionInfo()
    }
    else {
      currentThreadSessionInfo = currentThreadSessionInfo.clone()
    }
    currentThreadSessionInfo.getThreadParams.setExtraInfo(key, value)
    ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)
  }

  def threadUnset(key: String): Unit = {
    val currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (currentThreadSessionInfo != null) {
      val currentThreadSessionInfoClone = currentThreadSessionInfo.clone()
      val threadParams = currentThreadSessionInfoClone.getThreadParams
      CarbonSetCommand.unsetValue(threadParams, key)
      threadParams.removeExtraInfo(key)
      ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfoClone)
    }
  }

  def updateSessionInfoToCurrentThread(sparkSession: SparkSession): Unit = {
    val carbonSessionInfo = CarbonEnv.getInstance(sparkSession).carbonSessionInfo.clone()
    val currentThreadSessionInfoOrig = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (currentThreadSessionInfoOrig != null) {
      val currentThreadSessionInfo = currentThreadSessionInfoOrig.clone()
      // copy all the thread parameters to apply to session parameters
      currentThreadSessionInfo.getThreadParams.getAll.asScala
        .foreach(entry => carbonSessionInfo.getSessionParams.addProperty(entry._1, entry._2))
      carbonSessionInfo.setThreadParams(currentThreadSessionInfo.getThreadParams)
    }
    // preserve thread parameters across call
    ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)
    ThreadLocalSessionInfo
      .setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())
  }

}

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	package org.apache.spark.sql
18
19	import java.io.File
20	import java.util.concurrent.atomic.AtomicLong
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.commons.lang.StringUtils
25	import org.apache.hadoop.conf.Configuration
26	import org.apache.spark.{SparkConf, SparkContext}
27	import org.apache.spark.sql.SparkSession.Builder
28	import org.apache.spark.sql.catalyst.encoders.RowEncoder
29	import org.apache.spark.sql.catalyst.plans.logical._
30	import org.apache.spark.sql.execution.QueryExecution
31	import org.apache.spark.sql.hive.execution.command.CarbonSetCommand
32	import org.apache.spark.sql.internal.{SessionState, SharedState}
33	import org.apache.spark.sql.profiler.{Profiler, SQLStart}
34	import org.apache.spark.util.{CarbonReflectionUtils, Utils}
35
36	import org.apache.carbondata.common.annotations.InterfaceAudience
37	import org.apache.carbondata.core.constants.CarbonCommonConstants
38	import org.apache.carbondata.core.util.{CarbonProperties, CarbonSessionInfo, ThreadLocalSessionInfo}
39	import org.apache.carbondata.streaming.CarbonStreamingQueryListener
40
41	/**
42	* Session implementation for {org.apache.spark.sql.SparkSession}
43	* Implemented this class only to use our own SQL DDL commands.
44	* User needs to use {CarbonSession.getOrCreateCarbon} to create Carbon session.
45	*/
46	class CarbonSession(@transient val sc: SparkContext,	2✔
47	@transient private val existingSharedState: Option[SharedState],	2✔
48	@transient private val useHiveMetaStore: Boolean = true	2✔
49	) extends SparkSession(sc) { self =>	2✔
50
51	def this(sc: SparkContext) {
52	this(sc, None)	×
53	}
54
55	@transient
56	override lazy val sessionState: SessionState = {	2✔
57	CarbonReflectionUtils.getSessionState(sparkContext, this, useHiveMetaStore)	2✔
58	.asInstanceOf[SessionState]
59	}
60
61	/**
62	* State shared across sessions, including the `SparkContext`, cached data, listener,
63	* and a catalog that interacts with external systems.
64	*/
65	@transient
66	override lazy val sharedState: SharedState = {	2✔
67	existingSharedState match {	2✔
68	case Some(_) =>	2✔
69	val ss = existingSharedState.get	×
70	if (ss == null) {	×
71	new SharedState(sparkContext)	×
72	} else {
73	ss	×
74	}
75	case None =>	2✔
76	new SharedState(sparkContext)	2✔
77	}
78	}
79
80	override def newSession(): SparkSession = {
81	new CarbonSession(sparkContext, Some(sharedState), useHiveMetaStore)	×
82	}
83
84	/**
85	* Run search mode if enabled, otherwise run SparkSQL
86	*/
87	override def sql(sqlText: String): DataFrame = {
88	withProfiler(	2✔
89	sqlText,	2✔
90	(qe, sse) => {	2✔
91	new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	2✔
92	}
93	)
94	}
95
96	/**
97	* Return true if the specified sql statement will hit the datamap
98	* This API is for test purpose only
99	*/
100	@InterfaceAudience.Developer(Array("DataMap"))
101	def isDataMapHit(sqlStatement: String, dataMapName: String): Boolean = {
102	// explain command will output the dataMap information only if enable.query.statistics = true
103	val message = sql(s"EXPLAIN $sqlStatement").collect()	×
104	message(0).getString(0).contains(dataMapName)	×
105	}
106
107	/**
108	* Run SparkSQL directly
109	*/
110	def sparkSql(sqlText: String): DataFrame = {
111	withProfiler(	×
112	sqlText,	×
113	(qe, sse) => new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	×
114	)
115	}
116
117	private def withProfiler(
118	sqlText: String,
119	generateDF: (QueryExecution, SQLStart) => DataFrame): DataFrame = {
120	val sse = SQLStart(sqlText, CarbonSession.statementId.getAndIncrement())	2✔
121	CarbonSession.threadStatementId.set(sse.statementId)	2✔
122	sse.startTime = System.currentTimeMillis()	2✔
123
124	try {
125	val logicalPlan = sessionState.sqlParser.parsePlan(sqlText)	2✔
126	sse.parseEnd = System.currentTimeMillis()	2✔
127
128	val qe = sessionState.executePlan(logicalPlan)	2✔
129	qe.assertAnalyzed()	2✔
130	sse.isCommand = qe.analyzed match {	2✔
131	case c: Command => true	2✔
132	case u @ Union(children) if children.forall(_.isInstanceOf[Command]) => true	2✔
133	case _ => false	2✔
134	}
135	sse.analyzerEnd = System.currentTimeMillis()	2✔
136	generateDF(qe, sse)	2✔
137	} finally {
138	Profiler.invokeIfEnable {	2✔
139	if (sse.isCommand) {	2✔
140	sse.endTime = System.currentTimeMillis()	1✔
141	Profiler.send(sse)	1✔
142	} else {
143	Profiler.addStatementMessage(sse.statementId, sse)	1✔
144	}
145	}
146	}
147	}
148	}
149
150	object CarbonSession {	×
151
152	private val statementId = new AtomicLong(0)	2✔
153
154	private var enableInMemCatlog: Boolean = false	2✔
155
156	private[sql] val threadStatementId = new ThreadLocal[Long]()	2✔
157
158	implicit class CarbonBuilder(builder: Builder) {	2✔
159
160	def enableInMemoryCatalog(): Builder = {
161	enableInMemCatlog = true	×
162	builder	×
163	}
164	def getOrCreateCarbonSession(): SparkSession = {
165	getOrCreateCarbonSession(null, null)	2✔
166	}
167
168	def getOrCreateCarbonSession(storePath: String): SparkSession = {
169	getOrCreateCarbonSession(	×
170	storePath,	×
171	new File(CarbonCommonConstants.METASTORE_LOCATION_DEFAULT_VAL).getCanonicalPath)	×
172	}
173
174	def getOrCreateCarbonSession(storePath: String,
175	metaStorePath: String): SparkSession = synchronized {
176	if (!enableInMemCatlog) {	2✔
177	builder.enableHiveSupport()	2✔
178	}
179	val options =	2✔
180	getValue("options", builder).asInstanceOf[scala.collection.mutable.HashMap[String, String]]	2✔
181	val userSuppliedContext: Option[SparkContext] =	2✔
182	getValue("userSuppliedContext", builder).asInstanceOf[Option[SparkContext]]	2✔
183
184	if (StringUtils.isNotBlank(metaStorePath)) {	2✔
185	val hadoopConf = new Configuration()	2✔
186	val configFile = Utils.getContextOrSparkClassLoader.getResource("hive-site.xml")	2✔
187	if (configFile != null) {	2✔
188	hadoopConf.addResource(configFile)	×
189	}
190	if (options.get(CarbonCommonConstants.HIVE_CONNECTION_URL).isEmpty &&	2✔
191	hadoopConf.get(CarbonCommonConstants.HIVE_CONNECTION_URL) == null) {	2✔
192	val metaStorePathAbsolute = new File(metaStorePath).getCanonicalPath	2✔
193	val hiveMetaStoreDB = metaStorePathAbsolute + "/metastore_db"	2✔
194	options ++= Map[String, String]((CarbonCommonConstants.HIVE_CONNECTION_URL,	2✔
195	s"jdbc:derby:;databaseName=$hiveMetaStoreDB;create=true"))	2✔
196	}
197	}
198
199	// Get the session from current thread's active session.
200	var session: SparkSession = SparkSession.getActiveSession match {	2✔
201	case Some(sparkSession: CarbonSession) =>	2✔
202	if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {	2✔
203	options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }	2✔
204	sparkSession	2✔
205	} else {
206	null	×
207	}
208	case _ => null	2✔
209	}
210	if (session ne null) {	2✔
211	return session	2✔
212	}
213
214	// Global synchronization so we will only set the default session once.
215	SparkSession.synchronized {	2✔
216	// If the current thread does not have an active session, get it from the global session.
217	session = SparkSession.getDefaultSession match {	2✔
218	case Some(sparkSession: CarbonSession) =>	2✔
219	if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {	×
220	options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }	×
221	sparkSession	×
222	} else {
223	null	×
224	}
225	case _ => null	2✔
226	}
227	if (session ne null) {	2✔
228	return session	×
229	}
230
231	// No active nor global default session. Create a new one.
232	val sparkContext = userSuppliedContext.getOrElse {	2✔
233	// set app name if not given
234	val randomAppName = java.util.UUID.randomUUID().toString	2✔
235	val sparkConf = new SparkConf()	2✔
236	options.foreach { case (k, v) => sparkConf.set(k, v) }	2✔
237	if (!sparkConf.contains("spark.app.name")) {	2✔
238	sparkConf.setAppName(randomAppName)	×
239	}
240	val sc = SparkContext.getOrCreate(sparkConf)	2✔
241	// maybe this is an existing SparkContext, update its SparkConf which maybe used
242	// by SparkSession
243	options.foreach { case (k, v) => sc.conf.set(k, v) }	2✔
244	if (!sc.conf.contains("spark.app.name")) {	2✔
245	sc.conf.setAppName(randomAppName)	×
246	}
247	sc	2✔
248	}
249
250	session = new CarbonSession(sparkContext, None, !enableInMemCatlog)	2✔
251	val carbonProperties = CarbonProperties.getInstance()	2✔
252	if (StringUtils.isNotBlank(storePath)) {	2✔
253	carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)	×
254	// In case if it is in carbon.properties for backward compatible
255	} else if (carbonProperties.getProperty(CarbonCommonConstants.STORE_LOCATION) == null) {	2✔
256	carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION,	2✔
257	session.sessionState.conf.warehousePath)	2✔
258	}
259	options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) }	2✔
260	SparkSession.setDefaultSession(session)	2✔
261	// Setup monitor end point and register CarbonMonitorListener
262	Profiler.initialize(sparkContext)	2✔
263	// Register a successfully instantiated context to the singleton. This should be at the
264	// end of the class definition so that the singleton is updated only if there is no
265	// exception in the construction of the instance.
266	CarbonToSparkAdapater.addSparkListener(sparkContext)	2✔
267	session.streams.addListener(new CarbonStreamingQueryListener(session))	2✔
268	}
269
270	session	2✔
271	}
272
273	/**
274	* It is a hack to get the private field from class.
275	*/
276	def getValue(name: String, builder: Builder): Any = {
277	val currentMirror = scala.reflect.runtime.currentMirror	2✔
278	val instanceMirror = currentMirror.reflect(builder)	2✔
279	val m = currentMirror.classSymbol(builder.getClass).	2✔
280	toType.members.find { p =>	2✔
281	p.name.toString.equals(name)	2✔
282	}.get.asTerm	2✔
283	instanceMirror.reflectField(m).get	2✔
284	}
285	}
286
287	def threadSet(key: String, value: String): Unit = {	2✔
288	var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
289	if (currentThreadSessionInfo == null) {	2✔
290	currentThreadSessionInfo = new CarbonSessionInfo()	×
291	}
292	else {
293	currentThreadSessionInfo = currentThreadSessionInfo.clone()	2✔
294	}
295	val threadParams = currentThreadSessionInfo.getThreadParams	2✔
296	CarbonSetCommand.validateAndSetValue(threadParams, key, value)	2✔
297	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)	2✔
298	}
299
300
301	def threadSet(key: String, value: Object): Unit = {
302	var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
303	if (currentThreadSessionInfo == null) {	2✔
304	currentThreadSessionInfo = new CarbonSessionInfo()	×
305	}
306	else {
307	currentThreadSessionInfo = currentThreadSessionInfo.clone()	2✔
308	}
309	currentThreadSessionInfo.getThreadParams.setExtraInfo(key, value)	2✔
310	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)	2✔
311	}
312
313	def threadUnset(key: String): Unit = {	2✔
314	val currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
315	if (currentThreadSessionInfo != null) {	2✔
316	val currentThreadSessionInfoClone = currentThreadSessionInfo.clone()	2✔
317	val threadParams = currentThreadSessionInfoClone.getThreadParams	2✔
318	CarbonSetCommand.unsetValue(threadParams, key)	2✔
319	threadParams.removeExtraInfo(key)	2✔
320	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfoClone)	2✔
321	}
322	}
323
324	def updateSessionInfoToCurrentThread(sparkSession: SparkSession): Unit = {
325	val carbonSessionInfo = CarbonEnv.getInstance(sparkSession).carbonSessionInfo.clone()	2✔
326	val currentThreadSessionInfoOrig = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
327	if (currentThreadSessionInfoOrig != null) {	2✔
328	val currentThreadSessionInfo = currentThreadSessionInfoOrig.clone()	2✔
329	// copy all the thread parameters to apply to session parameters
330	currentThreadSessionInfo.getThreadParams.getAll.asScala	2✔
331	.foreach(entry => carbonSessionInfo.getSessionParams.addProperty(entry._1, entry._2))	2✔
332	carbonSessionInfo.setThreadParams(currentThreadSessionInfo.getThreadParams)	2✔
333	}
334	// preserve thread parameters across call
335	ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)	2✔
336	ThreadLocalSessionInfo
337	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
338	}
339
340	}	2✔

apache / carbondata / 9933

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