6672

Build # 6672

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/6d92b2683dc127fe1397c48fef243bdfcbb3a6af">6d92b2683<a href="https://github.com/apache/carbondata/commit/6d92b2683dc127fe1397c48fef243bdfcbb3a6af">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/d4a8844475467b6919d681bdac5fec42ea43e922">d4a884447</a>">d4a884447</a><a href="https://github.com/apache/carbondata/commit/6d92b2683dc127fe1397c48fef243bdfcbb3a6af"> into 9ee0f3579">9ee0f3579</a>

Run Details

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

57181 of 70654 relevant lines covered (80.93%)

1.08 hits per line

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

76.09

/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.io.File
import java.util.concurrent.ConcurrentHashMap

import scala.util.Try

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.spark.util.CarbonReflectionUtils

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

apache / carbondata / 6672

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