1395

Build # 1395

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/1b0c498e32ec80a1aaf688a558978a951938b9e8">1b0c498e3<a href="https://github.com/apache/carbondata/commit/1b0c498e32ec80a1aaf688a558978a951938b9e8">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/e9ec6f515175d459bf0190807e52e3ec03b982c6">e9ec6f515</a>">e9ec6f515</a><a href="https://github.com/apache/carbondata/commit/1b0c498e32ec80a1aaf688a558978a951938b9e8"> into 0e39abf81">0e39abf81</a>

Run Details

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

61763 of 77150 relevant lines covered (80.06%)

1.06 hits per line

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

0.0

/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonHiveMetaStore.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 scala.collection.JavaConverters._

import org.apache.hadoop.fs.Path
import org.apache.spark.sql.{CarbonEnv, CarbonSession, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier

import org.apache.carbondata.core.cache.dictionary.ManageDictionaryAndBTree
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder
import org.apache.carbondata.core.metadata.{schema, AbsoluteTableIdentifier, CarbonMetadata, CarbonTableIdentifier}
import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.format
import org.apache.carbondata.format.SchemaEvolutionEntry
import org.apache.carbondata.spark.util.CarbonSparkUtil

/**
 * Metastore to store carbonschema in hive
 */
class CarbonHiveMetaStore extends CarbonFileMetastore {

  override def isReadFromHiveMetaStore: Boolean = true

  /**
   * Create spark session from paramters.
   *
   * @param parameters
   * @param absIdentifier
   * @param sparkSession
   */
  override def createCarbonRelation(parameters: Map[String, String],
      absIdentifier: AbsoluteTableIdentifier,
      sparkSession: SparkSession): CarbonRelation = {
    val info = CarbonUtil.convertGsonToTableInfo(parameters.asJava)
    val carbonRelation = if (info != null) {
      val table = CarbonTable.buildFromTableInfo(info)
      CarbonRelation(info.getDatabaseName, info.getFactTable.getTableName,
        CarbonSparkUtil.createSparkMeta(table), table)
    } else {
      super.createCarbonRelation(parameters, absIdentifier, sparkSession)
    }
    carbonRelation.refresh()
    carbonRelation
  }


  override def isTablePathExists(tableIdentifier: TableIdentifier)
    (sparkSession: SparkSession): Boolean = {
    tableExists(tableIdentifier)(sparkSession)
  }

  override def dropTable(absoluteTableIdentifier: AbsoluteTableIdentifier)
    (sparkSession: SparkSession): Unit = {
    val dbName = absoluteTableIdentifier.getCarbonTableIdentifier.getDatabaseName
    val tableName = absoluteTableIdentifier.getCarbonTableIdentifier.getTableName
    val carbonTable = CarbonMetadata.getInstance.getCarbonTable(dbName, tableName)
    if (null != carbonTable) {
      // clear driver B-tree and dictionary cache
      ManageDictionaryAndBTree.clearBTreeAndDictionaryLRUCache(carbonTable)
    }
    checkSchemasModifiedTimeAndReloadTable(TableIdentifier(tableName, Some(dbName)))
    CarbonHiveMetadataUtil.invalidateAndDropTable(dbName, tableName, sparkSession)
    // discard cached table info in cachedDataSourceTables
    val tableIdentifier = TableIdentifier(tableName, Option(dbName))
    sparkSession.sessionState.catalog.refreshTable(tableIdentifier)
    DataMapStoreManager.getInstance().clearDataMaps(absoluteTableIdentifier)
    SegmentPropertiesAndSchemaHolder.getInstance().invalidate(absoluteTableIdentifier)
    removeTableFromMetadata(dbName, tableName)
  }

  override def checkSchemasModifiedTimeAndReloadTable(tableIdentifier: TableIdentifier): Boolean = {
    // do nothing
    false
  }

  override def listAllTables(sparkSession: SparkSession): Seq[CarbonTable] = {
    // Todo
    Seq()
  }

  override def getThriftTableInfo(carbonTable: CarbonTable): format.TableInfo = {
    val schemaConverter = new ThriftWrapperSchemaConverterImpl
    schemaConverter.fromWrapperToExternalTableInfo(carbonTable.getTableInfo,
      carbonTable.getDatabaseName,
      carbonTable.getTableName)
  }


  /**
   * This method will overwrite the existing schema and update it with the given details
   *
   * @param newTableIdentifier
   * @param thriftTableInfo
   * @param schemaEvolutionEntry
   * @param sparkSession
   */
  override def updateTableSchemaForAlter(newTableIdentifier: CarbonTableIdentifier,
      oldTableIdentifier: CarbonTableIdentifier,
      thriftTableInfo: format.TableInfo,
      schemaEvolutionEntry: SchemaEvolutionEntry,
      tablePath: String)
    (sparkSession: SparkSession): String = {
    val schemaConverter = new ThriftWrapperSchemaConverterImpl
    if (schemaEvolutionEntry != null) {
      thriftTableInfo.fact_table.schema_evolution.schema_evolution_history.add(schemaEvolutionEntry)
    }
    updateHiveMetaStoreForAlter(newTableIdentifier,
      oldTableIdentifier,
      thriftTableInfo,
      tablePath,
      sparkSession,
      schemaConverter)
  }

  /**
   * This method will overwrite the existing schema and update it with the given details
   *
   * @param newTableIdentifier
   * @param thriftTableInfo
   * @param carbonTablePath
   * @param sparkSession
   */
  override def updateTableSchemaForDataMap(newTableIdentifier: CarbonTableIdentifier,
      oldTableIdentifier: CarbonTableIdentifier,
      thriftTableInfo: org.apache.carbondata.format.TableInfo,
      carbonTablePath: String)(sparkSession: SparkSession): String = {
    val schemaConverter = new ThriftWrapperSchemaConverterImpl
    updateHiveMetaStoreForAlter(
      newTableIdentifier,
      oldTableIdentifier,
      thriftTableInfo,
      carbonTablePath,
      sparkSession,
      schemaConverter)
  }

  private def updateHiveMetaStoreForAlter(newTableIdentifier: CarbonTableIdentifier,
      oldTableIdentifier: CarbonTableIdentifier,
      thriftTableInfo: format.TableInfo,
      tablePath: String,
      sparkSession: SparkSession,
      schemaConverter: ThriftWrapperSchemaConverterImpl) = {
    val wrapperTableInfo = schemaConverter.fromExternalToWrapperTableInfo(
      thriftTableInfo,
      newTableIdentifier.getDatabaseName,
      newTableIdentifier.getTableName,
      tablePath)
    val dbName = newTableIdentifier.getDatabaseName
    val tableName = newTableIdentifier.getTableName
    val schemaParts = CarbonUtil.convertToMultiGsonStrings(wrapperTableInfo, "=", "'", "")
    val hiveClient = sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog]
      .getClient()
    hiveClient.runSqlHive(s"ALTER TABLE $dbName.$tableName SET SERDEPROPERTIES($schemaParts)")

    sparkSession.catalog.refreshTable(TableIdentifier(tableName, Some(dbName)).quotedString)
    removeTableFromMetadata(dbName, tableName)
    CarbonMetadata.getInstance().loadTableMetadata(wrapperTableInfo)
    tablePath
  }

  /**
   * Generates schema string from TableInfo
   */
  override def generateTableSchemaString(
      tableInfo: schema.table.TableInfo,
      absoluteTableIdentifier: AbsoluteTableIdentifier): String = {
    val schemaEvolutionEntry = new schema.SchemaEvolutionEntry
    schemaEvolutionEntry.setTimeStamp(tableInfo.getLastUpdatedTime)
    tableInfo.getFactTable.getSchemaEvolution.getSchemaEvolutionEntryList.add(schemaEvolutionEntry)
    CarbonUtil.convertToMultiGsonStrings(tableInfo, " ", "", ",")
  }

  /**
   * This method will is used to remove the evolution entry in case of failure.
   *
   * @param carbonTableIdentifier
   * @param thriftTableInfo
   * @param sparkSession
   */
  override def revertTableSchemaInAlterFailure(carbonTableIdentifier: CarbonTableIdentifier,
      thriftTableInfo: format.TableInfo,
      identifier: AbsoluteTableIdentifier)
    (sparkSession: SparkSession): String = {
    val schemaConverter = new ThriftWrapperSchemaConverterImpl
    val evolutionEntries = thriftTableInfo.fact_table.schema_evolution.schema_evolution_history
    evolutionEntries.remove(evolutionEntries.size() - 1)
    updateHiveMetaStoreForAlter(carbonTableIdentifier,
      carbonTableIdentifier,
      thriftTableInfo,
      identifier.getTablePath,
      sparkSession,
      schemaConverter)
  }

  override def revertTableSchemaForPreAggCreationFailure(absoluteTableIdentifier:
  AbsoluteTableIdentifier,
      thriftTableInfo: org.apache.carbondata.format.TableInfo)
    (sparkSession: SparkSession): String = {
    val schemaConverter = new ThriftWrapperSchemaConverterImpl
    val childSchemas = thriftTableInfo.dataMapSchemas
    childSchemas.remove(childSchemas.size() - 1)
    val carbonTableIdentifier = absoluteTableIdentifier.getCarbonTableIdentifier
    updateHiveMetaStoreForAlter(carbonTableIdentifier,
      carbonTableIdentifier,
      thriftTableInfo,
      absoluteTableIdentifier.getTablePath,
      sparkSession,
      schemaConverter)
  }
}

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 scala.collection.JavaConverters._
20
21	import org.apache.hadoop.fs.Path
22	import org.apache.spark.sql.{CarbonEnv, CarbonSession, SparkSession}
23	import org.apache.spark.sql.catalyst.TableIdentifier
24
25	import org.apache.carbondata.core.cache.dictionary.ManageDictionaryAndBTree
26	import org.apache.carbondata.core.datamap.DataMapStoreManager
27	import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder
28	import org.apache.carbondata.core.metadata.{schema, AbsoluteTableIdentifier, CarbonMetadata, CarbonTableIdentifier}
29	import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
30	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
31	import org.apache.carbondata.core.util.CarbonUtil
32	import org.apache.carbondata.core.util.path.CarbonTablePath
33	import org.apache.carbondata.format
34	import org.apache.carbondata.format.SchemaEvolutionEntry
35	import org.apache.carbondata.spark.util.CarbonSparkUtil
36
37	/**
38	* Metastore to store carbonschema in hive
39	*/
40	class CarbonHiveMetaStore extends CarbonFileMetastore {	×
41
42	override def isReadFromHiveMetaStore: Boolean = true	×
43
44	/**
45	* Create spark session from paramters.
46	*
47	* @param parameters
48	* @param absIdentifier
49	* @param sparkSession
50	*/
51	override def createCarbonRelation(parameters: Map[String, String],
52	absIdentifier: AbsoluteTableIdentifier,
53	sparkSession: SparkSession): CarbonRelation = {
54	val info = CarbonUtil.convertGsonToTableInfo(parameters.asJava)	×
55	val carbonRelation = if (info != null) {	×
56	val table = CarbonTable.buildFromTableInfo(info)	×
57	CarbonRelation(info.getDatabaseName, info.getFactTable.getTableName,	×
58	CarbonSparkUtil.createSparkMeta(table), table)	×
59	} else {
60	super.createCarbonRelation(parameters, absIdentifier, sparkSession)	×
61	}
62	carbonRelation.refresh()	×
63	carbonRelation	×
64	}
65
66
67	override def isTablePathExists(tableIdentifier: TableIdentifier)
68	(sparkSession: SparkSession): Boolean = {
69	tableExists(tableIdentifier)(sparkSession)	×
70	}
71
72	override def dropTable(absoluteTableIdentifier: AbsoluteTableIdentifier)
73	(sparkSession: SparkSession): Unit = {
74	val dbName = absoluteTableIdentifier.getCarbonTableIdentifier.getDatabaseName	×
75	val tableName = absoluteTableIdentifier.getCarbonTableIdentifier.getTableName	×
76	val carbonTable = CarbonMetadata.getInstance.getCarbonTable(dbName, tableName)	×
77	if (null != carbonTable) {	×
78	// clear driver B-tree and dictionary cache
79	ManageDictionaryAndBTree.clearBTreeAndDictionaryLRUCache(carbonTable)	×
80	}
81	checkSchemasModifiedTimeAndReloadTable(TableIdentifier(tableName, Some(dbName)))	×
82	CarbonHiveMetadataUtil.invalidateAndDropTable(dbName, tableName, sparkSession)	×
83	// discard cached table info in cachedDataSourceTables
84	val tableIdentifier = TableIdentifier(tableName, Option(dbName))	×
85	sparkSession.sessionState.catalog.refreshTable(tableIdentifier)	×
86	DataMapStoreManager.getInstance().clearDataMaps(absoluteTableIdentifier)	×
87	SegmentPropertiesAndSchemaHolder.getInstance().invalidate(absoluteTableIdentifier)	×
88	removeTableFromMetadata(dbName, tableName)	×
89	}
90
91	override def checkSchemasModifiedTimeAndReloadTable(tableIdentifier: TableIdentifier): Boolean = {
92	// do nothing
93	false	×
94	}
95
96	override def listAllTables(sparkSession: SparkSession): Seq[CarbonTable] = {
97	// Todo
98	Seq()	×
99	}
100
101	override def getThriftTableInfo(carbonTable: CarbonTable): format.TableInfo = {
102	val schemaConverter = new ThriftWrapperSchemaConverterImpl	×
103	schemaConverter.fromWrapperToExternalTableInfo(carbonTable.getTableInfo,	×
104	carbonTable.getDatabaseName,	×
105	carbonTable.getTableName)	×
106	}
107
108
109	/**
110	* This method will overwrite the existing schema and update it with the given details
111	*
112	* @param newTableIdentifier
113	* @param thriftTableInfo
114	* @param schemaEvolutionEntry
115	* @param sparkSession
116	*/
117	override def updateTableSchemaForAlter(newTableIdentifier: CarbonTableIdentifier,
118	oldTableIdentifier: CarbonTableIdentifier,
119	thriftTableInfo: format.TableInfo,
120	schemaEvolutionEntry: SchemaEvolutionEntry,
121	tablePath: String)
122	(sparkSession: SparkSession): String = {
123	val schemaConverter = new ThriftWrapperSchemaConverterImpl	×
124	if (schemaEvolutionEntry != null) {	×
125	thriftTableInfo.fact_table.schema_evolution.schema_evolution_history.add(schemaEvolutionEntry)	×
126	}
127	updateHiveMetaStoreForAlter(newTableIdentifier,	×
128	oldTableIdentifier,	×
129	thriftTableInfo,	×
130	tablePath,	×
131	sparkSession,	×
132	schemaConverter)	×
133	}
134
135	/**
136	* This method will overwrite the existing schema and update it with the given details
137	*
138	* @param newTableIdentifier
139	* @param thriftTableInfo
140	* @param carbonTablePath
141	* @param sparkSession
142	*/
143	override def updateTableSchemaForDataMap(newTableIdentifier: CarbonTableIdentifier,
144	oldTableIdentifier: CarbonTableIdentifier,
145	thriftTableInfo: org.apache.carbondata.format.TableInfo,
146	carbonTablePath: String)(sparkSession: SparkSession): String = {
147	val schemaConverter = new ThriftWrapperSchemaConverterImpl	×
148	updateHiveMetaStoreForAlter(	×
149	newTableIdentifier,	×
150	oldTableIdentifier,	×
151	thriftTableInfo,	×
152	carbonTablePath,	×
153	sparkSession,	×
154	schemaConverter)	×
155	}
156
157	private def updateHiveMetaStoreForAlter(newTableIdentifier: CarbonTableIdentifier,
158	oldTableIdentifier: CarbonTableIdentifier,
159	thriftTableInfo: format.TableInfo,
160	tablePath: String,
161	sparkSession: SparkSession,
162	schemaConverter: ThriftWrapperSchemaConverterImpl) = {
163	val wrapperTableInfo = schemaConverter.fromExternalToWrapperTableInfo(	×
164	thriftTableInfo,	×
165	newTableIdentifier.getDatabaseName,	×
166	newTableIdentifier.getTableName,	×
167	tablePath)	×
168	val dbName = newTableIdentifier.getDatabaseName	×
169	val tableName = newTableIdentifier.getTableName	×
170	val schemaParts = CarbonUtil.convertToMultiGsonStrings(wrapperTableInfo, "=", "'", "")	×
171	val hiveClient = sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog]	×
172	.getClient()	×
173	hiveClient.runSqlHive(s"ALTER TABLE $dbName.$tableName SET SERDEPROPERTIES($schemaParts)")	×
174
175	sparkSession.catalog.refreshTable(TableIdentifier(tableName, Some(dbName)).quotedString)	×
176	removeTableFromMetadata(dbName, tableName)	×
177	CarbonMetadata.getInstance().loadTableMetadata(wrapperTableInfo)	×
178	tablePath	×
179	}
180
181	/**
182	* Generates schema string from TableInfo
183	*/
184	override def generateTableSchemaString(
185	tableInfo: schema.table.TableInfo,
186	absoluteTableIdentifier: AbsoluteTableIdentifier): String = {
187	val schemaEvolutionEntry = new schema.SchemaEvolutionEntry	×
188	schemaEvolutionEntry.setTimeStamp(tableInfo.getLastUpdatedTime)	×
189	tableInfo.getFactTable.getSchemaEvolution.getSchemaEvolutionEntryList.add(schemaEvolutionEntry)	×
190	CarbonUtil.convertToMultiGsonStrings(tableInfo, " ", "", ",")	×
191	}
192
193	/**
194	* This method will is used to remove the evolution entry in case of failure.
195	*
196	* @param carbonTableIdentifier
197	* @param thriftTableInfo
198	* @param sparkSession
199	*/
200	override def revertTableSchemaInAlterFailure(carbonTableIdentifier: CarbonTableIdentifier,
201	thriftTableInfo: format.TableInfo,
202	identifier: AbsoluteTableIdentifier)
203	(sparkSession: SparkSession): String = {
204	val schemaConverter = new ThriftWrapperSchemaConverterImpl	×
205	val evolutionEntries = thriftTableInfo.fact_table.schema_evolution.schema_evolution_history	×
206	evolutionEntries.remove(evolutionEntries.size() - 1)	×
207	updateHiveMetaStoreForAlter(carbonTableIdentifier,	×
208	carbonTableIdentifier,	×
209	thriftTableInfo,	×
210	identifier.getTablePath,	×
211	sparkSession,	×
212	schemaConverter)	×
213	}
214
215	override def revertTableSchemaForPreAggCreationFailure(absoluteTableIdentifier:
216	AbsoluteTableIdentifier,
217	thriftTableInfo: org.apache.carbondata.format.TableInfo)
218	(sparkSession: SparkSession): String = {
219	val schemaConverter = new ThriftWrapperSchemaConverterImpl	×
220	val childSchemas = thriftTableInfo.dataMapSchemas	×
221	childSchemas.remove(childSchemas.size() - 1)	×
222	val carbonTableIdentifier = absoluteTableIdentifier.getCarbonTableIdentifier	×
223	updateHiveMetaStoreForAlter(carbonTableIdentifier,	×
224	carbonTableIdentifier,	×
225	thriftTableInfo,	×
226	absoluteTableIdentifier.getTablePath,	×
227	sparkSession,	×
228	schemaConverter)	×
229	}
230	}

apache / carbondata / 1395

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