11829

Build # 11829

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/5ccb2f55712a5d1cacdd943988f77193f68ab1f8">5ccb2f557<a href="https://github.com/apache/carbondata/commit/5ccb2f55712a5d1cacdd943988f77193f68ab1f8">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9928a69ce27c7f9e344b9bed2d51a35f12c4e1be">9928a69ce</a>">9928a69ce</a><a href="https://github.com/apache/carbondata/commit/5ccb2f55712a5d1cacdd943988f77193f68ab1f8"> into afef23270">afef23270</a>

Run Details

288 of 288 new or added lines in 12 files covered. (100.0%)

64140 of 82867 relevant lines covered (77.4%)

1.01 hits per line

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

0.0

/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.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 scala.collection.JavaConverters._

import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.hive.DistributionUtil

import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapFactory
import org.apache.carbondata.hadoop.CarbonInputSplit
import org.apache.carbondata.spark.rdd.CarbonRDD

class DistributedShowCacheRDD(@transient private val ss: SparkSession, tableName: String)
  extends CarbonRDD[String](ss, Nil) {

  val executorsList: Array[String] = DistributionUtil.getExecutors(ss.sparkContext).flatMap {
    case (host, executors) =>
      executors.map {
        executor => s"executor_${host}_$executor"
      }
  }.toArray

  override protected def internalGetPartitions: Array[Partition] = {
    executorsList.zipWithIndex.map {
      case (executor, idx) =>
        // create a dummy split for each executor to accumulate the cache size.
        val dummySplit = new CarbonInputSplit()
        dummySplit.setLocation(Array(executor))
        new DataMapRDDPartition(id, idx, dummySplit)
    }
  }

  override def internalCompute(split: Partition, context: TaskContext): Iterator[String] = {
    val dataMaps = DataMapStoreManager.getInstance().getAllDataMaps.asScala
    val tableList = tableName.split(",").map(_.replace("-", "_"))
    val iterator = dataMaps.collect {
      case (table, tableDataMaps) if tableName.isEmpty || tableList.contains(table) =>
        val sizeAndIndexLengths = tableDataMaps.asScala
          .map { dataMap =>
            val dataMapName = if (dataMap.getDataMapFactory.isInstanceOf[BlockletDataMapFactory]) {
              table
            } else {
              dataMap.getDataMapSchema.getRelationIdentifier.getDatabaseName + "_" + dataMap
              .getDataMapSchema.getDataMapName
            }
            s"${ dataMapName }:${ dataMap.getDataMapFactory.getCacheSize }:${
              dataMap.getDataMapSchema.getProviderName} "
          }
        sizeAndIndexLengths
    }.flatten.toIterator
    iterator
  }
}

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.carbondata.indexserver
18
19	import scala.collection.JavaConverters._
20
21	import org.apache.spark.{Partition, TaskContext}
22	import org.apache.spark.sql.SparkSession
23	import org.apache.spark.sql.hive.DistributionUtil
24
25	import org.apache.carbondata.core.datamap.DataMapStoreManager
26	import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapFactory
27	import org.apache.carbondata.hadoop.CarbonInputSplit
28	import org.apache.carbondata.spark.rdd.CarbonRDD
29
30	class DistributedShowCacheRDD(@transient private val ss: SparkSession, tableName: String)	×
31	extends CarbonRDD[String](ss, Nil) {	×
32
33	val executorsList: Array[String] = DistributionUtil.getExecutors(ss.sparkContext).flatMap {	×
34	case (host, executors) =>	×
35	executors.map {	×
36	executor => s"executor_${host}_$executor"	×
37	}
38	}.toArray	×
39
40	override protected def internalGetPartitions: Array[Partition] = {
41	executorsList.zipWithIndex.map {	×
42	case (executor, idx) =>	×
43	// create a dummy split for each executor to accumulate the cache size.
44	val dummySplit = new CarbonInputSplit()	×
45	dummySplit.setLocation(Array(executor))	×
46	new DataMapRDDPartition(id, idx, dummySplit)	×
47	}
48	}
49
50	override def internalCompute(split: Partition, context: TaskContext): Iterator[String] = {
51	val dataMaps = DataMapStoreManager.getInstance().getAllDataMaps.asScala	×
52	val tableList = tableName.split(",").map(_.replace("-", "_"))	×
53	val iterator = dataMaps.collect {	×
54	case (table, tableDataMaps) if tableName.isEmpty \|\| tableList.contains(table) =>	×
55	val sizeAndIndexLengths = tableDataMaps.asScala	×
56	.map { dataMap =>	×
57	val dataMapName = if (dataMap.getDataMapFactory.isInstanceOf[BlockletDataMapFactory]) {	×
58	table	×
59	} else {
60	dataMap.getDataMapSchema.getRelationIdentifier.getDatabaseName + "_" + dataMap	×
61	.getDataMapSchema.getDataMapName	×
62	}
63	s"${ dataMapName }:${ dataMap.getDataMapFactory.getCacheSize }:${	×
64	dataMap.getDataMapSchema.getProviderName} "	×
65	}
66	sizeAndIndexLengths	×
67	}.flatten.toIterator	×
68	iterator	×
69	}
70	}

apache / carbondata / 11829

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