apache / carbondata / 916

Build # 916

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/0053dbf7aa082c24f19cc361607a6c916cb8e008">0053dbf7a">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/0053dbf7aa082c24f19cc361607a6c916cb8e008">0053dbf7a</a><a href="https://github.com/apache/carbondata/commit/0053dbf7aa082c24f19cc361607a6c916cb8e008">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/0053dbf7aa082c24f19cc361607a6c916cb8e008&quot;&gt;0053dbf7a&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/0053dbf7aa082c24f19cc361607a6c916cb8e008&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

60629 of 75226 relevant lines covered (80.6%)

1.06 hits per line

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

70.91

/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, 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 RunnableCommand {

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

  override def run(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

    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
    }
  }
}

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, 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(	1✔
41	table: CatalogTable,	1✔
42	mode: SaveMode,	1✔
43	query: LogicalPlan)	1✔
44	extends RunnableCommand {	1✔
45
46	override protected def innerChildren: Seq[LogicalPlan] = Seq(query)	×
47
48	override def run(sparkSession: SparkSession): Seq[Row] = {
49	assert(table.tableType != CatalogTableType.VIEW)	1✔
50	assert(table.provider.isDefined)	1✔
51
52	val sessionState = sparkSession.sessionState	1✔
53	val db = table.identifier.database.getOrElse(sessionState.catalog.getCurrentDatabase)	1✔
54	val tableIdentWithDB = table.identifier.copy(database = Some(db))	1✔
55	val tableName = tableIdentWithDB.unquotedString	1✔
56
57	if (sessionState.catalog.tableExists(tableIdentWithDB)) {	1✔
58	assert(mode != SaveMode.Overwrite,	×
59	s"Expect the table $tableName has been dropped when the save mode is Overwrite")	×
60
61	if (mode == SaveMode.ErrorIfExists) {	×
62	throw new AnalysisException(s"Table $tableName already exists. You need to drop it first.")	×
63	}
64	if (mode == SaveMode.Ignore) {	×
65	// Since the table already exists and the save mode is Ignore, we will just return.
66	return Seq.empty	×
67	}
68
69	saveDataIntoTable(	×
70	sparkSession, table, table.storage.locationUri, query, SaveMode.Append, tableExists = true)	×
71	} else {
72	assert(table.schema.isEmpty)	1✔
73
74	val tableLocation = if (table.tableType == CatalogTableType.MANAGED) {	1✔
75	Some(sessionState.catalog.defaultTablePath(table.identifier))	1✔
76	} else {
77	table.storage.locationUri	×
78	}
79	val result = saveDataIntoTable(	1✔
80	sparkSession, table, tableLocation, query, SaveMode.Overwrite, tableExists = false)	1✔
81
82	result match {	1✔
83	case fs: HadoopFsRelation if table.partitionColumnNames.nonEmpty &&	1✔
84	sparkSession.sqlContext.conf.manageFilesourcePartitions =>	×
85	// Need to recover partitions into the metastore so our saved data is visible.
86	sessionState.executePlan(AlterTableRecoverPartitionsCommand(table.identifier)).toRdd	×
87	case _ =>	1✔
88	}
89	}
90
91	Seq.empty[Row]	1✔
92	}
93
94	private def saveDataIntoTable(
95	session: SparkSession,
96	table: CatalogTable,
97	tableLocation: Option[URI],
98	data: LogicalPlan,
99	mode: SaveMode,
100	tableExists: Boolean): BaseRelation = {	1✔
101	// Create the relation based on the input logical plan: `data`.
102	val pathOption = tableLocation.map("path" -> CatalogUtils.URIToString(_))	1✔
103	val dataSource = DataSource(	1✔
104	session,	1✔
105	className = table.provider.get,	1✔
106	partitionColumns = table.partitionColumnNames,	1✔
107	bucketSpec = table.bucketSpec,	1✔
108	options = table.storage.properties ++ pathOption,	1✔
109	catalogTable = if (tableExists) {	1✔
110	Some(table)	×
111	} else {
112	None	1✔
113	})
114
115	try {	×
116	val physicalPlan = session.sessionState.executePlan(data).executedPlan	1✔
117	CarbonReflectionUtils.invokewriteAndReadMethod(dataSource,	1✔
118	Dataset.ofRows(session, query),	1✔
119	data,	1✔
120	session,	1✔
121	mode,	1✔
122	query,	1✔
123	physicalPlan)	1✔
124	} catch {
125	case ex: AnalysisException =>
126	logError(s"Failed to write to table ${ table.identifier.unquotedString }", ex)	×
127	throw ex	×
128	}
129	}
130	}

apache / carbondata / 916

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