11596

Build # 11596

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c39622ab2e9a5d806cf9b3ad12b888bd0758983f">c39622ab2<a href="https://github.com/apache/carbondata/commit/c39622ab2e9a5d806cf9b3ad12b888bd0758983f">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/af8fbba4fbde76b5ee908aa63818ecfaa1a6739f">af8fbba4f</a>">af8fbba4f</a><a href="https://github.com/apache/carbondata/commit/c39622ab2e9a5d806cf9b3ad12b888bd0758983f"> into 7541ef25f">7541ef25f</a>

Run Details

56 of 56 new or added lines in 8 files covered. (100.0%)

64088 of 81961 relevant lines covered (78.19%)

1.03 hits per line

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

83.05

/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedPruneRDD.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.carbondata.indexserver

import java.text.SimpleDateFormat
import java.util.Date

import scala.collection.JavaConverters._

import org.apache.hadoop.mapred.TaskAttemptID
import org.apache.hadoop.mapreduce.{InputSplit, Job, TaskType}
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.spark.{Partition, SparkEnv, TaskContext, TaskKilledException}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.hive.DistributionUtil

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.cache.CacheProvider
import org.apache.carbondata.core.datamap.DistributableDataMapFormat
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.ExtendedBlocklet
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.spark.rdd.CarbonRDD
import org.apache.carbondata.spark.util.CarbonScalaUtil

private[indexserver] class DataMapRDDPartition(rddId: Int, idx: Int, val inputSplit: InputSplit)
  extends Partition {

  override def index: Int = idx

  override def hashCode(): Int = 41 * (41 + rddId) + idx
}

private[indexserver] class DistributedPruneRDD(@transient private val ss: SparkSession,
    dataMapFormat: DistributableDataMapFormat)
  extends CarbonRDD[(String, ExtendedBlocklet)](ss, Nil) {

  val executorsList: Set[String] = DistributionUtil.getNodeList(ss.sparkContext).toSet

  @transient private val LOGGER = LogServiceFactory.getLogService(classOf[DistributedPruneRDD]
    .getName)

  private val jobTrackerId: String = {
    val formatter = new SimpleDateFormat("yyyyMMddHHmm")
    formatter.format(new Date())
  }

  override protected def getPreferredLocations(split: Partition): Seq[String] = {
    if (split.asInstanceOf[DataMapRDDPartition].inputSplit.getLocations != null) {
      split.asInstanceOf[DataMapRDDPartition].inputSplit.getLocations.toSeq
    } else {
      Seq()
    }
  }

  override def internalCompute(split: Partition,
      context: TaskContext): Iterator[(String, ExtendedBlocklet)] = {
    val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
    val attemptContext = new TaskAttemptContextImpl(FileFactory.getConfiguration, attemptId)
    val inputSplit = split.asInstanceOf[DataMapRDDPartition].inputSplit
    val reader = dataMapFormat.createRecordReader(inputSplit, attemptContext)
    reader.initialize(inputSplit, attemptContext)
    val cacheSize = if (CacheProvider.getInstance().getCarbonCache != null) {
      CacheProvider.getInstance().getCarbonCache.getCurrentSize
    } else {
      0L
    }
    context.addTaskCompletionListener(_ => {
      if (reader != null) {
        reader.close()
      }
    })
    val iter: Iterator[(String, ExtendedBlocklet)] = new Iterator[(String, ExtendedBlocklet)] {

      private var havePair = false
      private var finished = false

      override def hasNext: Boolean = {
        if (context.isInterrupted) {
          throw new TaskKilledException
        }
        if (!finished && !havePair) {
          finished = !reader.nextKeyValue
          havePair = !finished
        }
        !finished
      }

      override def next(): (String, ExtendedBlocklet) = {
        if (!hasNext) {
          throw new java.util.NoSuchElementException("End of stream")
        }
        havePair = false
        val executorIP = SparkEnv.get.blockManager.blockManagerId.host
        val value = (executorIP + "_" + cacheSize.toString, reader.getCurrentValue)
        value
      }
    }
    iter
  }

  override protected def internalGetPartitions: Array[Partition] = {
    val job = Job.getInstance(FileFactory.getConfiguration)
    val splits = dataMapFormat.getSplits(job).asScala
    val isDistributedPruningEnabled = CarbonProperties.getInstance()
      .isDistributedPruningEnabled(dataMapFormat.getCarbonTable.getDatabaseName,
        dataMapFormat.getCarbonTable.getTableName)
    if (!isDistributedPruningEnabled || dataMapFormat.isFallbackJob || splits.isEmpty) {
      splits.zipWithIndex.map {
        f => new DataMapRDDPartition(id, f._2, f._1)
      }.toArray
    } else {
      val (response, time) = CarbonScalaUtil.logTime {
        DistributedRDDUtils.getExecutors(splits.toArray, executorsList, dataMapFormat
          .getCarbonTable.getTableUniqueName, id)
      }
      LOGGER.debug(s"Time taken to assign executors to ${splits.length} is $time ms")
      response.toArray
    }
  }
}

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.carbondata.indexserver
19
20	import java.text.SimpleDateFormat
21	import java.util.Date
22
23	import scala.collection.JavaConverters._
24
25	import org.apache.hadoop.mapred.TaskAttemptID
26	import org.apache.hadoop.mapreduce.{InputSplit, Job, TaskType}
27	import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
28	import org.apache.spark.{Partition, SparkEnv, TaskContext, TaskKilledException}
29	import org.apache.spark.sql.SparkSession
30	import org.apache.spark.sql.hive.DistributionUtil
31
32	import org.apache.carbondata.common.logging.LogServiceFactory
33	import org.apache.carbondata.core.cache.CacheProvider
34	import org.apache.carbondata.core.datamap.DistributableDataMapFormat
35	import org.apache.carbondata.core.datastore.impl.FileFactory
36	import org.apache.carbondata.core.indexstore.ExtendedBlocklet
37	import org.apache.carbondata.core.util.CarbonProperties
38	import org.apache.carbondata.spark.rdd.CarbonRDD
39	import org.apache.carbondata.spark.util.CarbonScalaUtil
40
41	private[indexserver] class DataMapRDDPartition(rddId: Int, idx: Int, val inputSplit: InputSplit)	2✔
42	extends Partition {	2✔
43
44	override def index: Int = idx	2✔
45
46	override def hashCode(): Int = 41 * (41 + rddId) + idx	×
47	}
48
49	private[indexserver] class DistributedPruneRDD(@transient private val ss: SparkSession,	2✔
50	dataMapFormat: DistributableDataMapFormat)	2✔
51	extends CarbonRDD[(String, ExtendedBlocklet)](ss, Nil) {	2✔
52
53	val executorsList: Set[String] = DistributionUtil.getNodeList(ss.sparkContext).toSet	2✔
54
55	@transient private val LOGGER = LogServiceFactory.getLogService(classOf[DistributedPruneRDD]	2✔
56	.getName)	2✔
57
58	private val jobTrackerId: String = {	2✔
59	val formatter = new SimpleDateFormat("yyyyMMddHHmm")	2✔
60	formatter.format(new Date())	2✔
61	}
62
63	override protected def getPreferredLocations(split: Partition): Seq[String] = {
64	if (split.asInstanceOf[DataMapRDDPartition].inputSplit.getLocations != null) {	2✔
65	split.asInstanceOf[DataMapRDDPartition].inputSplit.getLocations.toSeq	×
66	} else {
67	Seq()	2✔
68	}
69	}
70
71	override def internalCompute(split: Partition,
72	context: TaskContext): Iterator[(String, ExtendedBlocklet)] = {
73	val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)	2✔
74	val attemptContext = new TaskAttemptContextImpl(FileFactory.getConfiguration, attemptId)	2✔
75	val inputSplit = split.asInstanceOf[DataMapRDDPartition].inputSplit	2✔
76	val reader = dataMapFormat.createRecordReader(inputSplit, attemptContext)	2✔
77	reader.initialize(inputSplit, attemptContext)	2✔
78	val cacheSize = if (CacheProvider.getInstance().getCarbonCache != null) {	2✔
79	CacheProvider.getInstance().getCarbonCache.getCurrentSize	2✔
80	} else {
81	0L	×
82	}
83	context.addTaskCompletionListener(_ => {	2✔
84	if (reader != null) {	2✔
85	reader.close()	2✔
86	}
87	})
88	val iter: Iterator[(String, ExtendedBlocklet)] = new Iterator[(String, ExtendedBlocklet)] {	2✔
89
90	private var havePair = false	2✔
91	private var finished = false	2✔
92
93	override def hasNext: Boolean = {	2✔
94	if (context.isInterrupted) {	2✔
95	throw new TaskKilledException	×
96	}
97	if (!finished && !havePair) {	2✔
98	finished = !reader.nextKeyValue	2✔
99	havePair = !finished	2✔
100	}
101	!finished	2✔
102	}
103
104	override def next(): (String, ExtendedBlocklet) = {
105	if (!hasNext) {	2✔
106	throw new java.util.NoSuchElementException("End of stream")	×
107	}
108	havePair = false	2✔
109	val executorIP = SparkEnv.get.blockManager.blockManagerId.host	2✔
110	val value = (executorIP + "_" + cacheSize.toString, reader.getCurrentValue)	2✔
111	value	2✔
112	}
113	}
114	iter	2✔
115	}
116
117	override protected def internalGetPartitions: Array[Partition] = {	2✔
118	val job = Job.getInstance(FileFactory.getConfiguration)	2✔
119	val splits = dataMapFormat.getSplits(job).asScala	2✔
120	val isDistributedPruningEnabled = CarbonProperties.getInstance()	2✔
121	.isDistributedPruningEnabled(dataMapFormat.getCarbonTable.getDatabaseName,	2✔
122	dataMapFormat.getCarbonTable.getTableName)	2✔
123	if (!isDistributedPruningEnabled \|\| dataMapFormat.isFallbackJob \|\| splits.isEmpty) {	2✔
124	splits.zipWithIndex.map {	2✔
125	f => new DataMapRDDPartition(id, f._2, f._1)	2✔
126	}.toArray	2✔
127	} else {
128	val (response, time) = CarbonScalaUtil.logTime {	×
129	DistributedRDDUtils.getExecutors(splits.toArray, executorsList, dataMapFormat	×
130	.getCarbonTable.getTableUniqueName, id)	×
131	}
132	LOGGER.debug(s"Time taken to assign executors to ${splits.length} is $time ms")	×
133	response.toArray	×
134	}
135	}
136	}

apache / carbondata / 11596

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