apache / carbondata / 214

Build # 214

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ef364e748999a181cf4ef7d6cdf7d8ffe006725d">ef364e748<a href="https://github.com/apache/carbondata/commit/ef364e748999a181cf4ef7d6cdf7d8ffe006725d">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/f372e70f9d0caa0418ee684f71e4e8c92a90a810">f372e70f9</a>">f372e70f9</a><a href="https://github.com/apache/carbondata/commit/ef364e748999a181cf4ef7d6cdf7d8ffe006725d"> into 526e3bfa1">526e3bfa1</a>

Run Details

664 of 664 new or added lines in 33 files covered. (100.0%)

57869 of 71864 relevant lines covered (80.53%)

1.07 hits per line

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

91.3

/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.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

import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CarbonException

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.metadata.datatype.{DataTypes => CarbonType}
import org.apache.carbondata.spark.CarbonOption

class CarbonDataFrameWriter(sqlContext: SQLContext, val dataFrame: DataFrame) {

  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  def saveAsCarbonFile(parameters: Map[String, String] = Map()): Unit = {
    // create a new table using dataframe's schema and write its content into the table
    sqlContext.sparkSession.sql(
      makeCreateTableString(dataFrame.schema, new CarbonOption(parameters)))
    writeToCarbonFile(parameters)
  }

  def appendToCarbonFile(parameters: Map[String, String] = Map()): Unit = {
    writeToCarbonFile(parameters)
  }

  private def writeToCarbonFile(parameters: Map[String, String] = Map()): Unit = {
    val options = new CarbonOption(parameters)
    loadDataFrame(options)
  }
  /**
   * Loading DataFrame directly without saving DataFrame to CSV files.
   * @param options
   */
  private def loadDataFrame(options: CarbonOption): Unit = {
    val header = dataFrame.columns.mkString(",")
    CarbonLoadDataCommand(
      Some(CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)),
      options.tableName,
      null,
      Seq(),
      Map("fileheader" -> header) ++ options.toMap,
      isOverwriteTable = options.overwriteEnabled,
      null,
      Some(dataFrame)).run(sqlContext.sparkSession)
  }

  private def convertToCarbonType(sparkType: DataType): String = {
    sparkType match {
      case StringType => CarbonType.STRING.getName
      case IntegerType => CarbonType.INT.getName
      case ShortType => CarbonType.SHORT.getName
      case LongType => CarbonType.LONG.getName
      case FloatType => CarbonType.DOUBLE.getName
      case DoubleType => CarbonType.DOUBLE.getName
      case TimestampType => CarbonType.TIMESTAMP.getName
      case DateType => CarbonType.DATE.getName
      case decimal: DecimalType => s"decimal(${decimal.precision}, ${decimal.scale})"
      case BooleanType => CarbonType.BOOLEAN.getName
      case other => CarbonException.analysisException(s"unsupported type: $other")
    }
  }

  private def makeCreateTableString(schema: StructType, options: CarbonOption): String = {
    val property = Map(
      "SORT_COLUMNS" -> options.sortColumns,
      "DICTIONARY_INCLUDE" -> options.dictionaryInclude,
      "DICTIONARY_EXCLUDE" -> options.dictionaryExclude,
      "LONG_STRING_COLUMNS" -> options.longStringColumns,
      "TABLE_BLOCKSIZE" -> options.tableBlockSize,
      "TABLE_BLOCKLET_SIZE" -> options.tableBlockletSize,
      "STREAMING" -> Option(options.isStreaming.toString)
    ).filter(_._2.isDefined)
      .map(property => s"'${property._1}' = '${property._2.get}'").mkString(",")

    val partition: Seq[String] = if (options.partitionColumns.isDefined) {
      if (options.partitionColumns.get.toSet.size != options.partitionColumns.get.length) {
        throw new MalformedCarbonCommandException(s"repeated partition column")
      }
      options.partitionColumns.get.map { column =>
        val field = schema.fields.find(_.name.equalsIgnoreCase(column))
        if (field.isEmpty) {
          throw new MalformedCarbonCommandException(s"invalid partition column: $column")
        }
        s"$column ${field.get.dataType.typeName}"
      }
    } else {
      Seq()
    }

    val schemaWithoutPartition = if (options.partitionColumns.isDefined) {
      val partitionCols = options.partitionColumns.get
      val fields = schema.filterNot {
        field => partitionCols.exists(_.equalsIgnoreCase(field.name))
      }
      StructType(fields)
    } else {
      schema
    }

    val carbonSchema = schemaWithoutPartition.map { field =>
      s"${ field.name } ${ convertToCarbonType(field.dataType) }"
    }

    val dbName = CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)

    s"""
       | CREATE TABLE IF NOT EXISTS $dbName.${options.tableName}
       | (${ carbonSchema.mkString(", ") })
       | ${ if (partition.nonEmpty) s"PARTITIONED BY (${partition.mkString(", ")})" else ""}
       | STORED BY 'carbondata'
       | ${ if (options.tablePath.nonEmpty) s"LOCATION '${options.tablePath.get}'" else ""}
       |  ${ if (property.nonEmpty) "TBLPROPERTIES (" + property + ")" else "" }
       |
     """.stripMargin
  }

}

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
19
20	import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
21	import org.apache.spark.sql.types._
22	import org.apache.spark.sql.util.CarbonException
23
24	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
25	import org.apache.carbondata.common.logging.LogServiceFactory
26	import org.apache.carbondata.core.metadata.datatype.{DataTypes => CarbonType}
27	import org.apache.carbondata.spark.CarbonOption
28
29	class CarbonDataFrameWriter(sqlContext: SQLContext, val dataFrame: DataFrame) {	2✔
30
31	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	2✔
32
33	def saveAsCarbonFile(parameters: Map[String, String] = Map()): Unit = {	×
34	// create a new table using dataframe's schema and write its content into the table
35	sqlContext.sparkSession.sql(	2✔
36	makeCreateTableString(dataFrame.schema, new CarbonOption(parameters)))	2✔
37	writeToCarbonFile(parameters)	2✔
38	}
39
40	def appendToCarbonFile(parameters: Map[String, String] = Map()): Unit = {	×
41	writeToCarbonFile(parameters)	1✔
42	}
43
44	private def writeToCarbonFile(parameters: Map[String, String] = Map()): Unit = {	×
45	val options = new CarbonOption(parameters)	2✔
46	loadDataFrame(options)	2✔
47	}
48	/**
49	* Loading DataFrame directly without saving DataFrame to CSV files.
50	* @param options
51	*/
52	private def loadDataFrame(options: CarbonOption): Unit = {
53	val header = dataFrame.columns.mkString(",")	2✔
54	CarbonLoadDataCommand(	2✔
55	Some(CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)),	2✔
56	options.tableName,	2✔
57	null,	2✔
58	Seq(),	2✔
59	Map("fileheader" -> header) ++ options.toMap,	2✔
60	isOverwriteTable = options.overwriteEnabled,	2✔
61	null,	2✔
62	Some(dataFrame)).run(sqlContext.sparkSession)	2✔
63	}
64
65	private def convertToCarbonType(sparkType: DataType): String = {
66	sparkType match {	2✔
67	case StringType => CarbonType.STRING.getName	2✔
68	case IntegerType => CarbonType.INT.getName	2✔
69	case ShortType => CarbonType.SHORT.getName	2✔
70	case LongType => CarbonType.LONG.getName	2✔
71	case FloatType => CarbonType.DOUBLE.getName	2✔
72	case DoubleType => CarbonType.DOUBLE.getName	2✔
73	case TimestampType => CarbonType.TIMESTAMP.getName	1✔
74	case DateType => CarbonType.DATE.getName	1✔
75	case decimal: DecimalType => s"decimal(${decimal.precision}, ${decimal.scale})"	1✔
76	case BooleanType => CarbonType.BOOLEAN.getName	1✔
77	case other => CarbonException.analysisException(s"unsupported type: $other")	×
78	}
79	}
80
81	private def makeCreateTableString(schema: StructType, options: CarbonOption): String = {
82	val property = Map(	2✔
83	"SORT_COLUMNS" -> options.sortColumns,	2✔
84	"DICTIONARY_INCLUDE" -> options.dictionaryInclude,	2✔
85	"DICTIONARY_EXCLUDE" -> options.dictionaryExclude,	2✔
86	"LONG_STRING_COLUMNS" -> options.longStringColumns,	2✔
87	"TABLE_BLOCKSIZE" -> options.tableBlockSize,	2✔
88	"TABLE_BLOCKLET_SIZE" -> options.tableBlockletSize,	2✔
89	"STREAMING" -> Option(options.isStreaming.toString)	2✔
90	).filter(_._2.isDefined)	2✔
91	.map(property => s"'${property._1}' = '${property._2.get}'").mkString(",")	2✔
92
93	val partition: Seq[String] = if (options.partitionColumns.isDefined) {	2✔
94	if (options.partitionColumns.get.toSet.size != options.partitionColumns.get.length) {	1✔
95	throw new MalformedCarbonCommandException(s"repeated partition column")	×
96	}
97	options.partitionColumns.get.map { column =>	1✔
98	val field = schema.fields.find(_.name.equalsIgnoreCase(column))	1✔
99	if (field.isEmpty) {	1✔
100	throw new MalformedCarbonCommandException(s"invalid partition column: $column")	×
101	}
102	s"$column ${field.get.dataType.typeName}"	1✔
103	}
104	} else {
105	Seq()	2✔
106	}
107
108	val schemaWithoutPartition = if (options.partitionColumns.isDefined) {	2✔
109	val partitionCols = options.partitionColumns.get	1✔
110	val fields = schema.filterNot {	1✔
111	field => partitionCols.exists(_.equalsIgnoreCase(field.name))	1✔
112	}
113	StructType(fields)	1✔
114	} else {
115	schema	2✔
116	}
117
118	val carbonSchema = schemaWithoutPartition.map { field =>	2✔
119	s"${ field.name } ${ convertToCarbonType(field.dataType) }"	2✔
120	}
121
122	val dbName = CarbonEnv.getDatabaseName(options.dbName)(sqlContext.sparkSession)	2✔
123
124	s"""	2✔
125	\| CREATE TABLE IF NOT EXISTS $dbName.${options.tableName}	2✔
126	\| (${ carbonSchema.mkString(", ") })	2✔
127	\| ${ if (partition.nonEmpty) s"PARTITIONED BY (${partition.mkString(", ")})" else ""}	2✔
128	\| STORED BY 'carbondata'
129	\| ${ if (options.tablePath.nonEmpty) s"LOCATION '${options.tablePath.get}'" else ""}	2✔
130	\| ${ if (property.nonEmpty) "TBLPROPERTIES (" + property + ")" else "" }	2✔
131	\|
132	""".stripMargin	2✔
133	}
134
135	}	2✔

apache / carbondata / 214

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