3611

Build # 3611

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/5332b1e77ec468486504e7f334ba056399521287">5332b1e77<a href="https://github.com/apache/carbondata/commit/5332b1e77ec468486504e7f334ba056399521287">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/0505503e92d98d1b10f8909e4cdf2799d35ee3f6">0505503e9</a>">0505503e9</a><a href="https://github.com/apache/carbondata/commit/5332b1e77ec468486504e7f334ba056399521287"> into 3dda02d44">3dda02d44</a>

Run Details

63757 of 82509 relevant lines covered (77.27%)

1.01 hits per line

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

95.16

/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.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
import org.apache.spark.sql.types.MetadataBuilder
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]
        catalogTable.provider match {
          case Some(provider)
            if provider.equals("org.apache.spark.sql.CarbonSource") ||
               provider.equalsIgnoreCase("carbondata") =>
            // Update stats to none in case of carbon table as we are not expecting any stats from
            // Hive. Hive gives wrong stats for carbon table.
            catalogTable.stats match {
              case Some(stats) =>
                CarbonReflectionUtils.setFieldToCaseClass(catalogTable, "stats", None)
              case _ =>
            }
            isRelationRefreshed =
              CarbonEnv.refreshRelationFromCache(catalogTable.identifier)(sparkSession)
          case _ =>
        }
      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) {
        val structFiled =
          if (null != column.getColumnProperties &&
              column.getColumnProperties.get("comment") != null) {
            StructField(column.getColumnName,
              SparkTypeConverter
                .convertCarbonToSparkDataType(column,
                  carbonTable),
              true,
              // update the column comment if present in the schema
              new MetadataBuilder().putString("comment", column.getColumnProperties.get("comment"))
                .build())
          } else {
            StructField(column.getColumnName,
              SparkTypeConverter
                .convertCarbonToSparkDataType(column,
                  carbonTable))
          }
        colArray += structFiled
      }
    )
    // Alter the data schema of a table identified by the provided database and table name.
    // updated schema should contain all the existing data columns. This alterTableDataSchema API
    // should be called after any alter with existing schema which updates the catalog table with
    // new updated schema
    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.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
25	import org.apache.spark.sql.types.MetadataBuilder
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	catalogTable.provider match {	2✔
74	case Some(provider)	2✔
75	if provider.equals("org.apache.spark.sql.CarbonSource") \|\|	2✔
76	provider.equalsIgnoreCase("carbondata") =>	2✔
77	// Update stats to none in case of carbon table as we are not expecting any stats from
78	// Hive. Hive gives wrong stats for carbon table.
79	catalogTable.stats match {	2✔
80	case Some(stats) =>	2✔
81	CarbonReflectionUtils.setFieldToCaseClass(catalogTable, "stats", None)	×
82	case _ =>	2✔
83	}
84	isRelationRefreshed =	2✔
85	CarbonEnv.refreshRelationFromCache(catalogTable.identifier)(sparkSession)	2✔
86	case _ =>	2✔
87	}
88	case _ =>	2✔
89	}
90	isRelationRefreshed	2✔
91	}
92
93	/**
94	* This is alternate way of getting partition information. It first fetches all partitions from
95	* hive and then apply filter instead of querying hive along with filters.
96	*
97	* @param partitionFilters
98	* @param sparkSession
99	* @param identifier
100	* @return
101	*/
102	def prunePartitionsByFilter(partitionFilters: Seq[Expression],
103	sparkSession: SparkSession,
104	identifier: TableIdentifier): Seq[CatalogTablePartition] = {
105	val allPartitions = sparkSession.sessionState.catalog.listPartitions(identifier)	1✔
106	ExternalCatalogUtils.prunePartitionsByFilter(	1✔
107	sparkSession.sessionState.catalog.getTableMetadata(identifier),	1✔
108	allPartitions,	1✔
109	partitionFilters,	1✔
110	sparkSession.sessionState.conf.sessionLocalTimeZone	1✔
111	)
112	}
113
114	/**
115	* This method alter the table for datatype change or column rename operation, and update the
116	* external catalog directly
117	*
118	* @param tableIdentifier tableIdentifier for table
119	* @param cols all the column of table, which are updated with datatype change of
120	* new column name
121	* @param schemaParts schemaParts
122	* @param sparkSession sparkSession
123	*/
124	def alterExternalCatalogForTableWithUpdatedSchema(tableIdentifier: TableIdentifier,
125	cols: Option[Seq[ColumnSchema]],
126	schemaParts: String,
127	sparkSession: SparkSession): Unit = {
128	val carbonTable = CarbonEnv.getCarbonTable(tableIdentifier)(sparkSession)	2✔
129	val colArray: scala.collection.mutable.ArrayBuffer[StructField] = ArrayBuffer()	2✔
130	cols.get.foreach(column =>	2✔
131	if (!column.isInvisible) {	2✔
132	val structFiled =	2✔
133	if (null != column.getColumnProperties &&	2✔
134	column.getColumnProperties.get("comment") != null) {	2✔
135	StructField(column.getColumnName,	2✔
136	SparkTypeConverter	2✔
137	.convertCarbonToSparkDataType(column,	2✔
138	carbonTable),	2✔
139	true,	2✔
140	// update the column comment if present in the schema
141	new MetadataBuilder().putString("comment", column.getColumnProperties.get("comment"))	2✔
142	.build())	2✔
143	} else {
144	StructField(column.getColumnName,	2✔
145	SparkTypeConverter	2✔
146	.convertCarbonToSparkDataType(column,	2✔
147	carbonTable))	2✔
148	}
149	colArray += structFiled	2✔
150	}
151	)
152	// Alter the data schema of a table identified by the provided database and table name.
153	// updated schema should contain all the existing data columns. This alterTableDataSchema API
154	// should be called after any alter with existing schema which updates the catalog table with
155	// new updated schema
156	sparkSession.sessionState.catalog.externalCatalog	2✔
157	.alterTableDataSchema(tableIdentifier.database.get,	2✔
158	tableIdentifier.table,	2✔
159	StructType(colArray))	2✔
160	}
161
162	}	2✔

apache / carbondata / 3611

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