12132

Build # 12132

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c8c8893c986a24a33d487d7a45f6a6a9fd4b375a">c8c8893c9<a href="https://github.com/apache/carbondata/commit/c8c8893c986a24a33d487d7a45f6a6a9fd4b375a">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/30a720c4cb34cca53333f7af5c0e9bd1386f6dd6">30a720c4c</a>">30a720c4c</a><a href="https://github.com/apache/carbondata/commit/c8c8893c986a24a33d487d7a45f6a6a9fd4b375a"> into c0d8d347a">c0d8d347a</a>

Run Details

57 of 57 new or added lines in 1 file covered. (100.0%)

65055 of 83699 relevant lines covered (77.72%)

1.02 hits per line

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

94.12

/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionState.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.hive

import java.util.concurrent.Callable

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.{Analyzer, FunctionRegistry}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.optimizer.Optimizer
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, PreWriteCheck, ResolveSQLOnFile, _}
import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStrategy, StreamingTableStrategy}
import org.apache.spark.sql.hive.client.HiveClient
import org.apache.spark.sql.internal.{SQLConf, SessionState}
import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonLateDecodeRule, CarbonUDFTransformRule}
import org.apache.spark.sql.parser.CarbonSparkSqlParser

import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.spark.util.CarbonScalaUtil

/**
 * This class will have carbon catalog and refresh the relation from cache if the carbontable in
 * carbon catalog is not same as cached carbon relation's carbon table
 *
 * @param externalCatalog
 * @param globalTempViewManager
 * @param sparkSession
 * @param functionResourceLoader
 * @param functionRegistry
 * @param conf
 * @param hadoopConf
 */
class CarbonHiveSessionCatalog(
    externalCatalog: HiveExternalCatalog,
    globalTempViewManager: GlobalTempViewManager,
    functionRegistry: FunctionRegistry,
    sparkSession: SparkSession,
    conf: SQLConf,
    hadoopConf: Configuration,
    parser: ParserInterface,
    functionResourceLoader: FunctionResourceLoader)
  extends HiveSessionCatalog (
    externalCatalog,
    globalTempViewManager,
    new HiveMetastoreCatalog(sparkSession),
    functionRegistry,
    conf,
    hadoopConf,
    parser,
    functionResourceLoader
  ) with CarbonSessionCatalog {

  private lazy val carbonEnv = {
    val env = new CarbonEnv
    env.init(sparkSession)
    env
  }
  /**
   * return's the carbonEnv instance
   * @return
   */
  override def getCarbonEnv() : CarbonEnv = {
    carbonEnv
  }

  // Initialize all listeners to the Operation bus.
  CarbonEnv.init

  override def lookupRelation(name: TableIdentifier): LogicalPlan = {
    var rtnRelation = super.lookupRelation(name)
    val isRelationRefreshed =
      CarbonSessionUtil.refreshRelationAndSetStats(rtnRelation, name)(sparkSession)
    if (isRelationRefreshed) {
      rtnRelation = super.lookupRelation(name)
      // Reset the stats after lookup.
      CarbonSessionUtil.refreshRelationAndSetStats(rtnRelation, name)(sparkSession)
    }
    rtnRelation
  }

  override def getCachedPlan(t: QualifiedTableName,
      c: Callable[LogicalPlan]): LogicalPlan = {
    val plan = super.getCachedPlan(t, c)
    CarbonSessionUtil.updateCachedPlan(plan)
  }

  /**
   * returns hive client from HiveExternalCatalog
   *
   * @return
   */
  override def getClient(): org.apache.spark.sql.hive.client.HiveClient = {
    sparkSession.asInstanceOf[CarbonSession].sharedState.externalCatalog
      .asInstanceOf[HiveExternalCatalog].client
  }

  override def alterAddColumns(tableIdentifier: TableIdentifier,
      schemaParts: String,
      cols: Option[Seq[ColumnSchema]]): Unit = {
    updateCatalogTableForAlter(tableIdentifier, schemaParts, cols)
  }

  override def alterDropColumns(tableIdentifier: TableIdentifier,
      schemaParts: String,
      cols: Option[Seq[ColumnSchema]]): Unit = {
    updateCatalogTableForAlter(tableIdentifier, schemaParts, cols)
  }

  override def alterColumnChangeDataTypeOrRename(tableIdentifier: TableIdentifier,
      schemaParts: String,
      cols: Option[Seq[ColumnSchema]]): Unit = {
    updateCatalogTableForAlter(tableIdentifier, schemaParts, cols)
  }

  /**
   * This method alters table to set serde properties and updates the catalog table with new updated
   * schema for all the alter operations like add column, drop column, change datatype or rename
   * column
   * @param tableIdentifier
   * @param schemaParts
   * @param cols
   */
  private def updateCatalogTableForAlter(tableIdentifier: TableIdentifier,
      schemaParts: String,
      cols: Option[Seq[ColumnSchema]]): Unit = {
    alterTable(tableIdentifier, schemaParts, cols)
    CarbonSessionUtil
      .alterExternalCatalogForTableWithUpdatedSchema(tableIdentifier,
        cols,
        schemaParts,
        sparkSession)
  }

  override def createPartitions(
      tableName: TableIdentifier,
      parts: Seq[CatalogTablePartition],
      ignoreIfExists: Boolean): Unit = {
    try {
      val table = CarbonEnv.getCarbonTable(tableName)(sparkSession)
      val updatedParts = CarbonScalaUtil.updatePartitions(parts, table)
      super.createPartitions(tableName, updatedParts, ignoreIfExists)
    } catch {
      case e: Exception =>
        super.createPartitions(tableName, parts, ignoreIfExists)
    }
  }

  /**
   * This is alternate way of getting partition information. It first fetches all partitions from
   * hive and then apply filter instead of querying hive along with filters.
   * @param partitionFilters
   * @param sparkSession
   * @param identifier
   * @return
   */
  override def getPartitionsAlternate(partitionFilters: Seq[Expression],
      sparkSession: SparkSession,
      identifier: TableIdentifier) = {
    CarbonSessionUtil.prunePartitionsByFilter(partitionFilters, sparkSession, identifier)
  }

  /**
   * Update the storageformat with new location information
   */
  override def updateStorageLocation(
      path: Path,
      storage: CatalogStorageFormat,
      newTableName: String,
      dbName: String): CatalogStorageFormat = {
    storage.copy(locationUri = Some(path.toUri))
  }
}

/**
 * Session state implementation to override sql parser and adding strategies
 *
 * @param sparkSession
 */
class CarbonSessionStateBuilder(sparkSession: SparkSession,
    parentState: Option[SessionState] = None)
  extends HiveSessionStateBuilder(sparkSession, parentState) {

  override lazy val sqlParser: ParserInterface = new CarbonSparkSqlParser(conf, sparkSession)

  experimentalMethods.extraStrategies =
    Seq(new StreamingTableStrategy(sparkSession),
        new CarbonLateDecodeStrategy,
        new DDLStrategy(sparkSession)
    )
  experimentalMethods.extraOptimizations = Seq(new CarbonIUDRule,
    new CarbonUDFTransformRule,
    new CarbonLateDecodeRule)

  /**
   * Internal catalog for managing table and database states.
   */
  /**
   * Create a [[CarbonSessionStateBuilder]].
   */
  override protected lazy val catalog: CarbonHiveSessionCatalog = {
    val catalog = new CarbonHiveSessionCatalog(
      externalCatalog,
      session.sharedState.globalTempViewManager,
      functionRegistry,
      sparkSession,
      conf,
      SessionState.newHadoopConf(session.sparkContext.hadoopConfiguration, conf),
      sqlParser,
      resourceLoader)
    parentState.foreach(_.catalog.copyStateTo(catalog))
    catalog
  }

  private def externalCatalog: HiveExternalCatalog =
    session.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog]

  /**
   * Create a Hive aware resource loader.
   */
  override protected lazy val resourceLoader: HiveSessionResourceLoader = {
    val client: HiveClient = externalCatalog.client.newSession()
    new HiveSessionResourceLoader(session, client)
  }

  override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)

  override protected def analyzer: Analyzer = {
    new CarbonAnalyzer(catalog,
      conf,
      sparkSession,
      getAnalyzer(super.analyzer))
  }

  /**
   * This method adds carbon rules to Hive Analyzer and returns new analyzer
   * @param analyzer hiveSessionStateBuilder analyzer
   * @return
   */
  def getAnalyzer(analyzer: Analyzer): Analyzer = {
    new Analyzer(catalog, conf) {

      override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
        analyzer.extendedResolutionRules ++
        Seq(CarbonIUDAnalysisRule(sparkSession)) ++
        Seq(CarbonPreInsertionCasts(sparkSession)) ++ customResolutionRules

      override val extendedCheckRules: Seq[LogicalPlan => Unit] =
        analyzer.extendedCheckRules

      override val postHocResolutionRules: Seq[Rule[LogicalPlan]] =
        analyzer.postHocResolutionRules
    }
  }

  override protected def newBuilder: NewBuilder = new CarbonSessionStateBuilder(_, _)
}

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.hive
18
19	import java.util.concurrent.Callable
20
21	import org.apache.hadoop.conf.Configuration
22	import org.apache.hadoop.fs.Path
23	import org.apache.spark.sql._
24	import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
25	import org.apache.spark.sql.catalyst.analysis.{Analyzer, FunctionRegistry}
26	import org.apache.spark.sql.catalyst.catalog._
27	import org.apache.spark.sql.catalyst.expressions.Expression
28	import org.apache.spark.sql.catalyst.optimizer.Optimizer
29	import org.apache.spark.sql.catalyst.parser.ParserInterface
30	import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
31	import org.apache.spark.sql.catalyst.rules.Rule
32	import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, PreWriteCheck, ResolveSQLOnFile, _}
33	import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStrategy, StreamingTableStrategy}
34	import org.apache.spark.sql.hive.client.HiveClient
35	import org.apache.spark.sql.internal.{SQLConf, SessionState}
36	import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonLateDecodeRule, CarbonUDFTransformRule}
37	import org.apache.spark.sql.parser.CarbonSparkSqlParser
38
39	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
40	import org.apache.carbondata.spark.util.CarbonScalaUtil
41
42	/**
43	* This class will have carbon catalog and refresh the relation from cache if the carbontable in
44	* carbon catalog is not same as cached carbon relation's carbon table
45	*
46	* @param externalCatalog
47	* @param globalTempViewManager
48	* @param sparkSession
49	* @param functionResourceLoader
50	* @param functionRegistry
51	* @param conf
52	* @param hadoopConf
53	*/
54	class CarbonHiveSessionCatalog(	2✔
55	externalCatalog: HiveExternalCatalog,
56	globalTempViewManager: GlobalTempViewManager,
57	functionRegistry: FunctionRegistry,
58	sparkSession: SparkSession,	2✔
59	conf: SQLConf,
60	hadoopConf: Configuration,
61	parser: ParserInterface,
62	functionResourceLoader: FunctionResourceLoader)
63	extends HiveSessionCatalog (	2✔
64	externalCatalog,	2✔
65	globalTempViewManager,	2✔
66	new HiveMetastoreCatalog(sparkSession),	2✔
67	functionRegistry,	2✔
68	conf,	2✔
69	hadoopConf,	2✔
70	parser,	2✔
71	functionResourceLoader	2✔
72	) with CarbonSessionCatalog {
73
74	private lazy val carbonEnv = {	2✔
75	val env = new CarbonEnv	2✔
76	env.init(sparkSession)	2✔
77	env	2✔
78	}
79	/**
80	* return's the carbonEnv instance
81	* @return
82	*/
83	override def getCarbonEnv() : CarbonEnv = {
84	carbonEnv	2✔
85	}
86
87	// Initialize all listeners to the Operation bus.
88	CarbonEnv.init	2✔
89
90	override def lookupRelation(name: TableIdentifier): LogicalPlan = {
91	var rtnRelation = super.lookupRelation(name)	2✔
92	val isRelationRefreshed =	2✔
93	CarbonSessionUtil.refreshRelationAndSetStats(rtnRelation, name)(sparkSession)	2✔
94	if (isRelationRefreshed) {	2✔
95	rtnRelation = super.lookupRelation(name)	×
96	// Reset the stats after lookup.
97	CarbonSessionUtil.refreshRelationAndSetStats(rtnRelation, name)(sparkSession)	×
98	}
99	rtnRelation	2✔
100	}
101
102	override def getCachedPlan(t: QualifiedTableName,
103	c: Callable[LogicalPlan]): LogicalPlan = {
104	val plan = super.getCachedPlan(t, c)	2✔
105	CarbonSessionUtil.updateCachedPlan(plan)	2✔
106	}
107
108	/**
109	* returns hive client from HiveExternalCatalog
110	*
111	* @return
112	*/
113	override def getClient(): org.apache.spark.sql.hive.client.HiveClient = {
114	sparkSession.asInstanceOf[CarbonSession].sharedState.externalCatalog	2✔
115	.asInstanceOf[HiveExternalCatalog].client	2✔
116	}
117
118	override def alterAddColumns(tableIdentifier: TableIdentifier,
119	schemaParts: String,
120	cols: Option[Seq[ColumnSchema]]): Unit = {
121	updateCatalogTableForAlter(tableIdentifier, schemaParts, cols)	2✔
122	}
123
124	override def alterDropColumns(tableIdentifier: TableIdentifier,
125	schemaParts: String,
126	cols: Option[Seq[ColumnSchema]]): Unit = {
127	updateCatalogTableForAlter(tableIdentifier, schemaParts, cols)	2✔
128	}
129
130	override def alterColumnChangeDataTypeOrRename(tableIdentifier: TableIdentifier,
131	schemaParts: String,
132	cols: Option[Seq[ColumnSchema]]): Unit = {
133	updateCatalogTableForAlter(tableIdentifier, schemaParts, cols)	2✔
134	}
135
136	/**
137	* This method alters table to set serde properties and updates the catalog table with new updated
138	* schema for all the alter operations like add column, drop column, change datatype or rename
139	* column
140	* @param tableIdentifier
141	* @param schemaParts
142	* @param cols
143	*/
144	private def updateCatalogTableForAlter(tableIdentifier: TableIdentifier,
145	schemaParts: String,
146	cols: Option[Seq[ColumnSchema]]): Unit = {
147	alterTable(tableIdentifier, schemaParts, cols)	2✔
148	CarbonSessionUtil	2✔
149	.alterExternalCatalogForTableWithUpdatedSchema(tableIdentifier,	2✔
150	cols,	2✔
151	schemaParts,	2✔
152	sparkSession)	2✔
153	}
154
155	override def createPartitions(
156	tableName: TableIdentifier,
157	parts: Seq[CatalogTablePartition],
158	ignoreIfExists: Boolean): Unit = {
159	try {	2✔
160	val table = CarbonEnv.getCarbonTable(tableName)(sparkSession)	2✔
161	val updatedParts = CarbonScalaUtil.updatePartitions(parts, table)	2✔
162	super.createPartitions(tableName, updatedParts, ignoreIfExists)	2✔
163	} catch {
164	case e: Exception =>
165	super.createPartitions(tableName, parts, ignoreIfExists)	1✔
166	}
167	}
168
169	/**
170	* This is alternate way of getting partition information. It first fetches all partitions from
171	* hive and then apply filter instead of querying hive along with filters.
172	* @param partitionFilters
173	* @param sparkSession
174	* @param identifier
175	* @return
176	*/
177	override def getPartitionsAlternate(partitionFilters: Seq[Expression],
178	sparkSession: SparkSession,
179	identifier: TableIdentifier) = {
180	CarbonSessionUtil.prunePartitionsByFilter(partitionFilters, sparkSession, identifier)	1✔
181	}
182
183	/**
184	* Update the storageformat with new location information
185	*/
186	override def updateStorageLocation(
187	path: Path,
188	storage: CatalogStorageFormat,
189	newTableName: String,
190	dbName: String): CatalogStorageFormat = {
191	storage.copy(locationUri = Some(path.toUri))	×
192	}
193	}
194
195	/**
196	* Session state implementation to override sql parser and adding strategies
197	*
198	* @param sparkSession
199	*/
200	class CarbonSessionStateBuilder(sparkSession: SparkSession,	×
201	parentState: Option[SessionState] = None)	×
202	extends HiveSessionStateBuilder(sparkSession, parentState) {	2✔
203
204	override lazy val sqlParser: ParserInterface = new CarbonSparkSqlParser(conf, sparkSession)	2✔
205
206	experimentalMethods.extraStrategies =	2✔
207	Seq(new StreamingTableStrategy(sparkSession),	2✔
208	new CarbonLateDecodeStrategy,	2✔
209	new DDLStrategy(sparkSession)	2✔
210	)
211	experimentalMethods.extraOptimizations = Seq(new CarbonIUDRule,	2✔
212	new CarbonUDFTransformRule,	2✔
213	new CarbonLateDecodeRule)	2✔
214
215	/**
216	* Internal catalog for managing table and database states.
217	*/
218	/**
219	* Create a [[CarbonSessionStateBuilder]].
220	*/
221	override protected lazy val catalog: CarbonHiveSessionCatalog = {	2✔
222	val catalog = new CarbonHiveSessionCatalog(	2✔
223	externalCatalog,	2✔
224	session.sharedState.globalTempViewManager,	2✔
225	functionRegistry,	2✔
226	sparkSession,	2✔
227	conf,	2✔
228	SessionState.newHadoopConf(session.sparkContext.hadoopConfiguration, conf),	2✔
229	sqlParser,	2✔
230	resourceLoader)	2✔
231	parentState.foreach(_.catalog.copyStateTo(catalog))	2✔
232	catalog	2✔
233	}
234
235	private def externalCatalog: HiveExternalCatalog =
236	session.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog]	2✔
237
238	/**
239	* Create a Hive aware resource loader.
240	*/
241	override protected lazy val resourceLoader: HiveSessionResourceLoader = {	2✔
242	val client: HiveClient = externalCatalog.client.newSession()	2✔
243	new HiveSessionResourceLoader(session, client)	2✔
244	}
245
246	override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)	2✔
247
248	override protected def analyzer: Analyzer = {
249	new CarbonAnalyzer(catalog,	2✔
250	conf,	2✔
251	sparkSession,	2✔
252	getAnalyzer(super.analyzer))	2✔
253	}
254
255	/**
256	* This method adds carbon rules to Hive Analyzer and returns new analyzer
257	* @param analyzer hiveSessionStateBuilder analyzer
258	* @return
259	*/
260	def getAnalyzer(analyzer: Analyzer): Analyzer = {
261	new Analyzer(catalog, conf) {	2✔
262
263	override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
264	analyzer.extendedResolutionRules ++	2✔
265	Seq(CarbonIUDAnalysisRule(sparkSession)) ++	2✔
266	Seq(CarbonPreInsertionCasts(sparkSession)) ++ customResolutionRules	2✔
267
268	override val extendedCheckRules: Seq[LogicalPlan => Unit] =	2✔
269	analyzer.extendedCheckRules	2✔
270
271	override val postHocResolutionRules: Seq[Rule[LogicalPlan]] =	2✔
272	analyzer.postHocResolutionRules	2✔
273	}
274	}
275
276	override protected def newBuilder: NewBuilder = new CarbonSessionStateBuilder(_, _)	2✔
277	}

apache / carbondata / 12132

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