1784

Build # 1784

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/b2540d63e6c9a549f36b8f034dd73e889e9cc961">b2540d63e<a href="https://github.com/apache/carbondata/commit/b2540d63e6c9a549f36b8f034dd73e889e9cc961">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/09a469e926d43b4992084e5dc1727899ee04366f">09a469e92</a>">09a469e92</a><a href="https://github.com/apache/carbondata/commit/b2540d63e6c9a549f36b8f034dd73e889e9cc961"> into 1bbae2657">1bbae2657</a>

Run Details

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

61591 of 77478 relevant lines covered (79.49%)

1.05 hits per line

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

70.69

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

import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, CatalogUtils}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.command.{AlterTableRecoverPartitionsCommand, AtomicRunnableCommand, RunnableCommand}
import org.apache.spark.sql.execution.datasources.{DataSource, HadoopFsRelation}
import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.{AnalysisException, Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.util.CarbonReflectionUtils

/**
 * Create table 'using carbondata' and insert the query result into it.
 *
 * @param table the Catalog Table
 * @param mode  SaveMode:Ignore,OverWrite,ErrorIfExists,Append
 * @param query the query whose result will be insert into the new relation
 *
 */

case class CreateCarbonSourceTableAsSelectCommand(
    table: CatalogTable,
    mode: SaveMode,
    query: LogicalPlan)
  extends AtomicRunnableCommand {

  override protected def innerChildren: Seq[LogicalPlan] = Seq(query)

  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
    Seq.empty
  }

  override def processData(sparkSession: SparkSession): Seq[Row] ={
    assert(table.tableType != CatalogTableType.VIEW)
    assert(table.provider.isDefined)

    val sessionState = sparkSession.sessionState
    val db = table.identifier.database.getOrElse(sessionState.catalog.getCurrentDatabase)
    val tableIdentWithDB = table.identifier.copy(database = Some(db))
    val tableName = tableIdentWithDB.unquotedString
    setAuditTable(db, table.identifier.table)

    if (sessionState.catalog.tableExists(tableIdentWithDB)) {
      assert(mode != SaveMode.Overwrite,
        s"Expect the table $tableName has been dropped when the save mode is Overwrite")

      if (mode == SaveMode.ErrorIfExists) {
        throw new AnalysisException(s"Table $tableName already exists. You need to drop it first.")
      }
      if (mode == SaveMode.Ignore) {
        // Since the table already exists and the save mode is Ignore, we will just return.
        return Seq.empty
      }

      saveDataIntoTable(
        sparkSession, table, table.storage.locationUri, query, SaveMode.Append, tableExists = true)
    } else {
      assert(table.schema.isEmpty)

      val tableLocation = if (table.tableType == CatalogTableType.MANAGED) {
        Some(sessionState.catalog.defaultTablePath(table.identifier))
      } else {
        table.storage.locationUri
      }
      val result = saveDataIntoTable(
        sparkSession, table, tableLocation, query, SaveMode.Overwrite, tableExists = false)

      result match {
        case fs: HadoopFsRelation if table.partitionColumnNames.nonEmpty &&
                                     sparkSession.sqlContext.conf.manageFilesourcePartitions =>
          // Need to recover partitions into the metastore so our saved data is visible.
          sessionState.executePlan(AlterTableRecoverPartitionsCommand(table.identifier)).toRdd
        case _ =>
      }
    }

    Seq.empty[Row]
  }

  private def saveDataIntoTable(
      session: SparkSession,
      table: CatalogTable,
      tableLocation: Option[URI],
      data: LogicalPlan,
      mode: SaveMode,
      tableExists: Boolean): BaseRelation = {
    // Create the relation based on the input logical plan: `data`.
    val pathOption = tableLocation.map("path" -> CatalogUtils.URIToString(_))
    val dataSource = DataSource(
      session,
      className = table.provider.get,
      partitionColumns = table.partitionColumnNames,
      bucketSpec = table.bucketSpec,
      options = table.storage.properties ++ pathOption,
      catalogTable = if (tableExists) {
        Some(table)
      } else {
        None
      })

    try {
      val physicalPlan = session.sessionState.executePlan(data).executedPlan
      CarbonReflectionUtils.invokewriteAndReadMethod(dataSource,
        Dataset.ofRows(session, query),
        data,
        session,
        mode,
        query,
        physicalPlan)
    } catch {
      case ex: AnalysisException =>
        logError(s"Failed to write to table ${ table.identifier.unquotedString }", ex)
        throw ex
    }
  }

  override protected def opName: String = "CREATE TABLE AS SELECT"
}

1
2	/*
3	* Licensed to the Apache Software Foundation (ASF) under one or more
4	* contributor license agreements. See the NOTICE file distributed with
5	* this work for additional information regarding copyright ownership.
6	* The ASF licenses this file to You under the Apache License, Version 2.0
7	* (the "License"); you may not use this file except in compliance with
8	* the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing, software
13	* distributed under the License is distributed on an "AS IS" BASIS,
14	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	* See the License for the specific language governing permissions and
16	* limitations under the License.
17	*/
18
19	package org.apache.spark.sql.hive
20
21	import java.net.URI
22
23	import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, CatalogUtils}
24	import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
25	import org.apache.spark.sql.execution.command.{AlterTableRecoverPartitionsCommand, AtomicRunnableCommand, RunnableCommand}
26	import org.apache.spark.sql.execution.datasources.{DataSource, HadoopFsRelation}
27	import org.apache.spark.sql.sources.BaseRelation
28	import org.apache.spark.sql.{AnalysisException, Dataset, Row, SaveMode, SparkSession}
29	import org.apache.spark.util.CarbonReflectionUtils
30
31	/**
32	* Create table 'using carbondata' and insert the query result into it.
33	*
34	* @param table the Catalog Table
35	* @param mode SaveMode:Ignore,OverWrite,ErrorIfExists,Append
36	* @param query the query whose result will be insert into the new relation
37	*
38	*/
39
40	case class CreateCarbonSourceTableAsSelectCommand(	×
41	table: CatalogTable,	1✔
42	mode: SaveMode,	1✔
43	query: LogicalPlan)	1✔
44	extends AtomicRunnableCommand {	1✔
45
46	override protected def innerChildren: Seq[LogicalPlan] = Seq(query)	×
47
48	override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
49	Seq.empty	1✔
50	}
51
52	override def processData(sparkSession: SparkSession): Seq[Row] ={
53	assert(table.tableType != CatalogTableType.VIEW)	1✔
54	assert(table.provider.isDefined)	1✔
55
56	val sessionState = sparkSession.sessionState	1✔
57	val db = table.identifier.database.getOrElse(sessionState.catalog.getCurrentDatabase)	1✔
58	val tableIdentWithDB = table.identifier.copy(database = Some(db))	1✔
59	val tableName = tableIdentWithDB.unquotedString	1✔
60	setAuditTable(db, table.identifier.table)	1✔
61
62	if (sessionState.catalog.tableExists(tableIdentWithDB)) {	1✔
63	assert(mode != SaveMode.Overwrite,	×
64	s"Expect the table $tableName has been dropped when the save mode is Overwrite")	×
65
66	if (mode == SaveMode.ErrorIfExists) {	×
67	throw new AnalysisException(s"Table $tableName already exists. You need to drop it first.")	×
68	}
69	if (mode == SaveMode.Ignore) {	×
70	// Since the table already exists and the save mode is Ignore, we will just return.
71	return Seq.empty	×
72	}
73
74	saveDataIntoTable(	×
75	sparkSession, table, table.storage.locationUri, query, SaveMode.Append, tableExists = true)	×
76	} else {
77	assert(table.schema.isEmpty)	1✔
78
79	val tableLocation = if (table.tableType == CatalogTableType.MANAGED) {	1✔
80	Some(sessionState.catalog.defaultTablePath(table.identifier))	1✔
81	} else {
82	table.storage.locationUri	×
83	}
84	val result = saveDataIntoTable(	1✔
85	sparkSession, table, tableLocation, query, SaveMode.Overwrite, tableExists = false)	1✔
86
87	result match {	1✔
88	case fs: HadoopFsRelation if table.partitionColumnNames.nonEmpty &&	1✔
89	sparkSession.sqlContext.conf.manageFilesourcePartitions =>	×
90	// Need to recover partitions into the metastore so our saved data is visible.
91	sessionState.executePlan(AlterTableRecoverPartitionsCommand(table.identifier)).toRdd	×
92	case _ =>	1✔
93	}
94	}
95
96	Seq.empty[Row]	1✔
97	}
98
99	private def saveDataIntoTable(
100	session: SparkSession,
101	table: CatalogTable,
102	tableLocation: Option[URI],
103	data: LogicalPlan,
104	mode: SaveMode,
105	tableExists: Boolean): BaseRelation = {	1✔
106	// Create the relation based on the input logical plan: `data`.
107	val pathOption = tableLocation.map("path" -> CatalogUtils.URIToString(_))	1✔
108	val dataSource = DataSource(	1✔
109	session,	1✔
110	className = table.provider.get,	1✔
111	partitionColumns = table.partitionColumnNames,	1✔
112	bucketSpec = table.bucketSpec,	1✔
113	options = table.storage.properties ++ pathOption,	1✔
114	catalogTable = if (tableExists) {	1✔
115	Some(table)	×
116	} else {
117	None	1✔
118	})
119
120	try {	×
121	val physicalPlan = session.sessionState.executePlan(data).executedPlan	1✔
122	CarbonReflectionUtils.invokewriteAndReadMethod(dataSource,	1✔
123	Dataset.ofRows(session, query),	1✔
124	data,	1✔
125	session,	1✔
126	mode,	1✔
127	query,	1✔
128	physicalPlan)	1✔
129	} catch {
130	case ex: AnalysisException =>
131	logError(s"Failed to write to table ${ table.identifier.unquotedString }", ex)	×
132	throw ex	×
133	}
134	}
135
136	override protected def opName: String = "CREATE TABLE AS SELECT"	1✔
137	}

apache / carbondata / 1784

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