8226

Build # 8226

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/0ddb23e10534308a67280df30a6cffe018252bd8">0ddb23e10<a href="https://github.com/apache/carbondata/commit/0ddb23e10534308a67280df30a6cffe018252bd8">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/4154ba2c8749707a50359d2fd9d0d07656b51abd">4154ba2c8</a>">4154ba2c8</a><a href="https://github.com/apache/carbondata/commit/0ddb23e10534308a67280df30a6cffe018252bd8"> into b588cb655">b588cb655</a>

Run Details

28 of 28 new or added lines in 2 files covered. (100.0%)

57505 of 71359 relevant lines covered (80.59%)

1.07 hits per line

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

83.49

/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.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.SparkSession.Builder
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.hive.execution.command.CarbonSetCommand
import org.apache.spark.sql.internal.{SessionState, SharedState}
import org.apache.spark.sql.optimizer.CarbonFilters
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.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.{CarbonProperties, CarbonSessionInfo, ThreadLocalSessionInfo}
import org.apache.carbondata.store.SparkCarbonStore
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 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) => {
        if (isSearchModeEnabled) {
          try {
            trySearchMode(qe, sse)
          } catch {
            case e: Exception =>
              log.error(String.format(
                "Exception when executing search mode: %s", e.getMessage))
              throw e;
          }
        } else {
          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 = {
    val message = sql(s"EXPLAIN $sqlStatement").collect()
    message(0).getString(0).contains(dataMapName)
  }

  def isSearchModeEnabled: Boolean = carbonStore != null

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

  /**
   * If the query is a simple query with filter, we will try to use Search Mode,
   * otherwise execute in SparkSQL
   */
  private def trySearchMode(qe: QueryExecution, sse: SQLStart): DataFrame = {
    val analyzed = qe.analyzed
    val LOG: LogService = LogServiceFactory.getLogService(this.getClass.getName)
    analyzed match {
      case _@Project(columns, _@Filter(expr, s: SubqueryAlias))
        if s.child.isInstanceOf[LogicalRelation] &&
           s.child.asInstanceOf[LogicalRelation].relation
             .isInstanceOf[CarbonDatasourceHadoopRelation] =>
        LOG.info(s"Search service started and supports filter: ${sse.sqlText}")
        runSearch(analyzed, columns, expr, s.child.asInstanceOf[LogicalRelation])
      case gl@GlobalLimit(_, ll@LocalLimit(_, p@Project(columns, _@Filter(expr, s: SubqueryAlias))))
        if s.child.isInstanceOf[LogicalRelation] &&
           s.child.asInstanceOf[LogicalRelation].relation
             .isInstanceOf[CarbonDatasourceHadoopRelation] =>
        val logicalRelation = s.child.asInstanceOf[LogicalRelation]
        LOG.info(s"Search service started and supports limit: ${sse.sqlText}")
        runSearch(analyzed, columns, expr, logicalRelation, gl.maxRows, ll.maxRows)
      case _ =>
        LOG.info(s"Search service started, but don't support: ${sse.sqlText}," +
          s" and will run it with SparkSQL")
        new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))
    }
  }

  @transient private var carbonStore: SparkCarbonStore = _

  def startSearchMode(): Unit = {
    CarbonProperties.enableSearchMode(true)
    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "false")
    if (carbonStore == null) {
      carbonStore = new SparkCarbonStore(this)
      carbonStore.startSearchMode()
    }
  }

  def stopSearchMode(): Unit = {
    CarbonProperties.enableSearchMode(false)
    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "true")
    if (carbonStore != null) {
      try {
        carbonStore.stopSearchMode()
        carbonStore = null
      } catch {
        case e: RuntimeException =>
          LogServiceFactory.getLogService(this.getClass.getCanonicalName)
            .error(s"Stop search mode failed: ${e.getMessage}")
      }
    }
  }

  private def runSearch(
      logicalPlan: LogicalPlan,
      columns: Seq[NamedExpression],
      expr: Expression,
      relation: LogicalRelation,
      maxRows: Option[Long] = None,
      localMaxRows: Option[Long] = None): DataFrame = {
    val rows = carbonStore.search(
        relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable,
        columns.map(_.name).toArray,
        if (expr != null) CarbonFilters.transformExpression(expr) else null,
        maxRows.getOrElse(Long.MaxValue),
        localMaxRows.getOrElse(Long.MaxValue))
    val output = new java.util.ArrayList[Row]()
    while (rows.hasNext) {
      val row = rows.next()
      output.add(Row.fromSeq(row.getData))
    }
    createDataFrame(output, logicalPlan.schema)
  }

}

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 (metaStorePath != null) {
        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 (storePath != null) {
          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.
        sparkContext.addSparkListener(new SparkListener {
          override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
            SparkSession.setDefaultSession(null)
            SparkSession.sqlListener.set(null)
          }
        })
        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.hadoop.conf.Configuration
25	import org.apache.spark.{SparkConf, SparkContext}
26	import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
27	import org.apache.spark.sql.SparkSession.Builder
28	import org.apache.spark.sql.catalyst.encoders.RowEncoder
29	import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
30	import org.apache.spark.sql.catalyst.plans.logical._
31	import org.apache.spark.sql.execution.QueryExecution
32	import org.apache.spark.sql.execution.datasources.LogicalRelation
33	import org.apache.spark.sql.hive.execution.command.CarbonSetCommand
34	import org.apache.spark.sql.internal.{SessionState, SharedState}
35	import org.apache.spark.sql.optimizer.CarbonFilters
36	import org.apache.spark.sql.profiler.{Profiler, SQLStart}
37	import org.apache.spark.util.{CarbonReflectionUtils, Utils}
38
39	import org.apache.carbondata.common.annotations.InterfaceAudience
40	import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
41	import org.apache.carbondata.core.constants.CarbonCommonConstants
42	import org.apache.carbondata.core.util.{CarbonProperties, CarbonSessionInfo, ThreadLocalSessionInfo}
43	import org.apache.carbondata.store.SparkCarbonStore
44	import org.apache.carbondata.streaming.CarbonStreamingQueryListener
45
46	/**
47	* Session implementation for {org.apache.spark.sql.SparkSession}
48	* Implemented this class only to use our own SQL DDL commands.
49	* User needs to use {CarbonSession.getOrCreateCarbon} to create Carbon session.
50	*/
51	class CarbonSession(@transient val sc: SparkContext,	2✔
52	@transient private val existingSharedState: Option[SharedState],	2✔
53	@transient useHiveMetaStore: Boolean = true	2✔
54	) extends SparkSession(sc) { self =>	2✔
55
56	def this(sc: SparkContext) {
57	this(sc, None)	×
58	}
59
60	@transient
61	override lazy val sessionState: SessionState = {	2✔
62	CarbonReflectionUtils.getSessionState(sparkContext, this, useHiveMetaStore)	2✔
63	.asInstanceOf[SessionState]
64	}
65
66	/**
67	* State shared across sessions, including the `SparkContext`, cached data, listener,
68	* and a catalog that interacts with external systems.
69	*/
70	@transient
71	override lazy val sharedState: SharedState = {	2✔
72	existingSharedState match {	2✔
73	case Some(_) =>	2✔
74	val ss = existingSharedState.get	×
75	if (ss == null) {	×
76	new SharedState(sparkContext)	×
77	} else {
78	ss	×
79	}
80	case None =>	2✔
81	new SharedState(sparkContext)	2✔
82	}
83	}
84
85	override def newSession(): SparkSession = {
86	new CarbonSession(sparkContext, Some(sharedState), useHiveMetaStore)	×
87	}
88
89	/**
90	* Run search mode if enabled, otherwise run SparkSQL
91	*/
92	override def sql(sqlText: String): DataFrame = {
93	withProfiler(	2✔
94	sqlText,	2✔
95	(qe, sse) => {	2✔
96	if (isSearchModeEnabled) {	2✔
97	try {	×
98	trySearchMode(qe, sse)	1✔
99	} catch {
100	case e: Exception =>
101	log.error(String.format(	×
102	"Exception when executing search mode: %s", e.getMessage))	×
103	throw e;	×
104	}
105	} else {
106	new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	2✔
107	}
108	}
109	)
110	}
111
112	/**
113	* Return true if the specified sql statement will hit the datamap
114	* This API is for test purpose only
115	*/
116	@InterfaceAudience.Developer(Array("DataMap"))
117	def isDataMapHit(sqlStatement: String, dataMapName: String): Boolean = {
118	val message = sql(s"EXPLAIN $sqlStatement").collect()	×
119	message(0).getString(0).contains(dataMapName)	×
120	}
121
122	def isSearchModeEnabled: Boolean = carbonStore != null	2✔
123
124	/**
125	* Run SparkSQL directly
126	*/
127	def sparkSql(sqlText: String): DataFrame = {
128	withProfiler(	1✔
129	sqlText,	1✔
130	(qe, sse) => new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	1✔
131	)
132	}
133
134	private def withProfiler(
135	sqlText: String,
136	generateDF: (QueryExecution, SQLStart) => DataFrame): DataFrame = {
137	val sse = SQLStart(sqlText, CarbonSession.statementId.getAndIncrement())	2✔
138	CarbonSession.threadStatementId.set(sse.statementId)	2✔
139	sse.startTime = System.currentTimeMillis()	2✔
140
141	try {
142	val logicalPlan = sessionState.sqlParser.parsePlan(sqlText)	2✔
143	sse.parseEnd = System.currentTimeMillis()	2✔
144
145	val qe = sessionState.executePlan(logicalPlan)	2✔
146	qe.assertAnalyzed()	2✔
147	sse.isCommand = qe.analyzed match {	2✔
148	case c: Command => true	2✔
149	case u @ Union(children) if children.forall(_.isInstanceOf[Command]) => true	2✔
150	case _ => false	2✔
151	}
152	sse.analyzerEnd = System.currentTimeMillis()	2✔
153	generateDF(qe, sse)	2✔
154	} finally {
155	Profiler.invokeIfEnable {	2✔
156	if (sse.isCommand) {	2✔
157	sse.endTime = System.currentTimeMillis()	1✔
158	Profiler.send(sse)	1✔
159	} else {
160	Profiler.addStatementMessage(sse.statementId, sse)	1✔
161	}
162	}
163	}
164	}
165
166	/**
167	* If the query is a simple query with filter, we will try to use Search Mode,
168	* otherwise execute in SparkSQL
169	*/
170	private def trySearchMode(qe: QueryExecution, sse: SQLStart): DataFrame = {
171	val analyzed = qe.analyzed	1✔
172	val LOG: LogService = LogServiceFactory.getLogService(this.getClass.getName)	1✔
173	analyzed match {	1✔
174	case _@Project(columns, _@Filter(expr, s: SubqueryAlias))	1✔
175	if s.child.isInstanceOf[LogicalRelation] &&	1✔
176	s.child.asInstanceOf[LogicalRelation].relation	1✔
177	.isInstanceOf[CarbonDatasourceHadoopRelation] =>	1✔
178	LOG.info(s"Search service started and supports filter: ${sse.sqlText}")	1✔
179	runSearch(analyzed, columns, expr, s.child.asInstanceOf[LogicalRelation])	1✔
180	case gl@GlobalLimit(_, ll@LocalLimit(_, p@Project(columns, _@Filter(expr, s: SubqueryAlias))))	1✔
181	if s.child.isInstanceOf[LogicalRelation] &&	1✔
182	s.child.asInstanceOf[LogicalRelation].relation	1✔
183	.isInstanceOf[CarbonDatasourceHadoopRelation] =>	1✔
184	val logicalRelation = s.child.asInstanceOf[LogicalRelation]	1✔
185	LOG.info(s"Search service started and supports limit: ${sse.sqlText}")	1✔
186	runSearch(analyzed, columns, expr, logicalRelation, gl.maxRows, ll.maxRows)	1✔
187	case _ =>	1✔
188	LOG.info(s"Search service started, but don't support: ${sse.sqlText}," +	1✔
189	s" and will run it with SparkSQL")	1✔
190	new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	1✔
191	}
192	}
193
194	@transient private var carbonStore: SparkCarbonStore = _	2✔
195
196	def startSearchMode(): Unit = {	1✔
197	CarbonProperties.enableSearchMode(true)	1✔
198	CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "false")	1✔
199	if (carbonStore == null) {	1✔
200	carbonStore = new SparkCarbonStore(this)	1✔
201	carbonStore.startSearchMode()	1✔
202	}
203	}
204
205	def stopSearchMode(): Unit = {	1✔
206	CarbonProperties.enableSearchMode(false)	1✔
207	CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "true")	1✔
208	if (carbonStore != null) {	1✔
209	try {	×
210	carbonStore.stopSearchMode()	1✔
211	carbonStore = null	1✔
212	} catch {
213	case e: RuntimeException =>
214	LogServiceFactory.getLogService(this.getClass.getCanonicalName)	×
215	.error(s"Stop search mode failed: ${e.getMessage}")	×
216	}
217	}
218	}
219
220	private def runSearch(
221	logicalPlan: LogicalPlan,
222	columns: Seq[NamedExpression],
223	expr: Expression,
224	relation: LogicalRelation,
225	maxRows: Option[Long] = None,	1✔
226	localMaxRows: Option[Long] = None): DataFrame = {	1✔
227	val rows = carbonStore.search(	1✔
228	relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable,	1✔
229	columns.map(_.name).toArray,	1✔
230	if (expr != null) CarbonFilters.transformExpression(expr) else null,	1✔
231	maxRows.getOrElse(Long.MaxValue),	1✔
232	localMaxRows.getOrElse(Long.MaxValue))	1✔
233	val output = new java.util.ArrayList[Row]()	1✔
234	while (rows.hasNext) {	1✔
235	val row = rows.next()	1✔
236	output.add(Row.fromSeq(row.getData))	1✔
237	}
238	createDataFrame(output, logicalPlan.schema)	1✔
239	}
240
241	}
242
243	object CarbonSession {	×
244
245	private val statementId = new AtomicLong(0)	2✔
246
247	private var enableInMemCatlog: Boolean = false	2✔
248
249	private[sql] val threadStatementId = new ThreadLocal[Long]()	2✔
250
251	implicit class CarbonBuilder(builder: Builder) {	2✔
252
253	def enableInMemoryCatalog(): Builder = {
254	enableInMemCatlog = true	×
255	builder	×
256	}
257	def getOrCreateCarbonSession(): SparkSession = {
258	getOrCreateCarbonSession(null, null)	2✔
259	}
260
261	def getOrCreateCarbonSession(storePath: String): SparkSession = {
262	getOrCreateCarbonSession(	×
263	storePath,	×
264	new File(CarbonCommonConstants.METASTORE_LOCATION_DEFAULT_VAL).getCanonicalPath)	×
265	}
266
267	def getOrCreateCarbonSession(storePath: String,
268	metaStorePath: String): SparkSession = synchronized {
269	if (!enableInMemCatlog) {	2✔
270	builder.enableHiveSupport()	2✔
271	}
272	val options =	2✔
273	getValue("options", builder).asInstanceOf[scala.collection.mutable.HashMap[String, String]]	2✔
274	val userSuppliedContext: Option[SparkContext] =	2✔
275	getValue("userSuppliedContext", builder).asInstanceOf[Option[SparkContext]]	2✔
276
277	if (metaStorePath != null) {	2✔
278	val hadoopConf = new Configuration()	2✔
279	val configFile = Utils.getContextOrSparkClassLoader.getResource("hive-site.xml")	2✔
280	if (configFile != null) {	2✔
281	hadoopConf.addResource(configFile)	×
282	}
283	if (options.get(CarbonCommonConstants.HIVE_CONNECTION_URL).isEmpty &&	2✔
284	hadoopConf.get(CarbonCommonConstants.HIVE_CONNECTION_URL) == null) {	2✔
285	val metaStorePathAbsolute = new File(metaStorePath).getCanonicalPath	2✔
286	val hiveMetaStoreDB = metaStorePathAbsolute + "/metastore_db"	2✔
287	options ++= Map[String, String]((CarbonCommonConstants.HIVE_CONNECTION_URL,	2✔
288	s"jdbc:derby:;databaseName=$hiveMetaStoreDB;create=true"))	2✔
289	}
290	}
291
292	// Get the session from current thread's active session.
293	var session: SparkSession = SparkSession.getActiveSession match {	2✔
294	case Some(sparkSession: CarbonSession) =>	2✔
295	if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {	2✔
296	options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }	2✔
297	sparkSession	2✔
298	} else {
299	null	×
300	}
301	case _ => null	2✔
302	}
303	if (session ne null) {	2✔
304	return session	2✔
305	}
306
307	// Global synchronization so we will only set the default session once.
308	SparkSession.synchronized {	2✔
309	// If the current thread does not have an active session, get it from the global session.
310	session = SparkSession.getDefaultSession match {	2✔
311	case Some(sparkSession: CarbonSession) =>	2✔
312	if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {	×
313	options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }	×
314	sparkSession	×
315	} else {
316	null	×
317	}
318	case _ => null	2✔
319	}
320	if (session ne null) {	2✔
321	return session	×
322	}
323
324	// No active nor global default session. Create a new one.
325	val sparkContext = userSuppliedContext.getOrElse {	2✔
326	// set app name if not given
327	val randomAppName = java.util.UUID.randomUUID().toString	2✔
328	val sparkConf = new SparkConf()	2✔
329	options.foreach { case (k, v) => sparkConf.set(k, v) }	2✔
330	if (!sparkConf.contains("spark.app.name")) {	2✔
331	sparkConf.setAppName(randomAppName)	×
332	}
333	val sc = SparkContext.getOrCreate(sparkConf)	2✔
334	// maybe this is an existing SparkContext, update its SparkConf which maybe used
335	// by SparkSession
336	options.foreach { case (k, v) => sc.conf.set(k, v) }	2✔
337	if (!sc.conf.contains("spark.app.name")) {	2✔
338	sc.conf.setAppName(randomAppName)	×
339	}
340	sc	2✔
341	}
342
343	session = new CarbonSession(sparkContext, None, !enableInMemCatlog)	2✔
344	val carbonProperties = CarbonProperties.getInstance()	2✔
345	if (storePath != null) {	2✔
346	carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)	×
347	// In case if it is in carbon.properties for backward compatible
348	} else if (carbonProperties.getProperty(CarbonCommonConstants.STORE_LOCATION) == null) {	2✔
349	carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION,	2✔
350	session.sessionState.conf.warehousePath)	2✔
351	}
352	options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) }	2✔
353	SparkSession.setDefaultSession(session)	2✔
354	// Setup monitor end point and register CarbonMonitorListener
355	Profiler.initialize(sparkContext)	2✔
356	// Register a successfully instantiated context to the singleton. This should be at the
357	// end of the class definition so that the singleton is updated only if there is no
358	// exception in the construction of the instance.
359	sparkContext.addSparkListener(new SparkListener {	2✔
360	override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
361	SparkSession.setDefaultSession(null)	×
362	SparkSession.sqlListener.set(null)	×
363	}
364	})
365	session.streams.addListener(new CarbonStreamingQueryListener(session))	2✔
366	}
367
368	session	2✔
369	}
370
371	/**
372	* It is a hack to get the private field from class.
373	*/
374	def getValue(name: String, builder: Builder): Any = {
375	val currentMirror = scala.reflect.runtime.currentMirror	2✔
376	val instanceMirror = currentMirror.reflect(builder)	2✔
377	val m = currentMirror.classSymbol(builder.getClass).	2✔
378	toType.members.find { p =>	2✔
379	p.name.toString.equals(name)	2✔
380	}.get.asTerm	2✔
381	instanceMirror.reflectField(m).get	2✔
382	}
383	}
384
385	def threadSet(key: String, value: String): Unit = {	2✔
386	var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
387	if (currentThreadSessionInfo == null) {	2✔
388	currentThreadSessionInfo = new CarbonSessionInfo()	×
389	}
390	else {
391	currentThreadSessionInfo = currentThreadSessionInfo.clone()	2✔
392	}
393	val threadParams = currentThreadSessionInfo.getThreadParams	2✔
394	CarbonSetCommand.validateAndSetValue(threadParams, key, value)	2✔
395	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)	2✔
396	}
397
398
399	def threadSet(key: String, value: Object): Unit = {
400	var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	1✔
401	if (currentThreadSessionInfo == null) {	1✔
402	currentThreadSessionInfo = new CarbonSessionInfo()	×
403	}
404	else {
405	currentThreadSessionInfo = currentThreadSessionInfo.clone()	1✔
406	}
407	currentThreadSessionInfo.getThreadParams.setExtraInfo(key, value)	1✔
408	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)	1✔
409	}
410
411	def threadUnset(key: String): Unit = {	2✔
412	val currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
413	if (currentThreadSessionInfo != null) {	2✔
414	val currentThreadSessionInfoClone = currentThreadSessionInfo.clone()	2✔
415	val threadParams = currentThreadSessionInfoClone.getThreadParams	2✔
416	CarbonSetCommand.unsetValue(threadParams, key)	2✔
417	threadParams.removeExtraInfo(key)	2✔
418	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfoClone)	2✔
419	}
420	}
421
422	def updateSessionInfoToCurrentThread(sparkSession: SparkSession): Unit = {
423	val carbonSessionInfo = CarbonEnv.getInstance(sparkSession).carbonSessionInfo.clone()	2✔
424	val currentThreadSessionInfoOrig = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
425	if (currentThreadSessionInfoOrig != null) {	2✔
426	val currentThreadSessionInfo = currentThreadSessionInfoOrig.clone()	2✔
427	// copy all the thread parameters to apply to session parameters
428	currentThreadSessionInfo.getThreadParams.getAll.asScala	2✔
429	.foreach(entry => carbonSessionInfo.getSessionParams.addProperty(entry._1, entry._2))	2✔
430	carbonSessionInfo.setThreadParams(currentThreadSessionInfo.getThreadParams)	2✔
431	}
432	// preserve thread parameters across call
433	ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)	2✔
434	ThreadLocalSessionInfo
435	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
436	}
437
438	}	2✔

apache / carbondata / 8226

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