2179

Build # 2179

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/abbed02f43dfe2ea0441bbad8da6d267446a9a4e">abbed02f4<a href="https://github.com/apache/carbondata/commit/abbed02f43dfe2ea0441bbad8da6d267446a9a4e">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9f752b74081806d5d4f5aece1558b968006b8c7c">9f752b740</a>">9f752b740</a><a href="https://github.com/apache/carbondata/commit/abbed02f43dfe2ea0441bbad8da6d267446a9a4e"> into 04b52568d">04b52568d</a>

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

61507 of 77935 relevant lines covered (78.92%)

1.05 hits per line

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

92.98

/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.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.util.LinkedHashSet

import scala.Array.canBuildFrom
import scala.collection.JavaConverters._

import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan}
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.{CarbonMetastoreTypes, SparkTypeConverter}

import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.metadata.datatype.DataTypes
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.metadata.schema.table.column.{CarbonColumn, CarbonDimension}
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.path.CarbonTablePath

/**
 * Represents logical plan for one carbon table
 */
case class CarbonRelation(
    databaseName: String,
    tableName: String,
    var metaData: CarbonMetaData,
    carbonTable: CarbonTable)
  extends LeafNode with MultiInstanceRelation {

  override def newInstance(): LogicalPlan = {
    CarbonRelation(databaseName, tableName, metaData, carbonTable)
      .asInstanceOf[this.type]
  }

  val dimensionsAttr: Seq[AttributeReference] = {
    val sett = new LinkedHashSet(
      carbonTable.getDimensionByTableName(carbonTable.getTableName)
        .asScala.asJava)
    sett.asScala.toSeq.map(dim => {
      val dimval = metaData.carbonTable
        .getDimensionByName(metaData.carbonTable.getTableName, dim.getColName)
      val output: DataType = dimval.getDataType.getName.toLowerCase match {
        case "array" =>
          CarbonMetastoreTypes.toDataType(
            s"array<${SparkTypeConverter.getArrayChildren(carbonTable, dim.getColName)}>")
        case "struct" =>
          CarbonMetastoreTypes.toDataType(
            s"struct<${SparkTypeConverter.getStructChildren(carbonTable, dim.getColName)}>")
        case "map" =>
          CarbonMetastoreTypes.toDataType(
            s"map<${SparkTypeConverter.getMapChildren(carbonTable, dim.getColName)}>")
        case dType =>
          val dataType = addDecimalScaleAndPrecision(dimval, dType)
          CarbonMetastoreTypes.toDataType(dataType)
      }

      AttributeReference(
        dim.getColName,
        output,
        nullable = true)()
    })
  }

  val measureAttr = {
    val factTable = carbonTable.getTableName
    new LinkedHashSet(
      carbonTable.getMeasureByTableName(carbonTable.getTableName).asScala.asJava).asScala.toSeq
      .map { x =>
      val metastoreType = metaData.carbonTable.getMeasureByName(factTable, x.getColName)
        .getDataType.getName.toLowerCase match {
        case "decimal" => "decimal(" + x.getPrecision + "," + x.getScale + ")"
        case others => others
      }
      AttributeReference(
        x.getColName,
        CarbonMetastoreTypes.toDataType(metastoreType),
        nullable = true)()
    }
  }

  override val output = {
    val columns = carbonTable.getCreateOrderColumn(carbonTable.getTableName)
      .asScala
    val partitionColumnSchemas = if (carbonTable.getPartitionInfo() != null) {
      carbonTable.getPartitionInfo.getColumnSchemaList.asScala
    } else {
      Nil
    }
    val otherColumns = columns.filterNot(a => partitionColumnSchemas.contains(a.getColumnSchema))
    val partitionColumns = columns.filter(a => partitionColumnSchemas.contains(a.getColumnSchema))
    // convert each column to Attribute
    (otherColumns ++= partitionColumns).filter(!_.isInvisible).map { column: CarbonColumn =>
      if (column.isDimension()) {
        val output: DataType = column.getDataType.getName.toLowerCase match {
          case "array" =>
            CarbonMetastoreTypes.toDataType(
              s"array<${SparkTypeConverter.getArrayChildren(carbonTable, column.getColName)}>")
          case "struct" =>
            CarbonMetastoreTypes.toDataType(
              s"struct<${SparkTypeConverter.getStructChildren(carbonTable, column.getColName)}>")
          case "map" =>
            CarbonMetastoreTypes.toDataType(
              s"map<${SparkTypeConverter.getMapChildren(carbonTable, column.getColName)}>")
          case dType =>
            val dataType = SparkTypeConverter.addDecimalScaleAndPrecision(column, dType)
            CarbonMetastoreTypes.toDataType(dataType)
        }
        AttributeReference(column.getColName, output, nullable = true )(
          qualifier = Option(tableName + "." + column.getColName))
      } else {
        val output = CarbonMetastoreTypes.toDataType {
          column.getDataType.getName.toLowerCase match {
            case "decimal" => "decimal(" + column.getColumnSchema.getPrecision + "," + column
              .getColumnSchema.getScale + ")"
            case others => others
          }
        }
        AttributeReference(column.getColName, output, nullable = true)(
          qualifier = Option(tableName + "." + column.getColName))
      }
    }
  }

  // TODO: Use data from the footers.
  // TODO For 2.1
  //  override lazy val statistics = Statistics(sizeInBytes = this.sizeInBytes)
  // Todo for 2.2
  //  override def computeStats(conf: SQLConf): Statistics = Statistics(sizeInBytes =
  //  this.sizeInBytes)

  // override lazy val statistics = Statistics(sizeInBytes = this.sizeInBytes)

  override def equals(other: Any): Boolean = {
    other match {
      case p: CarbonRelation =>
        p.databaseName == databaseName && p.output == output && p.tableName == tableName
      case _ => false
    }
  }

  def addDecimalScaleAndPrecision(dimval: CarbonDimension, dataType: String): String = {
    var dType = dataType
    if (DataTypes.isDecimal(dimval.getDataType)) {
      dType +=
      "(" + dimval.getColumnSchema.getPrecision + "," + dimval.getColumnSchema.getScale + ")"
    }
    dType
  }

  private var tableStatusLastUpdateTime = 0L

  private var sizeInBytesLocalValue = 0L

  def sizeInBytes: Long = {
    if (carbonTable.isExternalTable) {
      val tablePath = carbonTable.getTablePath
      val fileType = FileFactory.getFileType(tablePath)
      if (FileFactory.isFileExist(tablePath, fileType)) {
        sizeInBytesLocalValue = FileFactory.getDirectorySize(tablePath)
      }
    } else {
      val tableStatusNewLastUpdatedTime = SegmentStatusManager.getTableStatusLastModifiedTime(
        carbonTable.getAbsoluteTableIdentifier)
      if (tableStatusLastUpdateTime != tableStatusNewLastUpdatedTime) {
        if (new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
          .getValidAndInvalidSegments.getValidSegments.isEmpty) {
          sizeInBytesLocalValue = 0L
        } else {
          val tablePath = carbonTable.getTablePath
          val fileType = FileFactory.getFileType(tablePath)
          if (FileFactory.isFileExist(tablePath, fileType)) {
            // get the valid segments
            val segments = new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
              .getValidAndInvalidSegments.getValidSegments.asScala
            var size = 0L
            // for each segment calculate the size
            segments.foreach { validSeg =>
              // for older store
              if (null != validSeg.getLoadMetadataDetails.getDataSize &&
                  null != validSeg.getLoadMetadataDetails.getIndexSize) {
                size = size + validSeg.getLoadMetadataDetails.getDataSize.toLong +
                       validSeg.getLoadMetadataDetails.getIndexSize.toLong
              } else {
                size = size + FileFactory.getDirectorySize(
                  CarbonTablePath.getSegmentPath(tablePath, validSeg.getSegmentNo))
              }
            }
            // update the new table status time
            tableStatusLastUpdateTime = tableStatusNewLastUpdatedTime
            // update the new size
            sizeInBytesLocalValue = size
          }
        }
      }
    }

    sizeInBytesLocalValue
  }

}

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	package org.apache.spark.sql.hive
18
19	import java.util.LinkedHashSet
20
21	import scala.Array.canBuildFrom
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
25	import org.apache.spark.sql.catalyst.expressions.AttributeReference
26	import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan}
27	import org.apache.spark.sql.types._
28	import org.apache.spark.sql.util.{CarbonMetastoreTypes, SparkTypeConverter}
29
30	import org.apache.carbondata.core.datastore.impl.FileFactory
31	import org.apache.carbondata.core.metadata.datatype.DataTypes
32	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
33	import org.apache.carbondata.core.metadata.schema.table.column.{CarbonColumn, CarbonDimension}
34	import org.apache.carbondata.core.statusmanager.SegmentStatusManager
35	import org.apache.carbondata.core.util.path.CarbonTablePath
36
37	/**
38	* Represents logical plan for one carbon table
39	*/
40	case class CarbonRelation(	×
41	databaseName: String,	2✔
42	tableName: String,	2✔
43	var metaData: CarbonMetaData,	2✔
44	carbonTable: CarbonTable)	2✔
45	extends LeafNode with MultiInstanceRelation {	2✔
46
47	override def newInstance(): LogicalPlan = {
48	CarbonRelation(databaseName, tableName, metaData, carbonTable)	×
49	.asInstanceOf[this.type]
50	}
51
52	val dimensionsAttr: Seq[AttributeReference] = {	2✔
53	val sett = new LinkedHashSet(	2✔
54	carbonTable.getDimensionByTableName(carbonTable.getTableName)	2✔
55	.asScala.asJava)	2✔
56	sett.asScala.toSeq.map(dim => {	2✔
57	val dimval = metaData.carbonTable	2✔
58	.getDimensionByName(metaData.carbonTable.getTableName, dim.getColName)	2✔
59	val output: DataType = dimval.getDataType.getName.toLowerCase match {	2✔
60	case "array" =>	2✔
61	CarbonMetastoreTypes.toDataType(	2✔
62	s"array<${SparkTypeConverter.getArrayChildren(carbonTable, dim.getColName)}>")	2✔
63	case "struct" =>	2✔
64	CarbonMetastoreTypes.toDataType(	2✔
65	s"struct<${SparkTypeConverter.getStructChildren(carbonTable, dim.getColName)}>")	2✔
66	case "map" =>	2✔
67	CarbonMetastoreTypes.toDataType(	1✔
68	s"map<${SparkTypeConverter.getMapChildren(carbonTable, dim.getColName)}>")	1✔
69	case dType =>	2✔
70	val dataType = addDecimalScaleAndPrecision(dimval, dType)	2✔
71	CarbonMetastoreTypes.toDataType(dataType)	2✔
72	}
73
74	AttributeReference(	2✔
75	dim.getColName,	2✔
76	output,	2✔
77	nullable = true)()	2✔
78	})
79	}
80
81	val measureAttr = {	2✔
82	val factTable = carbonTable.getTableName	2✔
83	new LinkedHashSet(	2✔
84	carbonTable.getMeasureByTableName(carbonTable.getTableName).asScala.asJava).asScala.toSeq	2✔
85	.map { x =>	2✔
86	val metastoreType = metaData.carbonTable.getMeasureByName(factTable, x.getColName)	2✔
87	.getDataType.getName.toLowerCase match {	2✔
88	case "decimal" => "decimal(" + x.getPrecision + "," + x.getScale + ")"	2✔
89	case others => others	2✔
90	}
91	AttributeReference(	2✔
92	x.getColName,	2✔
93	CarbonMetastoreTypes.toDataType(metastoreType),	2✔
94	nullable = true)()	2✔
95	}
96	}
97
98	override val output = {	2✔
99	val columns = carbonTable.getCreateOrderColumn(carbonTable.getTableName)	2✔
100	.asScala	2✔
101	val partitionColumnSchemas = if (carbonTable.getPartitionInfo() != null) {	2✔
102	carbonTable.getPartitionInfo.getColumnSchemaList.asScala	2✔
103	} else {
104	Nil	2✔
105	}
106	val otherColumns = columns.filterNot(a => partitionColumnSchemas.contains(a.getColumnSchema))	2✔
107	val partitionColumns = columns.filter(a => partitionColumnSchemas.contains(a.getColumnSchema))	2✔
108	// convert each column to Attribute
109	(otherColumns ++= partitionColumns).filter(!_.isInvisible).map { column: CarbonColumn =>	2✔
110	if (column.isDimension()) {	2✔
111	val output: DataType = column.getDataType.getName.toLowerCase match {	2✔
112	case "array" =>	2✔
113	CarbonMetastoreTypes.toDataType(	2✔
114	s"array<${SparkTypeConverter.getArrayChildren(carbonTable, column.getColName)}>")	2✔
115	case "struct" =>	2✔
116	CarbonMetastoreTypes.toDataType(	2✔
117	s"struct<${SparkTypeConverter.getStructChildren(carbonTable, column.getColName)}>")	2✔
118	case "map" =>	2✔
119	CarbonMetastoreTypes.toDataType(	1✔
120	s"map<${SparkTypeConverter.getMapChildren(carbonTable, column.getColName)}>")	1✔
121	case dType =>	2✔
122	val dataType = SparkTypeConverter.addDecimalScaleAndPrecision(column, dType)	2✔
123	CarbonMetastoreTypes.toDataType(dataType)	2✔
124	}
125	AttributeReference(column.getColName, output, nullable = true )(	2✔
126	qualifier = Option(tableName + "." + column.getColName))	2✔
127	} else {
128	val output = CarbonMetastoreTypes.toDataType {	2✔
129	column.getDataType.getName.toLowerCase match {	2✔
130	case "decimal" => "decimal(" + column.getColumnSchema.getPrecision + "," + column	2✔
131	.getColumnSchema.getScale + ")"	2✔
132	case others => others	2✔
133	}
134	}
135	AttributeReference(column.getColName, output, nullable = true)(	2✔
136	qualifier = Option(tableName + "." + column.getColName))	2✔
137	}
138	}
139	}
140
141	// TODO: Use data from the footers.
142	// TODO For 2.1
143	// override lazy val statistics = Statistics(sizeInBytes = this.sizeInBytes)
144	// Todo for 2.2
145	// override def computeStats(conf: SQLConf): Statistics = Statistics(sizeInBytes =
146	// this.sizeInBytes)
147
148	// override lazy val statistics = Statistics(sizeInBytes = this.sizeInBytes)
149
150	override def equals(other: Any): Boolean = {
151	other match {	×
152	case p: CarbonRelation =>	×
153	p.databaseName == databaseName && p.output == output && p.tableName == tableName	×
154	case _ => false	×
155	}
156	}
157
158	def addDecimalScaleAndPrecision(dimval: CarbonDimension, dataType: String): String = {
159	var dType = dataType	2✔
160	if (DataTypes.isDecimal(dimval.getDataType)) {	2✔
161	dType +=	2✔
162	"(" + dimval.getColumnSchema.getPrecision + "," + dimval.getColumnSchema.getScale + ")"	2✔
163	}
164	dType	2✔
165	}
166
167	private var tableStatusLastUpdateTime = 0L	2✔
168
169	private var sizeInBytesLocalValue = 0L	2✔
170
171	def sizeInBytes: Long = {
172	if (carbonTable.isExternalTable) {	2✔
173	val tablePath = carbonTable.getTablePath	1✔
174	val fileType = FileFactory.getFileType(tablePath)	1✔
175	if (FileFactory.isFileExist(tablePath, fileType)) {	1✔
176	sizeInBytesLocalValue = FileFactory.getDirectorySize(tablePath)	1✔
177	}
178	} else {
179	val tableStatusNewLastUpdatedTime = SegmentStatusManager.getTableStatusLastModifiedTime(	2✔
180	carbonTable.getAbsoluteTableIdentifier)	2✔
181	if (tableStatusLastUpdateTime != tableStatusNewLastUpdatedTime) {	2✔
182	if (new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)	2✔
183	.getValidAndInvalidSegments.getValidSegments.isEmpty) {	2✔
184	sizeInBytesLocalValue = 0L	2✔
185	} else {
186	val tablePath = carbonTable.getTablePath	2✔
187	val fileType = FileFactory.getFileType(tablePath)	2✔
188	if (FileFactory.isFileExist(tablePath, fileType)) {	2✔
189	// get the valid segments
190	val segments = new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)	2✔
191	.getValidAndInvalidSegments.getValidSegments.asScala	2✔
192	var size = 0L	2✔
193	// for each segment calculate the size
194	segments.foreach { validSeg =>	2✔
195	// for older store
196	if (null != validSeg.getLoadMetadataDetails.getDataSize &&	2✔
197	null != validSeg.getLoadMetadataDetails.getIndexSize) {	2✔
198	size = size + validSeg.getLoadMetadataDetails.getDataSize.toLong +	2✔
199	validSeg.getLoadMetadataDetails.getIndexSize.toLong	2✔
200	} else {
201	size = size + FileFactory.getDirectorySize(	×
202	CarbonTablePath.getSegmentPath(tablePath, validSeg.getSegmentNo))	×
203	}
204	}
205	// update the new table status time
206	tableStatusLastUpdateTime = tableStatusNewLastUpdatedTime	2✔
207	// update the new size
208	sizeInBytesLocalValue = size	2✔
209	}
210	}
211	}
212	}
213
214	sizeInBytesLocalValue	2✔
215	}
216
217	}

apache / carbondata / 2179

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