9443

Build # 9443

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/2c1fe160a245969575c6e2a4c87081b8dbd1dd62">2c1fe160a<a href="https://github.com/apache/carbondata/commit/2c1fe160a245969575c6e2a4c87081b8dbd1dd62">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/f9cc4dd03a0ca39f647fa502696618da13602c53">f9cc4dd03</a>">f9cc4dd03</a><a href="https://github.com/apache/carbondata/commit/2c1fe160a245969575c6e2a4c87081b8dbd1dd62"> into 0e39abf81">0e39abf81</a>

Run Details

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

61774 of 77064 relevant lines covered (80.16%)

1.06 hits per line

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

84.29

/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 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) => {
        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 = {
    // 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)
  }

  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 = 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.
        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.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 private val 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)	2✔
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	// explain command will output the dataMap information only if enable.query.statistics = true
119	val message = sql(s"EXPLAIN $sqlStatement").collect()	×
120	message(0).getString(0).contains(dataMapName)	×
121	}
122
123	def isSearchModeEnabled: Boolean = carbonStore != null	2✔
124
125	/**
126	* Run SparkSQL directly
127	*/
128	def sparkSql(sqlText: String): DataFrame = {
129	withProfiler(	1✔
130	sqlText,	1✔
131	(qe, sse) => new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	1✔
132	)
133	}
134
135	private def withProfiler(
136	sqlText: String,
137	generateDF: (QueryExecution, SQLStart) => DataFrame): DataFrame = {
138	val sse = SQLStart(sqlText, CarbonSession.statementId.getAndIncrement())	2✔
139	CarbonSession.threadStatementId.set(sse.statementId)	2✔
140	sse.startTime = System.currentTimeMillis()	2✔
141
142	try {
143	val logicalPlan = sessionState.sqlParser.parsePlan(sqlText)	2✔
144	sse.parseEnd = System.currentTimeMillis()	2✔
145
146	val qe = sessionState.executePlan(logicalPlan)	2✔
147	qe.assertAnalyzed()	2✔
148	sse.isCommand = qe.analyzed match {	2✔
149	case c: Command => true	2✔
150	case u @ Union(children) if children.forall(_.isInstanceOf[Command]) => true	2✔
151	case _ => false	2✔
152	}
153	sse.analyzerEnd = System.currentTimeMillis()	2✔
154	generateDF(qe, sse)	2✔
155	} finally {
156	Profiler.invokeIfEnable {	2✔
157	if (sse.isCommand) {	2✔
158	sse.endTime = System.currentTimeMillis()	1✔
159	Profiler.send(sse)	1✔
160	} else {
161	Profiler.addStatementMessage(sse.statementId, sse)	1✔
162	}
163	}
164	}
165	}
166
167	/**
168	* If the query is a simple query with filter, we will try to use Search Mode,
169	* otherwise execute in SparkSQL
170	*/
171	private def trySearchMode(qe: QueryExecution, sse: SQLStart): DataFrame = {
172	val analyzed = qe.analyzed	2✔
173	val LOG = LogServiceFactory.getLogService(this.getClass.getName)	2✔
174	analyzed match {	2✔
175	case _@Project(columns, _@Filter(expr, s: SubqueryAlias))	2✔
176	if s.child.isInstanceOf[LogicalRelation] &&	2✔
177	s.child.asInstanceOf[LogicalRelation].relation	2✔
178	.isInstanceOf[CarbonDatasourceHadoopRelation] =>	2✔
179	LOG.info(s"Search service started and supports filter: ${sse.sqlText}")	2✔
180	runSearch(analyzed, columns, expr, s.child.asInstanceOf[LogicalRelation])	2✔
181	case gl@GlobalLimit(_, ll@LocalLimit(_, p@Project(columns, _@Filter(expr, s: SubqueryAlias))))	1✔
182	if s.child.isInstanceOf[LogicalRelation] &&	1✔
183	s.child.asInstanceOf[LogicalRelation].relation	1✔
184	.isInstanceOf[CarbonDatasourceHadoopRelation] =>	1✔
185	val logicalRelation = s.child.asInstanceOf[LogicalRelation]	1✔
186	LOG.info(s"Search service started and supports limit: ${sse.sqlText}")	1✔
187	runSearch(analyzed, columns, expr, logicalRelation, gl.maxRows, ll.maxRows)	1✔
188	case _ =>	1✔
189	LOG.info(s"Search service started, but don't support: ${sse.sqlText}," +	1✔
190	s" and will run it with SparkSQL")	1✔
191	new Dataset[Row](self, qe, RowEncoder(qe.analyzed.schema))	1✔
192	}
193	}
194
195	@transient private var carbonStore: SparkCarbonStore = _	2✔
196
197	def startSearchMode(): Unit = {	2✔
198	CarbonProperties.enableSearchMode(true)	2✔
199	CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "false")	2✔
200	if (carbonStore == null) {	2✔
201	carbonStore = new SparkCarbonStore(this)	2✔
202	carbonStore.startSearchMode()	2✔
203	}
204	}
205
206	def stopSearchMode(): Unit = {	2✔
207	CarbonProperties.enableSearchMode(false)	2✔
208	CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "true")	2✔
209	if (carbonStore != null) {	2✔
210	try {	×
211	carbonStore.stopSearchMode()	2✔
212	carbonStore = null	2✔
213	} catch {
214	case e: RuntimeException =>
215	LogServiceFactory.getLogService(this.getClass.getCanonicalName)	×
216	.error(s"Stop search mode failed: ${e.getMessage}")	×
217	}
218	}
219	}
220
221	private def runSearch(
222	logicalPlan: LogicalPlan,
223	columns: Seq[NamedExpression],
224	expr: Expression,
225	relation: LogicalRelation,
226	maxRows: Option[Long] = None,	2✔
227	localMaxRows: Option[Long] = None): DataFrame = {	2✔
228	val rows = carbonStore.search(	2✔
229	relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable,	2✔
230	columns.map(_.name).toArray,	2✔
231	if (expr != null) CarbonFilters.transformExpression(expr) else null,	2✔
232	maxRows.getOrElse(Long.MaxValue),	2✔
233	localMaxRows.getOrElse(Long.MaxValue))	2✔
234	val output = new java.util.ArrayList[Row]()	2✔
235	while (rows.hasNext) {	2✔
236	val row = rows.next()	2✔
237	output.add(Row.fromSeq(row.getData))	2✔
238	}
239	createDataFrame(output, logicalPlan.schema)	2✔
240	}
241
242	}
243
244	object CarbonSession {	×
245
246	private val statementId = new AtomicLong(0)	2✔
247
248	private var enableInMemCatlog: Boolean = false	2✔
249
250	private[sql] val threadStatementId = new ThreadLocal[Long]()	2✔
251
252	implicit class CarbonBuilder(builder: Builder) {	2✔
253
254	def enableInMemoryCatalog(): Builder = {
255	enableInMemCatlog = true	×
256	builder	×
257	}
258	def getOrCreateCarbonSession(): SparkSession = {
259	getOrCreateCarbonSession(null, null)	2✔
260	}
261
262	def getOrCreateCarbonSession(storePath: String): SparkSession = {
263	getOrCreateCarbonSession(	×
264	storePath,	×
265	new File(CarbonCommonConstants.METASTORE_LOCATION_DEFAULT_VAL).getCanonicalPath)	×
266	}
267
268	def getOrCreateCarbonSession(storePath: String,
269	metaStorePath: String): SparkSession = synchronized {
270	if (!enableInMemCatlog) {	2✔
271	builder.enableHiveSupport()	2✔
272	}
273	val options =	2✔
274	getValue("options", builder).asInstanceOf[scala.collection.mutable.HashMap[String, String]]	2✔
275	val userSuppliedContext: Option[SparkContext] =	2✔
276	getValue("userSuppliedContext", builder).asInstanceOf[Option[SparkContext]]	2✔
277
278	if (metaStorePath != null) {	2✔
279	val hadoopConf = new Configuration()	2✔
280	val configFile = Utils.getContextOrSparkClassLoader.getResource("hive-site.xml")	2✔
281	if (configFile != null) {	2✔
282	hadoopConf.addResource(configFile)	×
283	}
284	if (options.get(CarbonCommonConstants.HIVE_CONNECTION_URL).isEmpty &&	2✔
285	hadoopConf.get(CarbonCommonConstants.HIVE_CONNECTION_URL) == null) {	2✔
286	val metaStorePathAbsolute = new File(metaStorePath).getCanonicalPath	2✔
287	val hiveMetaStoreDB = metaStorePathAbsolute + "/metastore_db"	2✔
288	options ++= Map[String, String]((CarbonCommonConstants.HIVE_CONNECTION_URL,	2✔
289	s"jdbc:derby:;databaseName=$hiveMetaStoreDB;create=true"))	2✔
290	}
291	}
292
293	// Get the session from current thread's active session.
294	var session: SparkSession = SparkSession.getActiveSession match {	2✔
295	case Some(sparkSession: CarbonSession) =>	2✔
296	if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {	2✔
297	options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }	2✔
298	sparkSession	2✔
299	} else {
300	null	×
301	}
302	case _ => null	2✔
303	}
304	if (session ne null) {	2✔
305	return session	2✔
306	}
307
308	// Global synchronization so we will only set the default session once.
309	SparkSession.synchronized {	2✔
310	// If the current thread does not have an active session, get it from the global session.
311	session = SparkSession.getDefaultSession match {	2✔
312	case Some(sparkSession: CarbonSession) =>	2✔
313	if ((sparkSession ne null) && !sparkSession.sparkContext.isStopped) {	×
314	options.foreach { case (k, v) => sparkSession.sessionState.conf.setConfString(k, v) }	×
315	sparkSession	×
316	} else {
317	null	×
318	}
319	case _ => null	2✔
320	}
321	if (session ne null) {	2✔
322	return session	×
323	}
324
325	// No active nor global default session. Create a new one.
326	val sparkContext = userSuppliedContext.getOrElse {	2✔
327	// set app name if not given
328	val randomAppName = java.util.UUID.randomUUID().toString	2✔
329	val sparkConf = new SparkConf()	2✔
330	options.foreach { case (k, v) => sparkConf.set(k, v) }	2✔
331	if (!sparkConf.contains("spark.app.name")) {	2✔
332	sparkConf.setAppName(randomAppName)	×
333	}
334	val sc = SparkContext.getOrCreate(sparkConf)	2✔
335	// maybe this is an existing SparkContext, update its SparkConf which maybe used
336	// by SparkSession
337	options.foreach { case (k, v) => sc.conf.set(k, v) }	2✔
338	if (!sc.conf.contains("spark.app.name")) {	2✔
339	sc.conf.setAppName(randomAppName)	×
340	}
341	sc	2✔
342	}
343
344	session = new CarbonSession(sparkContext, None, !enableInMemCatlog)	2✔
345	val carbonProperties = CarbonProperties.getInstance()	2✔
346	if (storePath != null) {	2✔
347	carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)	×
348	// In case if it is in carbon.properties for backward compatible
349	} else if (carbonProperties.getProperty(CarbonCommonConstants.STORE_LOCATION) == null) {	2✔
350	carbonProperties.addProperty(CarbonCommonConstants.STORE_LOCATION,	2✔
351	session.sessionState.conf.warehousePath)	2✔
352	}
353	options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) }	2✔
354	SparkSession.setDefaultSession(session)	2✔
355	// Setup monitor end point and register CarbonMonitorListener
356	Profiler.initialize(sparkContext)	2✔
357	// Register a successfully instantiated context to the singleton. This should be at the
358	// end of the class definition so that the singleton is updated only if there is no
359	// exception in the construction of the instance.
360	CarbonToSparkAdapater.addSparkListener(sparkContext)	2✔
361	session.streams.addListener(new CarbonStreamingQueryListener(session))	2✔
362	}
363
364	session	2✔
365	}
366
367	/**
368	* It is a hack to get the private field from class.
369	*/
370	def getValue(name: String, builder: Builder): Any = {
371	val currentMirror = scala.reflect.runtime.currentMirror	2✔
372	val instanceMirror = currentMirror.reflect(builder)	2✔
373	val m = currentMirror.classSymbol(builder.getClass).	2✔
374	toType.members.find { p =>	2✔
375	p.name.toString.equals(name)	2✔
376	}.get.asTerm	2✔
377	instanceMirror.reflectField(m).get	2✔
378	}
379	}
380
381	def threadSet(key: String, value: String): Unit = {	2✔
382	var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
383	if (currentThreadSessionInfo == null) {	2✔
384	currentThreadSessionInfo = new CarbonSessionInfo()	×
385	}
386	else {
387	currentThreadSessionInfo = currentThreadSessionInfo.clone()	2✔
388	}
389	val threadParams = currentThreadSessionInfo.getThreadParams	2✔
390	CarbonSetCommand.validateAndSetValue(threadParams, key, value)	2✔
391	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)	2✔
392	}
393
394
395	def threadSet(key: String, value: Object): Unit = {
396	var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
397	if (currentThreadSessionInfo == null) {	2✔
398	currentThreadSessionInfo = new CarbonSessionInfo()	×
399	}
400	else {
401	currentThreadSessionInfo = currentThreadSessionInfo.clone()	2✔
402	}
403	currentThreadSessionInfo.getThreadParams.setExtraInfo(key, value)	2✔
404	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)	2✔
405	}
406
407	def threadUnset(key: String): Unit = {	2✔
408	val currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
409	if (currentThreadSessionInfo != null) {	2✔
410	val currentThreadSessionInfoClone = currentThreadSessionInfo.clone()	2✔
411	val threadParams = currentThreadSessionInfoClone.getThreadParams	2✔
412	CarbonSetCommand.unsetValue(threadParams, key)	2✔
413	threadParams.removeExtraInfo(key)	2✔
414	ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfoClone)	2✔
415	}
416	}
417
418	def updateSessionInfoToCurrentThread(sparkSession: SparkSession): Unit = {
419	val carbonSessionInfo = CarbonEnv.getInstance(sparkSession).carbonSessionInfo.clone()	2✔
420	val currentThreadSessionInfoOrig = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
421	if (currentThreadSessionInfoOrig != null) {	2✔
422	val currentThreadSessionInfo = currentThreadSessionInfoOrig.clone()	2✔
423	// copy all the thread parameters to apply to session parameters
424	currentThreadSessionInfo.getThreadParams.getAll.asScala	2✔
425	.foreach(entry => carbonSessionInfo.getSessionParams.addProperty(entry._1, entry._2))	2✔
426	carbonSessionInfo.setThreadParams(currentThreadSessionInfo.getThreadParams)	2✔
427	}
428	// preserve thread parameters across call
429	ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)	2✔
430	ThreadLocalSessionInfo
431	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
432	}
433
434	}	2✔

apache / carbondata / 9443

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