11764

Build # 11764

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/f2f82d06c45e7eb22763c11a4ca0c913935f1b55">f2f82d06c<a href="https://github.com/apache/carbondata/commit/f2f82d06c45e7eb22763c11a4ca0c913935f1b55">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1add819ac07e49225d92c367ca09b4959da6869c">1add819ac</a>">1add819ac</a><a href="https://github.com/apache/carbondata/commit/f2f82d06c45e7eb22763c11a4ca0c913935f1b55"> into 1baa485d8">1baa485d8</a>

Run Details

325 of 325 new or added lines in 13 files covered. (100.0%)

64102 of 82839 relevant lines covered (77.38%)

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

apache / carbondata / 11764

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