10301

Build # 10301

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/71a6d0cd3e9691dd027cb5cd18659bb36e20f025">71a6d0cd3<a href="https://github.com/apache/carbondata/commit/71a6d0cd3e9691dd027cb5cd18659bb36e20f025">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/e3c451b45b01e659af9b33174a3f9115773592f1">e3c451b45</a>">e3c451b45</a><a href="https://github.com/apache/carbondata/commit/71a6d0cd3e9691dd027cb5cd18659bb36e20f025"> into d5a2c698c">d5a2c698c</a>

Run Details

62 of 62 new or added lines in 7 files covered. (100.0%)

61586 of 77937 relevant lines covered (79.02%)

1.05 hits per line

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

95.24

/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionUtil.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 org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTablePartition, ExternalCatalogUtils}
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
import org.apache.spark.util.CarbonReflectionUtils

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.spark.sql.types.{StructField, StructType}
import org.apache.spark.sql.util.SparkTypeConverter

import scala.collection.mutable.ArrayBuffer

/**
 * This class refresh the relation from cache if the carbontable in
 * carbon catalog is not same as cached carbon relation's carbon table.
 */
object CarbonSessionUtil {

  val LOGGER = LogServiceFactory.getLogService("CarbonSessionUtil")

  /**
   * The method refreshes the cache entry
   *
   * @param rtnRelation [[LogicalPlan]] represents the given table or view.
   * @param name        tableName
   * @param sparkSession
   * @return
   */
  def refreshRelation(rtnRelation: LogicalPlan, name: TableIdentifier)
    (sparkSession: SparkSession): Boolean = {
    var isRelationRefreshed = false
    rtnRelation match {
      case SubqueryAlias(_,
      MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _)
      ) =>
        isRelationRefreshed = CarbonEnv.refreshRelationFromCache(name)(sparkSession)
      case MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _) =>
        isRelationRefreshed = CarbonEnv.refreshRelationFromCache(name)(sparkSession)
      case SubqueryAlias(_, relation) if
      relation.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.CatalogRelation") ||
      relation.getClass.getName
        .equals("org.apache.spark.sql.catalyst.catalog.HiveTableRelation") ||
      relation.getClass.getName.equals(
        "org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation"
      ) =>
        val catalogTable =
          CarbonReflectionUtils.getFieldOfCatalogTable(
            "tableMeta",
            relation
          ).asInstanceOf[CatalogTable]
        isRelationRefreshed =
          CarbonEnv.refreshRelationFromCache(catalogTable.identifier)(sparkSession)
      case _ =>
    }
    isRelationRefreshed
  }

  /**
   * 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
   */
  def prunePartitionsByFilter(partitionFilters: Seq[Expression],
      sparkSession: SparkSession,
      identifier: TableIdentifier): Seq[CatalogTablePartition] = {
    val allPartitions = sparkSession.sessionState.catalog.listPartitions(identifier)
    ExternalCatalogUtils.prunePartitionsByFilter(
      sparkSession.sessionState.catalog.getTableMetadata(identifier),
      allPartitions,
      partitionFilters,
      sparkSession.sessionState.conf.sessionLocalTimeZone
    )
  }

  /**
   * This method alter the table for datatype change or column rename operation, and update the
   * external catalog directly
   *
   * @param tableIdentifier tableIdentifier for table
   * @param cols            all the column of table, which are updated with datatype change of
   *                        new column name
   * @param schemaParts     schemaParts
   * @param sparkSession    sparkSession
   */
  def alterExternalCatalogForTableWithUpdatedSchema(tableIdentifier: TableIdentifier,
      cols: Option[Seq[ColumnSchema]],
      schemaParts: String,
      sparkSession: SparkSession): Unit = {
    val carbonTable = CarbonEnv.getCarbonTable(tableIdentifier)(sparkSession)
    val colArray: scala.collection.mutable.ArrayBuffer[StructField] = ArrayBuffer()
    cols.get.foreach(column =>
      if (!column.isInvisible) {
        colArray += StructField(column.getColumnName,
          SparkTypeConverter
            .convertCarbonToSparkDataType(column,
              carbonTable))
      }
    )
    sparkSession.sessionState.catalog.externalCatalog
      .alterTableDataSchema(tableIdentifier.database.get,
        tableIdentifier.table,
        StructType(colArray))
  }

}

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.hive
19
20	import org.apache.spark.sql.catalyst.TableIdentifier
21	import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTablePartition, ExternalCatalogUtils}
22	import org.apache.spark.sql.catalyst.expressions.Expression
23	import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
24	import org.apache.spark.sql.execution.datasources.LogicalRelation
25	import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
26	import org.apache.spark.util.CarbonReflectionUtils
27
28	import org.apache.carbondata.common.logging.LogServiceFactory
29	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
30	import org.apache.spark.sql.types.{StructField, StructType}
31	import org.apache.spark.sql.util.SparkTypeConverter
32
33	import scala.collection.mutable.ArrayBuffer
34
35	/**
36	* This class refresh the relation from cache if the carbontable in
37	* carbon catalog is not same as cached carbon relation's carbon table.
38	*/
39	object CarbonSessionUtil {
40
41	val LOGGER = LogServiceFactory.getLogService("CarbonSessionUtil")	2✔
42
43	/**
44	* The method refreshes the cache entry
45	*
46	* @param rtnRelation [[LogicalPlan]] represents the given table or view.
47	* @param name tableName
48	* @param sparkSession
49	* @return
50	*/
51	def refreshRelation(rtnRelation: LogicalPlan, name: TableIdentifier)
52	(sparkSession: SparkSession): Boolean = {
53	var isRelationRefreshed = false	2✔
54	rtnRelation match {	2✔
55	case SubqueryAlias(_,	2✔
56	MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _)	2✔
57	) =>
58	isRelationRefreshed = CarbonEnv.refreshRelationFromCache(name)(sparkSession)	×
59	case MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _) =>	2✔
60	isRelationRefreshed = CarbonEnv.refreshRelationFromCache(name)(sparkSession)	×
61	case SubqueryAlias(_, relation) if	2✔
62	relation.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.CatalogRelation") \|\|	2✔
63	relation.getClass.getName	2✔
64	.equals("org.apache.spark.sql.catalyst.catalog.HiveTableRelation") \|\|	2✔
65	relation.getClass.getName.equals(	2✔
66	"org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation"	2✔
67	) =>	2✔
68	val catalogTable =	2✔
69	CarbonReflectionUtils.getFieldOfCatalogTable(	2✔
70	"tableMeta",	2✔
71	relation	2✔
72	).asInstanceOf[CatalogTable]
73	isRelationRefreshed =	2✔
74	CarbonEnv.refreshRelationFromCache(catalogTable.identifier)(sparkSession)	2✔
75	case _ =>	2✔
76	}
77	isRelationRefreshed	2✔
78	}
79
80	/**
81	* This is alternate way of getting partition information. It first fetches all partitions from
82	* hive and then apply filter instead of querying hive along with filters.
83	*
84	* @param partitionFilters
85	* @param sparkSession
86	* @param identifier
87	* @return
88	*/
89	def prunePartitionsByFilter(partitionFilters: Seq[Expression],
90	sparkSession: SparkSession,
91	identifier: TableIdentifier): Seq[CatalogTablePartition] = {
92	val allPartitions = sparkSession.sessionState.catalog.listPartitions(identifier)	1✔
93	ExternalCatalogUtils.prunePartitionsByFilter(	1✔
94	sparkSession.sessionState.catalog.getTableMetadata(identifier),	1✔
95	allPartitions,	1✔
96	partitionFilters,	1✔
97	sparkSession.sessionState.conf.sessionLocalTimeZone	1✔
98	)
99	}
100
101	/**
102	* This method alter the table for datatype change or column rename operation, and update the
103	* external catalog directly
104	*
105	* @param tableIdentifier tableIdentifier for table
106	* @param cols all the column of table, which are updated with datatype change of
107	* new column name
108	* @param schemaParts schemaParts
109	* @param sparkSession sparkSession
110	*/
111	def alterExternalCatalogForTableWithUpdatedSchema(tableIdentifier: TableIdentifier,
112	cols: Option[Seq[ColumnSchema]],
113	schemaParts: String,
114	sparkSession: SparkSession): Unit = {
115	val carbonTable = CarbonEnv.getCarbonTable(tableIdentifier)(sparkSession)	2✔
116	val colArray: scala.collection.mutable.ArrayBuffer[StructField] = ArrayBuffer()	2✔
117	cols.get.foreach(column =>	2✔
118	if (!column.isInvisible) {	2✔
119	colArray += StructField(column.getColumnName,	2✔
120	SparkTypeConverter	2✔
121	.convertCarbonToSparkDataType(column,	2✔
122	carbonTable))	2✔
123	}
124	)
125	sparkSession.sessionState.catalog.externalCatalog	2✔
126	.alterTableDataSchema(tableIdentifier.database.get,	2✔
127	tableIdentifier.table,	2✔
128	StructType(colArray))	2✔
129	}
130
131	}	2✔

apache / carbondata / 10301

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