apache / carbondata / 191

Build # 191

Build Type

push

jenkins

Committed by GitHub

Commit Message

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

Run Details

27 of 27 new or added lines in 3 files covered. (100.0%)

57700 of 71451 relevant lines covered (80.75%)

1.07 hits per line

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

77.14

/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.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.util.concurrent.ConcurrentHashMap

import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.catalyst.catalog.SessionCatalog
import org.apache.spark.sql.events.MergeIndexEventListener
import org.apache.spark.sql.execution.command.preaaggregate._
import org.apache.spark.sql.execution.command.timeseries.TimeSeriesFunction
import org.apache.spark.sql.hive._

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util._
import org.apache.carbondata.datamap.{TextMatchMaxDocUDF, TextMatchUDF}
import org.apache.carbondata.events._
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent, LoadTablePostStatusUpdateEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.spark.rdd.SparkReadSupport
import org.apache.carbondata.spark.readsupport.SparkRowReadSupportImpl

/**
 * Carbon Environment for unified context
 */
class CarbonEnv {

  var carbonMetastore: CarbonMetaStore = _

  var sessionParams: SessionParams = _

  var carbonSessionInfo: CarbonSessionInfo = _

  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  // set readsupport class global so that the executor can get it.
  SparkReadSupport.readSupportClass = classOf[SparkRowReadSupportImpl]

  var initialized = false

  def init(sparkSession: SparkSession): Unit = {
    val properties = CarbonProperties.getInstance()
    var storePath = properties.getProperty(CarbonCommonConstants.STORE_LOCATION)
    if (storePath == null) {
      storePath = sparkSession.conf.get("spark.sql.warehouse.dir")
      properties.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)
    }
    LOGGER.info(s"Initializing CarbonEnv, store location: $storePath")

    sparkSession.udf.register("getTupleId", () => "")
    // added for handling preaggregate table creation. when user will fire create ddl for
    // create table we are adding a udf so no need to apply PreAggregate rules.
    sparkSession.udf.register("preAgg", () => "")
    // added to apply proper rules for loading data into pre-agg table. If this UDF is present
    // only then the CarbonPreAggregateDataLoadingRules would be applied to split the average
    // column to sum and count.
    sparkSession.udf.register("preAggLoad", () => "")

    // register for lucene datamap
    // TODO: move it to proper place, it should be registered by datamap implementation
    sparkSession.udf.register("text_match", new TextMatchUDF)
    sparkSession.udf.register("text_match_with_limit", new TextMatchMaxDocUDF)

    // added for handling timeseries function like hour, minute, day , month , year
    sparkSession.udf.register("timeseries", new TimeSeriesFunction)
    // acquiring global level lock so global configuration will be updated by only one thread
    CarbonEnv.carbonEnvMap.synchronized {
      if (!initialized) {
        // update carbon session parameters , preserve thread parameters
        val currentThreadSesssionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
        carbonSessionInfo = new CarbonSessionInfo()
        // We should not corrupt the information in carbonSessionInfo object which is at the
        // session level. Instead create a new object and in that set the user specified values in
        // thread/session params
        val threadLevelCarbonSessionInfo = new CarbonSessionInfo()
        if (currentThreadSesssionInfo != null) {
          threadLevelCarbonSessionInfo.setThreadParams(currentThreadSesssionInfo.getThreadParams)
        }
        ThreadLocalSessionInfo.setCarbonSessionInfo(threadLevelCarbonSessionInfo)
        ThreadLocalSessionInfo.setConfigurationToCurrentThread(sparkSession
          .sessionState.newHadoopConf())
        val config = new CarbonSQLConf(sparkSession)
        if (sparkSession.conf.getOption(CarbonCommonConstants.ENABLE_UNSAFE_SORT).isEmpty) {
          config.addDefaultCarbonParams()
        }
        // add session params after adding DefaultCarbonParams
        config.addDefaultCarbonSessionParams()
        carbonMetastore = {
          // trigger event for CarbonEnv create
          val operationContext = new OperationContext
          val carbonEnvInitPreEvent: CarbonEnvInitPreEvent =
            CarbonEnvInitPreEvent(sparkSession, carbonSessionInfo, storePath)
          OperationListenerBus.getInstance.fireEvent(carbonEnvInitPreEvent, operationContext)

          CarbonMetaStoreFactory.createCarbonMetaStore(sparkSession.conf)
        }
        CarbonProperties.getInstance
          .addNonSerializableProperty(CarbonCommonConstants.IS_DRIVER_INSTANCE, "true")
        initialized = true
      }
    }
    LOGGER.info("Initialize CarbonEnv completed...")
  }
}

object CarbonEnv {

  val carbonEnvMap = new ConcurrentHashMap[SparkSession, CarbonEnv]

  val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)

  def getInstance(sparkSession: SparkSession): CarbonEnv = {
    if (sparkSession.isInstanceOf[CarbonSession]) {
      sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog].getCarbonEnv
    } else {
      var carbonEnv: CarbonEnv = carbonEnvMap.get(sparkSession)
      if (carbonEnv == null) {
        carbonEnv = new CarbonEnv
        carbonEnv.init(sparkSession)
        carbonEnvMap.put(sparkSession, carbonEnv)
      }
      carbonEnv
    }
  }

  /**
   * Method
   * 1. To initialize Listeners to their respective events in the OperationListenerBus
   * 2. To register common listeners
   *
   */
  def init(sparkSession: SparkSession): Unit = {
    initListeners
  }

  /**
   * Method to initialize Listeners to their respective events in the OperationListenerBus.
   */
  def initListeners(): Unit = {
    OperationListenerBus.getInstance()
      .addListener(classOf[LoadTablePreStatusUpdateEvent], LoadPostAggregateListener)
      .addListener(classOf[DeleteSegmentByIdPreEvent], PreAggregateDeleteSegmentByIdPreListener)
      .addListener(classOf[DeleteSegmentByDatePreEvent], PreAggregateDeleteSegmentByDatePreListener)
      .addListener(classOf[UpdateTablePreEvent], UpdatePreAggregatePreListener)
      .addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)
      .addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)
      .addListener(classOf[AlterTableDropColumnPreEvent], PreAggregateDropColumnPreListener)
      .addListener(classOf[AlterTableRenamePreEvent], RenameTablePreListener)
      .addListener(classOf[AlterTableDataTypeChangePreEvent], PreAggregateDataTypeChangePreListener)
      .addListener(classOf[AlterTableAddColumnPreEvent], PreAggregateAddColumnsPreListener)
      .addListener(classOf[LoadTablePreExecutionEvent], LoadPreAggregateTablePreListener)
      .addListener(classOf[AlterTableCompactionPreStatusUpdateEvent],
        AlterPreAggregateTableCompactionPostListener)
      .addListener(classOf[LoadMetadataEvent], LoadProcessMetaListener)
      .addListener(classOf[LoadMetadataEvent], CompactionProcessMetaListener)
      .addListener(classOf[LoadTablePostStatusUpdateEvent], CommitPreAggregateListener)
      .addListener(classOf[AlterTableCompactionPostStatusUpdateEvent], CommitPreAggregateListener)
      .addListener(classOf[AlterTableDropPartitionPreStatusEvent],
        AlterTableDropPartitionPreStatusListener)
      .addListener(classOf[AlterTableDropPartitionPostStatusEvent],
        AlterTableDropPartitionPostStatusListener)
      .addListener(classOf[AlterTableDropPartitionMetaEvent], AlterTableDropPartitionMetaListener)
      .addListener(classOf[LoadTablePostExecutionEvent], new MergeIndexEventListener)
      .addListener(classOf[AlterTableCompactionPostEvent], new MergeIndexEventListener)
      .addListener(classOf[AlterTableMergeIndexEvent], new MergeIndexEventListener)
  }

  /**
   * Return carbon table instance from cache or by looking up table in `sparkSession`
   */
  def getCarbonTable(
      databaseNameOp: Option[String],
      tableName: String)
    (sparkSession: SparkSession): CarbonTable = {
    refreshRelationFromCache(TableIdentifier(tableName, databaseNameOp))(sparkSession)
    val databaseName = getDatabaseName(databaseNameOp)(sparkSession)
    val catalog = getInstance(sparkSession).carbonMetastore
    // refresh cache
    catalog.checkSchemasModifiedTimeAndReloadTable(TableIdentifier(tableName, databaseNameOp))

    // try to get it from catch, otherwise lookup in catalog
    catalog.getTableFromMetadataCache(databaseName, tableName)
      .getOrElse(
        catalog
          .lookupRelation(databaseNameOp, tableName)(sparkSession)
          .asInstanceOf[CarbonRelation]
          .carbonTable)
  }

  def refreshRelationFromCache(identifier: TableIdentifier)(sparkSession: SparkSession): Boolean = {
    var isRefreshed = false
    val carbonEnv = getInstance(sparkSession)
    val table = carbonEnv.carbonMetastore.getTableFromMetadataCache(
      identifier.database.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase),
      identifier.table)
    if (carbonEnv.carbonMetastore
          .checkSchemasModifiedTimeAndReloadTable(identifier)  && table.isDefined) {
      sparkSession.sessionState.catalog.refreshTable(identifier)
      val tablePath = table.get.getTablePath
      DataMapStoreManager.getInstance().
        clearDataMaps(AbsoluteTableIdentifier.from(tablePath,
          identifier.database.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase),
          identifier.table, table.get.getTableInfo.getFactTable.getTableId))
      isRefreshed = true
    }
    isRefreshed
  }

  /**
   * Return carbon table instance by looking up table in `sparkSession`
   */
  def getCarbonTable(
      tableIdentifier: TableIdentifier)
    (sparkSession: SparkSession): CarbonTable = {
    getCarbonTable(tableIdentifier.database, tableIdentifier.table)(sparkSession)
  }

  /**
   * Return database name or get default name from sparkSession
   */
  def getDatabaseName(
      databaseNameOp: Option[String]
  )(sparkSession: SparkSession): String = {
    databaseNameOp.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase)
  }

  /**
   * The method returns the database location
   * if carbon.storeLocation does  point to spark.sql.warehouse.dir then returns
   * the database locationUri as database location else follows the old behaviour
   * making database location from carbon fixed store and database name.
   * @return database location
   */
  def getDatabaseLocation(dbName: String, sparkSession: SparkSession): String = {
    var databaseLocation =
      sparkSession.sessionState.catalog.asInstanceOf[SessionCatalog].getDatabaseMetadata(dbName)
        .locationUri.toString
    // for default database and db ends with .db
    // check whether the carbon store and hive store is same or different.
    if (dbName.equals("default") || databaseLocation.endsWith(".db")) {
      val properties = CarbonProperties.getInstance()
      val carbonStorePath =
        FileFactory.getUpdatedFilePath(properties.getProperty(CarbonCommonConstants.STORE_LOCATION))
      val hiveStorePath =
        FileFactory.getUpdatedFilePath(sparkSession.conf.get("spark.sql.warehouse.dir"))
      // if carbon.store does not point to spark.sql.warehouse.dir then follow the old table path
      // format
      if (!hiveStorePath.equals(carbonStorePath)) {
        databaseLocation = CarbonProperties.getStorePath +
                           CarbonCommonConstants.FILE_SEPARATOR +
                           dbName
      }
    }

    FileFactory.getUpdatedFilePath(databaseLocation)
  }

  /**
   * Return table path from carbon table. If table does not exist, construct it using
   * database location and table name
   */
  def getTablePath(
      databaseNameOp: Option[String],
      tableName: String
  )(sparkSession: SparkSession): String = {
    try {
      getCarbonTable(databaseNameOp, tableName)(sparkSession).getTablePath
    } catch {
      case _: NoSuchTableException =>
        val dbName = getDatabaseName(databaseNameOp)(sparkSession)
        val dbLocation = getDatabaseLocation(dbName, sparkSession)
        dbLocation + CarbonCommonConstants.FILE_SEPARATOR + tableName
    }
  }

  def getIdentifier(
      databaseNameOp: Option[String],
      tableName: String
  )(sparkSession: SparkSession): AbsoluteTableIdentifier = {
    AbsoluteTableIdentifier.from(
      getTablePath(databaseNameOp, tableName)(sparkSession),
      getDatabaseName(databaseNameOp)(sparkSession),
      tableName)
  }

  def getThreadParam(key: String, defaultValue: String) : String = {
    val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (null != carbonSessionInfo) {
      carbonSessionInfo.getThreadParams.getProperty(key, defaultValue)
    } else {
      defaultValue
    }
  }

  def getSessionParam(key: String, defaultValue: String) : String = {
    val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (null != carbonSessionInfo) {
      carbonSessionInfo.getThreadParams.getProperty(key, defaultValue)
    } else {
      defaultValue
    }
  }

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql
19
20	import java.util.concurrent.ConcurrentHashMap
21
22	import org.apache.spark.sql.catalyst.TableIdentifier
23	import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
24	import org.apache.spark.sql.catalyst.catalog.SessionCatalog
25	import org.apache.spark.sql.events.MergeIndexEventListener
26	import org.apache.spark.sql.execution.command.preaaggregate._
27	import org.apache.spark.sql.execution.command.timeseries.TimeSeriesFunction
28	import org.apache.spark.sql.hive._
29
30	import org.apache.carbondata.common.logging.LogServiceFactory
31	import org.apache.carbondata.core.constants.CarbonCommonConstants
32	import org.apache.carbondata.core.datamap.DataMapStoreManager
33	import org.apache.carbondata.core.datastore.impl.FileFactory
34	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
35	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
36	import org.apache.carbondata.core.util._
37	import org.apache.carbondata.datamap.{TextMatchMaxDocUDF, TextMatchUDF}
38	import org.apache.carbondata.events._
39	import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent, LoadTablePostStatusUpdateEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
40	import org.apache.carbondata.spark.rdd.SparkReadSupport
41	import org.apache.carbondata.spark.readsupport.SparkRowReadSupportImpl
42
43	/**
44	* Carbon Environment for unified context
45	*/
46	class CarbonEnv {
47
48	var carbonMetastore: CarbonMetaStore = _	2✔
49
50	var sessionParams: SessionParams = _	×
51
52	var carbonSessionInfo: CarbonSessionInfo = _	2✔
53
54	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	2✔
55
56	// set readsupport class global so that the executor can get it.
57	SparkReadSupport.readSupportClass = classOf[SparkRowReadSupportImpl]	2✔
58
59	var initialized = false	2✔
60
61	def init(sparkSession: SparkSession): Unit = {
62	val properties = CarbonProperties.getInstance()	2✔
63	var storePath = properties.getProperty(CarbonCommonConstants.STORE_LOCATION)	2✔
64	if (storePath == null) {	2✔
65	storePath = sparkSession.conf.get("spark.sql.warehouse.dir")	×
66	properties.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)	×
67	}
68	LOGGER.info(s"Initializing CarbonEnv, store location: $storePath")	2✔
69
70	sparkSession.udf.register("getTupleId", () => "")	2✔
71	// added for handling preaggregate table creation. when user will fire create ddl for
72	// create table we are adding a udf so no need to apply PreAggregate rules.
73	sparkSession.udf.register("preAgg", () => "")	2✔
74	// added to apply proper rules for loading data into pre-agg table. If this UDF is present
75	// only then the CarbonPreAggregateDataLoadingRules would be applied to split the average
76	// column to sum and count.
77	sparkSession.udf.register("preAggLoad", () => "")	2✔
78
79	// register for lucene datamap
80	// TODO: move it to proper place, it should be registered by datamap implementation
81	sparkSession.udf.register("text_match", new TextMatchUDF)	2✔
82	sparkSession.udf.register("text_match_with_limit", new TextMatchMaxDocUDF)	2✔
83
84	// added for handling timeseries function like hour, minute, day , month , year
85	sparkSession.udf.register("timeseries", new TimeSeriesFunction)	2✔
86	// acquiring global level lock so global configuration will be updated by only one thread
87	CarbonEnv.carbonEnvMap.synchronized {	2✔
88	if (!initialized) {	2✔
89	// update carbon session parameters , preserve thread parameters
90	val currentThreadSesssionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	2✔
91	carbonSessionInfo = new CarbonSessionInfo()	2✔
92	// We should not corrupt the information in carbonSessionInfo object which is at the
93	// session level. Instead create a new object and in that set the user specified values in
94	// thread/session params
95	val threadLevelCarbonSessionInfo = new CarbonSessionInfo()	2✔
96	if (currentThreadSesssionInfo != null) {	2✔
97	threadLevelCarbonSessionInfo.setThreadParams(currentThreadSesssionInfo.getThreadParams)	2✔
98	}
99	ThreadLocalSessionInfo.setCarbonSessionInfo(threadLevelCarbonSessionInfo)	2✔
100	ThreadLocalSessionInfo.setConfigurationToCurrentThread(sparkSession	2✔
101	.sessionState.newHadoopConf())	2✔
102	val config = new CarbonSQLConf(sparkSession)	2✔
103	if (sparkSession.conf.getOption(CarbonCommonConstants.ENABLE_UNSAFE_SORT).isEmpty) {	2✔
104	config.addDefaultCarbonParams()	2✔
105	}
106	// add session params after adding DefaultCarbonParams
107	config.addDefaultCarbonSessionParams()	2✔
108	carbonMetastore = {	2✔
109	// trigger event for CarbonEnv create
110	val operationContext = new OperationContext	2✔
111	val carbonEnvInitPreEvent: CarbonEnvInitPreEvent =	2✔
112	CarbonEnvInitPreEvent(sparkSession, carbonSessionInfo, storePath)	2✔
113	OperationListenerBus.getInstance.fireEvent(carbonEnvInitPreEvent, operationContext)	2✔
114
115	CarbonMetaStoreFactory.createCarbonMetaStore(sparkSession.conf)	2✔
116	}
117	CarbonProperties.getInstance	2✔
118	.addNonSerializableProperty(CarbonCommonConstants.IS_DRIVER_INSTANCE, "true")	2✔
119	initialized = true	2✔
120	}
121	}
122	LOGGER.info("Initialize CarbonEnv completed...")	2✔
123	}
124	}
125
126	object CarbonEnv {	2✔
127
128	val carbonEnvMap = new ConcurrentHashMap[SparkSession, CarbonEnv]	2✔
129
130	val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	2✔
131
132	def getInstance(sparkSession: SparkSession): CarbonEnv = {
133	if (sparkSession.isInstanceOf[CarbonSession]) {	2✔
134	sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog].getCarbonEnv	2✔
135	} else {
136	var carbonEnv: CarbonEnv = carbonEnvMap.get(sparkSession)	×
137	if (carbonEnv == null) {	×
138	carbonEnv = new CarbonEnv	×
139	carbonEnv.init(sparkSession)	×
140	carbonEnvMap.put(sparkSession, carbonEnv)	×
141	}
142	carbonEnv	×
143	}
144	}
145
146	/**
147	* Method
148	* 1. To initialize Listeners to their respective events in the OperationListenerBus
149	* 2. To register common listeners
150	*
151	*/
152	def init(sparkSession: SparkSession): Unit = {
153	initListeners	×
154	}
155
156	/**
157	* Method to initialize Listeners to their respective events in the OperationListenerBus.
158	*/
159	def initListeners(): Unit = {
160	OperationListenerBus.getInstance()	2✔
161	.addListener(classOf[LoadTablePreStatusUpdateEvent], LoadPostAggregateListener)	2✔
162	.addListener(classOf[DeleteSegmentByIdPreEvent], PreAggregateDeleteSegmentByIdPreListener)	2✔
163	.addListener(classOf[DeleteSegmentByDatePreEvent], PreAggregateDeleteSegmentByDatePreListener)	2✔
164	.addListener(classOf[UpdateTablePreEvent], UpdatePreAggregatePreListener)	2✔
165	.addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)	2✔
166	.addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)	2✔
167	.addListener(classOf[AlterTableDropColumnPreEvent], PreAggregateDropColumnPreListener)	2✔
168	.addListener(classOf[AlterTableRenamePreEvent], RenameTablePreListener)	2✔
169	.addListener(classOf[AlterTableDataTypeChangePreEvent], PreAggregateDataTypeChangePreListener)	2✔
170	.addListener(classOf[AlterTableAddColumnPreEvent], PreAggregateAddColumnsPreListener)	2✔
171	.addListener(classOf[LoadTablePreExecutionEvent], LoadPreAggregateTablePreListener)	2✔
172	.addListener(classOf[AlterTableCompactionPreStatusUpdateEvent],	2✔
173	AlterPreAggregateTableCompactionPostListener)	2✔
174	.addListener(classOf[LoadMetadataEvent], LoadProcessMetaListener)	2✔
175	.addListener(classOf[LoadMetadataEvent], CompactionProcessMetaListener)	2✔
176	.addListener(classOf[LoadTablePostStatusUpdateEvent], CommitPreAggregateListener)	2✔
177	.addListener(classOf[AlterTableCompactionPostStatusUpdateEvent], CommitPreAggregateListener)	2✔
178	.addListener(classOf[AlterTableDropPartitionPreStatusEvent],	2✔
179	AlterTableDropPartitionPreStatusListener)	2✔
180	.addListener(classOf[AlterTableDropPartitionPostStatusEvent],	2✔
181	AlterTableDropPartitionPostStatusListener)	2✔
182	.addListener(classOf[AlterTableDropPartitionMetaEvent], AlterTableDropPartitionMetaListener)	2✔
183	.addListener(classOf[LoadTablePostExecutionEvent], new MergeIndexEventListener)	2✔
184	.addListener(classOf[AlterTableCompactionPostEvent], new MergeIndexEventListener)	2✔
185	.addListener(classOf[AlterTableMergeIndexEvent], new MergeIndexEventListener)	2✔
186	}
187
188	/**
189	* Return carbon table instance from cache or by looking up table in `sparkSession`
190	*/
191	def getCarbonTable(
192	databaseNameOp: Option[String],
193	tableName: String)
194	(sparkSession: SparkSession): CarbonTable = {
195	refreshRelationFromCache(TableIdentifier(tableName, databaseNameOp))(sparkSession)	2✔
196	val databaseName = getDatabaseName(databaseNameOp)(sparkSession)	2✔
197	val catalog = getInstance(sparkSession).carbonMetastore	2✔
198	// refresh cache
199	catalog.checkSchemasModifiedTimeAndReloadTable(TableIdentifier(tableName, databaseNameOp))	2✔
200
201	// try to get it from catch, otherwise lookup in catalog
202	catalog.getTableFromMetadataCache(databaseName, tableName)	2✔
203	.getOrElse(	2✔
204	catalog	2✔
205	.lookupRelation(databaseNameOp, tableName)(sparkSession)	×
206	.asInstanceOf[CarbonRelation]
207	.carbonTable)	2✔
208	}
209
210	def refreshRelationFromCache(identifier: TableIdentifier)(sparkSession: SparkSession): Boolean = {
211	var isRefreshed = false	2✔
212	val carbonEnv = getInstance(sparkSession)	2✔
213	val table = carbonEnv.carbonMetastore.getTableFromMetadataCache(	2✔
214	identifier.database.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase),	2✔
215	identifier.table)	2✔
216	if (carbonEnv.carbonMetastore	2✔
217	.checkSchemasModifiedTimeAndReloadTable(identifier) && table.isDefined) {	2✔
218	sparkSession.sessionState.catalog.refreshTable(identifier)	×
219	val tablePath = table.get.getTablePath	×
220	DataMapStoreManager.getInstance().	×
221	clearDataMaps(AbsoluteTableIdentifier.from(tablePath,	×
222	identifier.database.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase),	×
223	identifier.table, table.get.getTableInfo.getFactTable.getTableId))	×
224	isRefreshed = true	×
225	}
226	isRefreshed	2✔
227	}
228
229	/**
230	* Return carbon table instance by looking up table in `sparkSession`
231	*/
232	def getCarbonTable(
233	tableIdentifier: TableIdentifier)
234	(sparkSession: SparkSession): CarbonTable = {
235	getCarbonTable(tableIdentifier.database, tableIdentifier.table)(sparkSession)	2✔
236	}
237
238	/**
239	* Return database name or get default name from sparkSession
240	*/
241	def getDatabaseName(
242	databaseNameOp: Option[String]
243	)(sparkSession: SparkSession): String = {
244	databaseNameOp.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase)	2✔
245	}
246
247	/**
248	* The method returns the database location
249	* if carbon.storeLocation does point to spark.sql.warehouse.dir then returns
250	* the database locationUri as database location else follows the old behaviour
251	* making database location from carbon fixed store and database name.
252	* @return database location
253	*/
254	def getDatabaseLocation(dbName: String, sparkSession: SparkSession): String = {
255	var databaseLocation =	2✔
256	sparkSession.sessionState.catalog.asInstanceOf[SessionCatalog].getDatabaseMetadata(dbName)	2✔
257	.locationUri.toString	2✔
258	// for default database and db ends with .db
259	// check whether the carbon store and hive store is same or different.
260	if (dbName.equals("default") \|\| databaseLocation.endsWith(".db")) {	2✔
261	val properties = CarbonProperties.getInstance()	2✔
262	val carbonStorePath =	2✔
263	FileFactory.getUpdatedFilePath(properties.getProperty(CarbonCommonConstants.STORE_LOCATION))	2✔
264	val hiveStorePath =	2✔
265	FileFactory.getUpdatedFilePath(sparkSession.conf.get("spark.sql.warehouse.dir"))	2✔
266	// if carbon.store does not point to spark.sql.warehouse.dir then follow the old table path
267	// format
268	if (!hiveStorePath.equals(carbonStorePath)) {	2✔
269	databaseLocation = CarbonProperties.getStorePath +	2✔
270	CarbonCommonConstants.FILE_SEPARATOR +	2✔
271	dbName	2✔
272	}
273	}
274
275	FileFactory.getUpdatedFilePath(databaseLocation)	2✔
276	}
277
278	/**
279	* Return table path from carbon table. If table does not exist, construct it using
280	* database location and table name
281	*/
282	def getTablePath(
283	databaseNameOp: Option[String],
284	tableName: String
285	)(sparkSession: SparkSession): String = {
286	try {	2✔
287	getCarbonTable(databaseNameOp, tableName)(sparkSession).getTablePath	2✔
288	} catch {
289	case _: NoSuchTableException =>
290	val dbName = getDatabaseName(databaseNameOp)(sparkSession)	2✔
291	val dbLocation = getDatabaseLocation(dbName, sparkSession)	2✔
292	dbLocation + CarbonCommonConstants.FILE_SEPARATOR + tableName	2✔
293	}
294	}
295
296	def getIdentifier(
297	databaseNameOp: Option[String],
298	tableName: String
299	)(sparkSession: SparkSession): AbsoluteTableIdentifier = {
300	AbsoluteTableIdentifier.from(	×
301	getTablePath(databaseNameOp, tableName)(sparkSession),	×
302	getDatabaseName(databaseNameOp)(sparkSession),	×
303	tableName)	×
304	}
305
306	def getThreadParam(key: String, defaultValue: String) : String = {	×
307	val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	×
308	if (null != carbonSessionInfo) {	×
309	carbonSessionInfo.getThreadParams.getProperty(key, defaultValue)	×
310	} else {
311	defaultValue	×
312	}
313	}
314
315	def getSessionParam(key: String, defaultValue: String) : String = {	×
316	val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo	×
317	if (null != carbonSessionInfo) {	×
318	carbonSessionInfo.getThreadParams.getProperty(key, defaultValue)	×
319	} else {
320	defaultValue	×
321	}
322	}
323
324	}	2✔

apache / carbondata / 191

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