11663

Build # 11663

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/15574ac6151b78e04e0b83b82b2c4c7248366f2f">15574ac61">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/15574ac6151b78e04e0b83b82b2c4c7248366f2f">15574ac61</a><a href="https://github.com/apache/carbondata/commit/15574ac6151b78e04e0b83b82b2c4c7248366f2f">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/15574ac6151b78e04e0b83b82b2c4c7248366f2f&quot;&gt;15574ac61&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/15574ac6151b78e04e0b83b82b2c4c7248366f2f&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

137 of 137 new or added lines in 3 files covered. (100.0%)

63809 of 82435 relevant lines covered (77.41%)

1.01 hits per line

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

89.54

/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.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.util.concurrent.ConcurrentHashMap

import scala.collection.JavaConverters._

import org.apache.hadoop.mapreduce.InputSplit
import org.apache.spark.Partition

import org.apache.carbondata.core.datamap.{DataMapDistributable, Segment}
import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper

object DistributedRDDUtils {
  // Segment number to executorNode mapping
  val tableToExecutorMapping: ConcurrentHashMap[String, ConcurrentHashMap[String, String]] =
    new ConcurrentHashMap[String, ConcurrentHashMap[String, String]]()

  // executorNode to segmentSize mapping
  val executorToCacheSizeMapping: ConcurrentHashMap[String, ConcurrentHashMap[String, Long]] =
    new ConcurrentHashMap[String, ConcurrentHashMap[String, Long]]()

  def getExecutors(segment: Array[InputSplit], executorsList : Map[String, Seq[String]],
      tableUniqueName: String, rddId: Int): Seq[Partition] = {
    // sort the partitions in increasing order of index size.
    val (segments, legacySegments) = segment.span(split => split
      .asInstanceOf[DataMapDistributableWrapper].getDistributable.getSegment.getIndexSize > 0)
    val sortedPartitions = segments.sortWith(_.asInstanceOf[DataMapDistributableWrapper]
                                              .getDistributable.getSegment.getIndexSize >
                                            _.asInstanceOf[DataMapDistributableWrapper]
                                              .getDistributable.getSegment.getIndexSize)
    val executorCache = DistributedRDDUtils.executorToCacheSizeMapping
    // check if any executor is dead.
    val invalidHosts = executorCache.keySet().asScala.diff(executorsList.keySet)
    if (invalidHosts.nonEmpty) {
      // extract the dead executor host name
      DistributedRDDUtils.invalidateHosts(invalidHosts.toSeq)
    }
    val invalidExecutorIds = executorsList.collect {
      case (host, executors) if executorCache.get(host) != null =>
        val toBeRemovedExecutors = executorCache.get(host).keySet().asScala.diff(executors.toSet)
        if (executors.size == toBeRemovedExecutors.size) {
          DistributedRDDUtils.invalidateHosts(Seq(host))
          Seq()
        } else {
          toBeRemovedExecutors.map(executor => host + "_" + executor)
        }
    }.flatten
    if (invalidExecutorIds.nonEmpty) {
      DistributedRDDUtils.invalidateExecutors(invalidExecutorIds.toSeq)
    }
    (convertToPartition(legacySegments, tableUniqueName, executorsList) ++
     convertToPartition(sortedPartitions, tableUniqueName, executorsList)).zipWithIndex.map {
      case (dataMapDistributable, index) =>
        new DataMapRDDPartition(rddId, index, dataMapDistributable)
    }
  }

  private def convertToPartition(segments: Seq[InputSplit], tableUniqueName: String,
      executorList: Map[String, Seq[String]]): Seq[InputSplit] = {
    segments.map { partition =>
      val wrapper: DataMapDistributable = partition.asInstanceOf[DataMapDistributableWrapper]
        .getDistributable
      if (wrapper.getSegment.getIndexSize == 0L) {
        wrapper.getSegment.setIndexSize(1L)
      }
      wrapper.setLocations(Array(DistributedRDDUtils
        .assignExecutor(tableUniqueName, wrapper.getSegment, executorList)))
      partition
    }
  }

  /**
   * Update the cache size returned by the executors to the driver mapping.
   */
  def updateExecutorCacheSize(cacheSizes: Set[String]): Unit = {
    synchronized {
      cacheSizes.foreach {
        executorCacheSize =>
          // executorCacheSize would be in the form of 127.0.0.1_10024 where the left of '_'
          // would be the executor IP and the right would be the cache that executor is holding.
          val hostAndExecutor = executorCacheSize.substring(0,
            executorCacheSize.lastIndexOf('_'))
          val (host, executor) = (hostAndExecutor
            .substring(0, hostAndExecutor.lastIndexOf('_')), hostAndExecutor
            .substring(hostAndExecutor.lastIndexOf('_') + 1, hostAndExecutor.length))
          val size = executorCacheSize.substring(executorCacheSize.lastIndexOf('_') + 1,
            executorCacheSize.length)
          val executorMapping = executorToCacheSizeMapping.get(host)
          if (executorMapping != null) {
            executorMapping.put(executor, size.toLong)
            executorToCacheSizeMapping.put(host, executorMapping)
          }
      }
    }
  }

  def invalidateCache(tableUniqueName: String): Unit = {
    tableToExecutorMapping.keySet().asScala.foreach {
      key =>
        if (key.equalsIgnoreCase(tableUniqueName)) {
          tableToExecutorMapping.remove(key)
        }
    }
  }

  /**
   * Invalidate the dead executors from the mapping and assign the segments to some other
   * executor, so that the query can load the segments to the new assigned executor.
   */
  def invalidateHosts(invalidHosts: Seq[String]): Unit = {
    synchronized {
      val validInvalidExecutors: Map[String, String] = invalidHosts.flatMap {
        host =>
          val invalidExecutorToSizeMapping = executorToCacheSizeMapping.remove(host)
          invalidExecutorToSizeMapping.asScala.map {
            case (invalidExecutor, size) =>
              getLeastLoadedExecutor match {
                case Some((reassignedHost, reassignedExecutorId)) =>
                  val existingExecutorMapping = executorToCacheSizeMapping.get(reassignedHost)
                  if (existingExecutorMapping != null) {
                    val existingSize = existingExecutorMapping.get(reassignedExecutorId)
                    existingExecutorMapping.put(reassignedExecutorId, existingSize + size)
                  } else {
                    existingExecutorMapping.put(reassignedExecutorId, size)
                  }
                  executorToCacheSizeMapping.put(reassignedHost, existingExecutorMapping)
                  s"${host}_$invalidExecutor" -> s"${ reassignedHost }_$reassignedExecutorId"
                case None => "" -> ""
              }
          }
      }.toMap
      updateTableMappingForInvalidExecutors(validInvalidExecutors)
    }
  }

  private def updateTableMappingForInvalidExecutors(validInvalidExecutors: Map[String, String]) {
    // remove all invalidExecutor mapping from cache.
    for ((tableName: String, segmentToExecutorMapping) <- tableToExecutorMapping.asScala) {
      // find the invalid executor in cache.
      val newSegmentToExecutorMapping = new ConcurrentHashMap[String, String]()
      val existingMapping = tableToExecutorMapping.get(tableName)
      segmentToExecutorMapping.asScala.collect {
        case (segmentNumber, executorUniqueName) if validInvalidExecutors
          .contains(executorUniqueName) =>
          val newExecutorId = validInvalidExecutors(executorUniqueName)
          // remove mapping for the invalid executor.
          val executorIdSplits = newExecutorId.split("_")
          val (host, executorId) = (executorIdSplits(0), executorIdSplits(1))
          // find a new executor for the segment
          newSegmentToExecutorMapping
            .put(segmentNumber, s"${ host }_$executorId")
      }
      existingMapping.putAll(newSegmentToExecutorMapping)
      tableToExecutorMapping.put(tableName, existingMapping)
    }
  }

  def invalidateExecutors(invalidExecutors: Seq[String]): Unit = {
    synchronized {
      val validInvalidExecutors: Map[String, String] = invalidExecutors.map {
        invalidExecutor =>
          val executorIdSplits = invalidExecutor.split("_")
          val (host, executor) = (executorIdSplits(0), executorIdSplits(1))
          val invalidExecutorSize = executorToCacheSizeMapping.get(host).remove(executor)
          getLeastLoadedExecutor match {
            case Some((reassignedHost, reassignedExecutorId)) =>
              val existingExecutorMapping = executorToCacheSizeMapping.get(reassignedHost)
              if (existingExecutorMapping != null) {
                val existingSize = existingExecutorMapping.get(reassignedExecutorId)
                existingExecutorMapping
                  .put(reassignedExecutorId, existingSize + invalidExecutorSize)
              } else {
                existingExecutorMapping.put(reassignedExecutorId, invalidExecutorSize)
              }
              executorToCacheSizeMapping.put(reassignedHost, existingExecutorMapping)
              invalidExecutor -> s"${ reassignedHost }_$reassignedExecutorId"
            case None => "" -> ""
          }
      }.toMap
      updateTableMappingForInvalidExecutors(validInvalidExecutors)
    }
  }

  /**
   * Sorts the executor cache based on the size each one is handling and returns the least of them.
   *
   * @return
   */
  private def getLeastLoadedExecutor: Option[(String, String)] = {
    val leastHostExecutor = executorToCacheSizeMapping.asScala.flatMap {
      case (host, executorToCacheMap) =>
        executorToCacheMap.asScala.map {
          case (executor, size) =>
            (host, executor, size)
        }
    }.toSeq.sortWith(_._3 < _._3).toList
    leastHostExecutor match {
      case head :: _ =>
        Some(head._1, head._2)
      case _ => None
    }
  }

  private def checkForUnassignedExecutors(validExecutorIds: Seq[String]): Option[String] = {
    val usedExecutorIds = executorToCacheSizeMapping.asScala.flatMap {
      case (host, executorMap) =>
        executorMap.keySet().asScala.map {
          executor => s"${ host }_$executor"
        }
    }
    val unassignedExecutor = validExecutorIds.diff(usedExecutorIds.toSeq)
    unassignedExecutor.headOption
  }

  /**
   * Assign a executor for the current segment. If a executor was previously assigned to the
   * segment then the same would be returned.
   *
   * @return
   */
  def assignExecutor(tableUniqueName: String,
      segment: Segment,
      validExecutors: Map[String, Seq[String]]): String = {
    val segmentMapping = tableToExecutorMapping.get(tableUniqueName)
    lazy val executor = segmentMapping.get(segment.getSegmentNo)
    if (segmentMapping != null && executor != null) {
      s"executor_$executor"
    } else {
      // check if any executor is not assigned. If yes then give priority to that executor
      // otherwise get the executor which has handled the least size.
      val validExecutorIds = validExecutors.flatMap {
        case (host, executors) => executors.map {
          executor => s"${host}_$executor"
        }
      }.toSeq
      val unassignedExecutor = checkForUnassignedExecutors(validExecutorIds)
      val (newHost, newExecutor) = if (unassignedExecutor.nonEmpty) {
        val freeExecutor = unassignedExecutor.get.split("_")
        (freeExecutor(0), freeExecutor(1))
      } else {
        getLeastLoadedExecutor match {
          case Some((host, executorID)) => (host, executorID)
          case None => throw new RuntimeException("Could not find any alive executors.")
        }
      }
      val existingExecutorMapping = executorToCacheSizeMapping.get(newHost)
      if (existingExecutorMapping != null) {
        val existingSize = existingExecutorMapping.get(newExecutor)
        if (existingSize != null) {
          existingExecutorMapping.put(newExecutor, existingSize + segment.getIndexSize
            .toInt)
        } else {
          existingExecutorMapping.put(newExecutor, segment.getIndexSize
            .toInt)
        }
      } else {
        val newExecutorMapping = new ConcurrentHashMap[String, Long]()
        newExecutorMapping.put(newExecutor, segment.getIndexSize)
        executorToCacheSizeMapping.put(newHost, newExecutorMapping)
      }
      val existingSegmentMapping = tableToExecutorMapping.get(tableUniqueName)
      if (existingSegmentMapping == null) {
        val newSegmentMapping = new ConcurrentHashMap[String, String]()
        newSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")
        tableToExecutorMapping.put(tableUniqueName, newSegmentMapping)
      } else {
        existingSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")
        tableToExecutorMapping.put(tableUniqueName, existingSegmentMapping)
      }
      s"executor_${newHost}_$newExecutor"
    }
  }

}

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 java.util.concurrent.ConcurrentHashMap
20
21	import scala.collection.JavaConverters._
22
23	import org.apache.hadoop.mapreduce.InputSplit
24	import org.apache.spark.Partition
25
26	import org.apache.carbondata.core.datamap.{DataMapDistributable, Segment}
27	import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper
28
29	object DistributedRDDUtils {
30	// Segment number to executorNode mapping
31	val tableToExecutorMapping: ConcurrentHashMap[String, ConcurrentHashMap[String, String]] =	2✔
32	new ConcurrentHashMap[String, ConcurrentHashMap[String, String]]()	2✔
33
34	// executorNode to segmentSize mapping
35	val executorToCacheSizeMapping: ConcurrentHashMap[String, ConcurrentHashMap[String, Long]] =	2✔
36	new ConcurrentHashMap[String, ConcurrentHashMap[String, Long]]()	2✔
37
38	def getExecutors(segment: Array[InputSplit], executorsList : Map[String, Seq[String]],
39	tableUniqueName: String, rddId: Int): Seq[Partition] = {
40	// sort the partitions in increasing order of index size.
41	val (segments, legacySegments) = segment.span(split => split	2✔
42	.asInstanceOf[DataMapDistributableWrapper].getDistributable.getSegment.getIndexSize > 0)	2✔
43	val sortedPartitions = segments.sortWith(_.asInstanceOf[DataMapDistributableWrapper]	2✔
44	.getDistributable.getSegment.getIndexSize >	2✔
45	_.asInstanceOf[DataMapDistributableWrapper]	2✔
46	.getDistributable.getSegment.getIndexSize)	2✔
47	val executorCache = DistributedRDDUtils.executorToCacheSizeMapping	2✔
48	// check if any executor is dead.
49	val invalidHosts = executorCache.keySet().asScala.diff(executorsList.keySet)	2✔
50	if (invalidHosts.nonEmpty) {	2✔
51	// extract the dead executor host name
52	DistributedRDDUtils.invalidateHosts(invalidHosts.toSeq)	2✔
53	}
54	val invalidExecutorIds = executorsList.collect {	2✔
55	case (host, executors) if executorCache.get(host) != null =>	2✔
56	val toBeRemovedExecutors = executorCache.get(host).keySet().asScala.diff(executors.toSet)	×
57	if (executors.size == toBeRemovedExecutors.size) {	×
58	DistributedRDDUtils.invalidateHosts(Seq(host))	×
59	Seq()	×
60	} else {
61	toBeRemovedExecutors.map(executor => host + "_" + executor)	×
62	}
63	}.flatten	2✔
64	if (invalidExecutorIds.nonEmpty) {	2✔
65	DistributedRDDUtils.invalidateExecutors(invalidExecutorIds.toSeq)	×
66	}
67	(convertToPartition(legacySegments, tableUniqueName, executorsList) ++	2✔
68	convertToPartition(sortedPartitions, tableUniqueName, executorsList)).zipWithIndex.map {	2✔
69	case (dataMapDistributable, index) =>	2✔
70	new DataMapRDDPartition(rddId, index, dataMapDistributable)	2✔
71	}
72	}
73
74	private def convertToPartition(segments: Seq[InputSplit], tableUniqueName: String,
75	executorList: Map[String, Seq[String]]): Seq[InputSplit] = {
76	segments.map { partition =>	2✔
77	val wrapper: DataMapDistributable = partition.asInstanceOf[DataMapDistributableWrapper]	2✔
78	.getDistributable	2✔
79	if (wrapper.getSegment.getIndexSize == 0L) {	2✔
80	wrapper.getSegment.setIndexSize(1L)	×
81	}
82	wrapper.setLocations(Array(DistributedRDDUtils	2✔
83	.assignExecutor(tableUniqueName, wrapper.getSegment, executorList)))	2✔
84	partition	2✔
85	}
86	}
87
88	/**
89	* Update the cache size returned by the executors to the driver mapping.
90	*/
91	def updateExecutorCacheSize(cacheSizes: Set[String]): Unit = {
92	synchronized {
93	cacheSizes.foreach {	2✔
94	executorCacheSize =>	2✔
95	// executorCacheSize would be in the form of 127.0.0.1_10024 where the left of '_'
96	// would be the executor IP and the right would be the cache that executor is holding.
97	val hostAndExecutor = executorCacheSize.substring(0,	2✔
98	executorCacheSize.lastIndexOf('_'))	2✔
99	val (host, executor) = (hostAndExecutor	2✔
100	.substring(0, hostAndExecutor.lastIndexOf('_')), hostAndExecutor	2✔
101	.substring(hostAndExecutor.lastIndexOf('_') + 1, hostAndExecutor.length))	2✔
102	val size = executorCacheSize.substring(executorCacheSize.lastIndexOf('_') + 1,	2✔
103	executorCacheSize.length)	2✔
104	val executorMapping = executorToCacheSizeMapping.get(host)	2✔
105	if (executorMapping != null) {	2✔
106	executorMapping.put(executor, size.toLong)	×
107	executorToCacheSizeMapping.put(host, executorMapping)	×
108	}
109	}
110	}
111	}
112
113	def invalidateCache(tableUniqueName: String): Unit = {
114	tableToExecutorMapping.keySet().asScala.foreach {	2✔
115	key =>	2✔
116	if (key.equalsIgnoreCase(tableUniqueName)) {	2✔
117	tableToExecutorMapping.remove(key)	×
118	}
119	}
120	}
121
122	/**
123	* Invalidate the dead executors from the mapping and assign the segments to some other
124	* executor, so that the query can load the segments to the new assigned executor.
125	*/
126	def invalidateHosts(invalidHosts: Seq[String]): Unit = {
127	synchronized {
128	val validInvalidExecutors: Map[String, String] = invalidHosts.flatMap {	2✔
129	host =>	2✔
130	val invalidExecutorToSizeMapping = executorToCacheSizeMapping.remove(host)	2✔
131	invalidExecutorToSizeMapping.asScala.map {	2✔
132	case (invalidExecutor, size) =>	2✔
133	getLeastLoadedExecutor match {	2✔
134	case Some((reassignedHost, reassignedExecutorId)) =>	2✔
135	val existingExecutorMapping = executorToCacheSizeMapping.get(reassignedHost)	2✔
136	if (existingExecutorMapping != null) {	2✔
137	val existingSize = existingExecutorMapping.get(reassignedExecutorId)	2✔
138	existingExecutorMapping.put(reassignedExecutorId, existingSize + size)	2✔
139	} else {
140	existingExecutorMapping.put(reassignedExecutorId, size)	×
141	}
142	executorToCacheSizeMapping.put(reassignedHost, existingExecutorMapping)	2✔
143	s"${host}_$invalidExecutor" -> s"${ reassignedHost }_$reassignedExecutorId"	2✔
144	case None => "" -> ""	2✔
145	}
146	}
147	}.toMap	2✔
148	updateTableMappingForInvalidExecutors(validInvalidExecutors)	2✔
149	}
150	}
151
152	private def updateTableMappingForInvalidExecutors(validInvalidExecutors: Map[String, String]) {
153	// remove all invalidExecutor mapping from cache.
154	for ((tableName: String, segmentToExecutorMapping) <- tableToExecutorMapping.asScala) {	2✔
155	// find the invalid executor in cache.
156	val newSegmentToExecutorMapping = new ConcurrentHashMap[String, String]()	2✔
157	val existingMapping = tableToExecutorMapping.get(tableName)	2✔
158	segmentToExecutorMapping.asScala.collect {	2✔
159	case (segmentNumber, executorUniqueName) if validInvalidExecutors	2✔
160	.contains(executorUniqueName) =>	2✔
161	val newExecutorId = validInvalidExecutors(executorUniqueName)	2✔
162	// remove mapping for the invalid executor.
163	val executorIdSplits = newExecutorId.split("_")	2✔
164	val (host, executorId) = (executorIdSplits(0), executorIdSplits(1))	2✔
165	// find a new executor for the segment
166	newSegmentToExecutorMapping	2✔
167	.put(segmentNumber, s"${ host }_$executorId")	2✔
168	}
169	existingMapping.putAll(newSegmentToExecutorMapping)	2✔
170	tableToExecutorMapping.put(tableName, existingMapping)	2✔
171	}
172	}
173
174	def invalidateExecutors(invalidExecutors: Seq[String]): Unit = {
175	synchronized {
176	val validInvalidExecutors: Map[String, String] = invalidExecutors.map {	2✔
177	invalidExecutor =>	2✔
178	val executorIdSplits = invalidExecutor.split("_")	2✔
179	val (host, executor) = (executorIdSplits(0), executorIdSplits(1))	2✔
180	val invalidExecutorSize = executorToCacheSizeMapping.get(host).remove(executor)	2✔
181	getLeastLoadedExecutor match {	2✔
182	case Some((reassignedHost, reassignedExecutorId)) =>	2✔
183	val existingExecutorMapping = executorToCacheSizeMapping.get(reassignedHost)	2✔
184	if (existingExecutorMapping != null) {	2✔
185	val existingSize = existingExecutorMapping.get(reassignedExecutorId)	2✔
186	existingExecutorMapping	2✔
187	.put(reassignedExecutorId, existingSize + invalidExecutorSize)	2✔
188	} else {
189	existingExecutorMapping.put(reassignedExecutorId, invalidExecutorSize)	×
190	}
191	executorToCacheSizeMapping.put(reassignedHost, existingExecutorMapping)	2✔
192	invalidExecutor -> s"${ reassignedHost }_$reassignedExecutorId"	2✔
193	case None => "" -> ""	×
194	}
195	}.toMap	2✔
196	updateTableMappingForInvalidExecutors(validInvalidExecutors)	2✔
197	}
198	}
199
200	/**
201	* Sorts the executor cache based on the size each one is handling and returns the least of them.
202	*
203	* @return
204	*/
205	private def getLeastLoadedExecutor: Option[(String, String)] = {
206	val leastHostExecutor = executorToCacheSizeMapping.asScala.flatMap {	2✔
207	case (host, executorToCacheMap) =>	2✔
208	executorToCacheMap.asScala.map {	2✔
209	case (executor, size) =>	2✔
210	(host, executor, size)	2✔
211	}
212	}.toSeq.sortWith(_._3 < _._3).toList	2✔
213	leastHostExecutor match {	2✔
214	case head :: _ =>	2✔
215	Some(head._1, head._2)	2✔
216	case _ => None	2✔
217	}
218	}
219
220	private def checkForUnassignedExecutors(validExecutorIds: Seq[String]): Option[String] = {
221	val usedExecutorIds = executorToCacheSizeMapping.asScala.flatMap {	2✔
222	case (host, executorMap) =>	2✔
223	executorMap.keySet().asScala.map {	2✔
224	executor => s"${ host }_$executor"	2✔
225	}
226	}
227	val unassignedExecutor = validExecutorIds.diff(usedExecutorIds.toSeq)	2✔
228	unassignedExecutor.headOption	2✔
229	}
230
231	/**
232	* Assign a executor for the current segment. If a executor was previously assigned to the
233	* segment then the same would be returned.
234	*
235	* @return
236	*/
237	def assignExecutor(tableUniqueName: String,	2✔
238	segment: Segment,
239	validExecutors: Map[String, Seq[String]]): String = {
240	val segmentMapping = tableToExecutorMapping.get(tableUniqueName)	2✔
241	lazy val executor = segmentMapping.get(segment.getSegmentNo)	2✔
242	if (segmentMapping != null && executor != null) {	2✔
243	s"executor_$executor"	2✔
244	} else {
245	// check if any executor is not assigned. If yes then give priority to that executor
246	// otherwise get the executor which has handled the least size.
247	val validExecutorIds = validExecutors.flatMap {	2✔
248	case (host, executors) => executors.map {	2✔
249	executor => s"${host}_$executor"	2✔
250	}
251	}.toSeq	2✔
252	val unassignedExecutor = checkForUnassignedExecutors(validExecutorIds)	2✔
253	val (newHost, newExecutor) = if (unassignedExecutor.nonEmpty) {	2✔
254	val freeExecutor = unassignedExecutor.get.split("_")	2✔
255	(freeExecutor(0), freeExecutor(1))	2✔
256	} else {
257	getLeastLoadedExecutor match {	2✔
258	case Some((host, executorID)) => (host, executorID)	2✔
259	case None => throw new RuntimeException("Could not find any alive executors.")	×
260	}
261	}
262	val existingExecutorMapping = executorToCacheSizeMapping.get(newHost)	2✔
263	if (existingExecutorMapping != null) {	2✔
264	val existingSize = existingExecutorMapping.get(newExecutor)	2✔
265	if (existingSize != null) {	2✔
266	existingExecutorMapping.put(newExecutor, existingSize + segment.getIndexSize	2✔
267	.toInt)	2✔
268	} else {
269	existingExecutorMapping.put(newExecutor, segment.getIndexSize	×
270	.toInt)	×
271	}
272	} else {
273	val newExecutorMapping = new ConcurrentHashMap[String, Long]()	2✔
274	newExecutorMapping.put(newExecutor, segment.getIndexSize)	2✔
275	executorToCacheSizeMapping.put(newHost, newExecutorMapping)	2✔
276	}
277	val existingSegmentMapping = tableToExecutorMapping.get(tableUniqueName)	2✔
278	if (existingSegmentMapping == null) {	2✔
279	val newSegmentMapping = new ConcurrentHashMap[String, String]()	2✔
280	newSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")	2✔
281	tableToExecutorMapping.put(tableUniqueName, newSegmentMapping)	2✔
282	} else {
283	existingSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")	2✔
284	tableToExecutorMapping.put(tableUniqueName, existingSegmentMapping)	2✔
285	}
286	s"executor_${newHost}_$newExecutor"	2✔
287	}
288	}
289
290	}	2✔

apache / carbondata / 11663

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