3209

Build # 3209

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/785739bc0906ea61e06e984bf364e0a815b03176">785739bc0<a href="https://github.com/apache/carbondata/commit/785739bc0906ea61e06e984bf364e0a815b03176">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/af813af4cd1d2bf69ac1668cc1ae5549dfb3b2a7">af813af4c</a>">af813af4c</a><a href="https://github.com/apache/carbondata/commit/785739bc0906ea61e06e984bf364e0a815b03176"> into 7fd8dad68">7fd8dad68</a>

Run Details

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

61186 of 80051 relevant lines covered (76.43%)

0.99 hits per line

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

97.5

/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCountStar.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 scala.collection.JavaConverters._

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapred.JobConf
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.execution.LeafExecNode
import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.sql.types.StringType
import org.apache.spark.unsafe.types.UTF8String

import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
import org.apache.carbondata.core.util.ThreadLocalSessionInfo
import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil

case class CarbonCountStar(
    attributesRaw: Seq[Attribute],
    carbonTable: CarbonTable,
    sparkSession: SparkSession,
    outUnsafeRows: Boolean = true) extends LeafExecNode {

  override def doExecute(): RDD[InternalRow] = {
    ThreadLocalSessionInfo
      .setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())
    val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
    val (job, tableInputFormat) = createCarbonInputFormat(absoluteTableIdentifier)
    CarbonInputFormat.setQuerySegment(job.getConfiguration, carbonTable)

    // get row count
    val rowCount = CarbonUpdateUtil.getRowCount(
      tableInputFormat.getBlockRowCount(
        job,
        carbonTable,
        CarbonFilters.getPartitions(
          Seq.empty,
          sparkSession,
          TableIdentifier(
            carbonTable.getTableName,
            Some(carbonTable.getDatabaseName))).map(_.asJava).orNull, false),
      carbonTable)
    val valueRaw =
      attributesRaw.head.dataType match {
        case StringType => Seq(UTF8String.fromString(Long.box(rowCount).toString)).toArray
          .asInstanceOf[Array[Any]]
        case _ => Seq(Long.box(rowCount)).toArray.asInstanceOf[Array[Any]]
      }
    val value = new GenericInternalRow(valueRaw)
    val unsafeProjection = UnsafeProjection.create(output.map(_.dataType).toArray)
    val row = if (outUnsafeRows) unsafeProjection(value) else value
    sparkContext.parallelize(Seq(row))
  }

  override def output: Seq[Attribute] = {
    attributesRaw
  }

  private def createCarbonInputFormat(absoluteTableIdentifier: AbsoluteTableIdentifier
  ): (Job, CarbonTableInputFormat[Array[Object]]) = {
    val carbonInputFormat = new CarbonTableInputFormat[Array[Object]]()
    val jobConf: JobConf = new JobConf(FileFactory.getConfiguration)
    SparkHadoopUtil.get.addCredentials(jobConf)
    CarbonInputFormat.setTableInfo(jobConf, carbonTable.getTableInfo)
    val job = new Job(jobConf)
    FileInputFormat.addInputPath(job, new Path(absoluteTableIdentifier.getTablePath))
    CarbonInputFormat
      .setTransactionalTable(job.getConfiguration,
        carbonTable.getTableInfo.isTransactionalTable)
    CarbonInputFormatUtil.setDataMapJobIfConfigured(job.getConfiguration)
    (job, carbonInputFormat)
  }
}

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 scala.collection.JavaConverters._
21
22	import org.apache.hadoop.conf.Configuration
23	import org.apache.hadoop.fs.Path
24	import org.apache.hadoop.mapred.JobConf
25	import org.apache.hadoop.mapreduce.Job
26	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
27	import org.apache.spark.deploy.SparkHadoopUtil
28	import org.apache.spark.rdd.RDD
29	import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
30	import org.apache.spark.sql.catalyst.expressions._
31	import org.apache.spark.sql.execution.LeafExecNode
32	import org.apache.spark.sql.optimizer.CarbonFilters
33	import org.apache.spark.sql.types.StringType
34	import org.apache.spark.unsafe.types.UTF8String
35
36	import org.apache.carbondata.core.datastore.impl.FileFactory
37	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
38	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
39	import org.apache.carbondata.core.mutate.CarbonUpdateUtil
40	import org.apache.carbondata.core.util.ThreadLocalSessionInfo
41	import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
42	import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
43
44	case class CarbonCountStar(	2✔
45	attributesRaw: Seq[Attribute],	2✔
46	carbonTable: CarbonTable,	2✔
47	sparkSession: SparkSession,	2✔
48	outUnsafeRows: Boolean = true) extends LeafExecNode {	2✔
49
50	override def doExecute(): RDD[InternalRow] = {
51	ThreadLocalSessionInfo
52	.setConfigurationToCurrentThread(sparkSession.sessionState.newHadoopConf())	2✔
53	val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier	2✔
54	val (job, tableInputFormat) = createCarbonInputFormat(absoluteTableIdentifier)	2✔
55	CarbonInputFormat.setQuerySegment(job.getConfiguration, carbonTable)	2✔
56
57	// get row count
58	val rowCount = CarbonUpdateUtil.getRowCount(	2✔
59	tableInputFormat.getBlockRowCount(	2✔
60	job,	2✔
61	carbonTable,	2✔
62	CarbonFilters.getPartitions(	2✔
63	Seq.empty,	2✔
64	sparkSession,	2✔
65	TableIdentifier(	2✔
66	carbonTable.getTableName,	2✔
67	Some(carbonTable.getDatabaseName))).map(_.asJava).orNull, false),	2✔
68	carbonTable)	2✔
69	val valueRaw =	2✔
70	attributesRaw.head.dataType match {	2✔
71	case StringType => Seq(UTF8String.fromString(Long.box(rowCount).toString)).toArray	2✔
72	.asInstanceOf[Array[Any]]	×
73	case _ => Seq(Long.box(rowCount)).toArray.asInstanceOf[Array[Any]]	2✔
74	}
75	val value = new GenericInternalRow(valueRaw)	2✔
76	val unsafeProjection = UnsafeProjection.create(output.map(_.dataType).toArray)	2✔
77	val row = if (outUnsafeRows) unsafeProjection(value) else value	2✔
78	sparkContext.parallelize(Seq(row))	2✔
79	}
80
81	override def output: Seq[Attribute] = {
82	attributesRaw	2✔
83	}
84
85	private def createCarbonInputFormat(absoluteTableIdentifier: AbsoluteTableIdentifier
86	): (Job, CarbonTableInputFormat[Array[Object]]) = {
87	val carbonInputFormat = new CarbonTableInputFormat[Array[Object]]()	2✔
88	val jobConf: JobConf = new JobConf(FileFactory.getConfiguration)	2✔
89	SparkHadoopUtil.get.addCredentials(jobConf)	2✔
90	CarbonInputFormat.setTableInfo(jobConf, carbonTable.getTableInfo)	2✔
91	val job = new Job(jobConf)	2✔
92	FileInputFormat.addInputPath(job, new Path(absoluteTableIdentifier.getTablePath))	2✔
93	CarbonInputFormat
94	.setTransactionalTable(job.getConfiguration,	2✔
95	carbonTable.getTableInfo.isTransactionalTable)	2✔
96	CarbonInputFormatUtil.setDataMapJobIfConfigured(job.getConfiguration)	2✔
97	(job, carbonInputFormat)	2✔
98	}
99	}

apache / carbondata / 3209

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