12041

Build # 12041

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ea951311ef235f1dbbf1cee4246c2958dae33bb8">ea951311e<a href="https://github.com/apache/carbondata/commit/ea951311ef235f1dbbf1cee4246c2958dae33bb8">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/07d4efa41b23d8d6d5c77cfbabd8af22d9ba8424">07d4efa41</a>">07d4efa41</a><a href="https://github.com/apache/carbondata/commit/ea951311ef235f1dbbf1cee4246c2958dae33bb8"> into 91a799847">91a799847</a>

Run Details

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

65000 of 83565 relevant lines covered (77.78%)

1.02 hits per line

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

77.01

/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)
    }
    val groupedPartitions = convertToPartition(sortedPartitions,
      legacySegments,
      tableUniqueName,
      executorsList).groupBy {
        partition =>
          partition.getLocations.head
      }
    groupedPartitions.zipWithIndex.map {
      case ((location, splitList), index) =>
        new DataMapRDDPartition(rddId,
          index, splitList,
          Array(location))
    }.toArray.sortBy(_.index)
  }

  private def convertToPartition(segments: Seq[InputSplit], legacySegments: Seq[InputSplit],
      tableUniqueName: String,
      executorList: Map[String, Seq[String]]): Seq[InputSplit] = {
    if (legacySegments.nonEmpty) {
      val validExecutorIds = executorList.flatMap {
        case (host, executors) => executors.map {
          executor => s"${host}_$executor"
        }
      }.toSeq
      legacySegments.zipWithIndex.map {
        case (legacySegment, index) =>
          val wrapper: DataMapDistributable = legacySegment
            .asInstanceOf[DataMapDistributableWrapper].getDistributable
          val executor = validExecutorIds(index % validExecutorIds.length)
          wrapper.setLocations(Array("executor_" + executor))
          legacySegment
      }
    } else { Seq() } ++ segments.map { partition =>
      val wrapper: DataMapDistributable = partition.asInstanceOf[DataMapDistributableWrapper]
        .getDistributable
      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)
          }
      }
    }
  }

  /**
   * Remove the invalid segment mapping from index server when segments are compacted or become
   * invalid.
   */
  def invalidateSegmentMapping(tableUniqueName: String,
      invalidSegmentList: Seq[String]): Unit = {
    synchronized {
      if (tableToExecutorMapping.get(tableUniqueName) != null) {
        invalidSegmentList.foreach {
          invalidSegment => tableToExecutorMapping.get(tableUniqueName).remove(invalidSegment)
        }
        if (tableToExecutorMapping.get(tableUniqueName).isEmpty) {
          invalidateTableMapping(tableUniqueName)
        }
      }
    }
  }

  /**
   * Remove the table mapping from index server when the table is dropped.
   */
  def invalidateTableMapping(tableUniqueName: String): Unit = {
    synchronized {
      tableToExecutorMapping.remove(tableUniqueName)
    }
  }

  /**
   * 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	val groupedPartitions = convertToPartition(sortedPartitions,	2✔
68	legacySegments,	2✔
69	tableUniqueName,	2✔
70	executorsList).groupBy {	2✔
71	partition =>	2✔
72	partition.getLocations.head	2✔
73	}
74	groupedPartitions.zipWithIndex.map {	2✔
75	case ((location, splitList), index) =>	2✔
76	new DataMapRDDPartition(rddId,	2✔
77	index, splitList,	2✔
78	Array(location))	2✔
79	}.toArray.sortBy(_.index)	2✔
80	}
81
82	private def convertToPartition(segments: Seq[InputSplit], legacySegments: Seq[InputSplit],
83	tableUniqueName: String,
84	executorList: Map[String, Seq[String]]): Seq[InputSplit] = {
85	if (legacySegments.nonEmpty) {	2✔
86	val validExecutorIds = executorList.flatMap {	×
87	case (host, executors) => executors.map {	×
88	executor => s"${host}_$executor"	×
89	}
90	}.toSeq	×
91	legacySegments.zipWithIndex.map {	×
92	case (legacySegment, index) =>	×
93	val wrapper: DataMapDistributable = legacySegment	×
94	.asInstanceOf[DataMapDistributableWrapper].getDistributable	×
95	val executor = validExecutorIds(index % validExecutorIds.length)	×
96	wrapper.setLocations(Array("executor_" + executor))	×
97	legacySegment	×
98	}
99	} else { Seq() } ++ segments.map { partition =>	2✔
100	val wrapper: DataMapDistributable = partition.asInstanceOf[DataMapDistributableWrapper]	2✔
101	.getDistributable	2✔
102	wrapper.setLocations(Array(DistributedRDDUtils	2✔
103	.assignExecutor(tableUniqueName, wrapper.getSegment, executorList)))	2✔
104	partition	2✔
105	}
106	}
107
108	/**
109	* Update the cache size returned by the executors to the driver mapping.
110	*/
111	def updateExecutorCacheSize(cacheSizes: Set[String]): Unit = {
112	synchronized {
113	cacheSizes.foreach {	×
114	executorCacheSize =>	×
115	// executorCacheSize would be in the form of 127.0.0.1_10024 where the left of '_'
116	// would be the executor IP and the right would be the cache that executor is holding.
117	val hostAndExecutor = executorCacheSize.substring(0,	×
118	executorCacheSize.lastIndexOf('_'))	×
119	val (host, executor) = (hostAndExecutor	×
120	.substring(0, hostAndExecutor.lastIndexOf('_')), hostAndExecutor	×
121	.substring(hostAndExecutor.lastIndexOf('_') + 1, hostAndExecutor.length))	×
122	val size = executorCacheSize.substring(executorCacheSize.lastIndexOf('_') + 1,	×
123	executorCacheSize.length)	×
124	val executorMapping = executorToCacheSizeMapping.get(host)	×
125	if (executorMapping != null) {	×
126	executorMapping.put(executor, size.toLong)	×
127	executorToCacheSizeMapping.put(host, executorMapping)	×
128	}
129	}
130	}
131	}
132
133	/**
134	* Remove the invalid segment mapping from index server when segments are compacted or become
135	* invalid.
136	*/
137	def invalidateSegmentMapping(tableUniqueName: String,
138	invalidSegmentList: Seq[String]): Unit = {
139	synchronized {
140	if (tableToExecutorMapping.get(tableUniqueName) != null) {	2✔
141	invalidSegmentList.foreach {	×
142	invalidSegment => tableToExecutorMapping.get(tableUniqueName).remove(invalidSegment)	×
143	}
144	if (tableToExecutorMapping.get(tableUniqueName).isEmpty) {	×
145	invalidateTableMapping(tableUniqueName)	×
146	}
147	}
148	}
149	}
150
151	/**
152	* Remove the table mapping from index server when the table is dropped.
153	*/
154	def invalidateTableMapping(tableUniqueName: String): Unit = {
155	synchronized {	2✔
156	tableToExecutorMapping.remove(tableUniqueName)	2✔
157	}
158	}
159
160	/**
161	* Invalidate the dead executors from the mapping and assign the segments to some other
162	* executor, so that the query can load the segments to the new assigned executor.
163	*/
164	def invalidateHosts(invalidHosts: Seq[String]): Unit = {
165	synchronized {
166	val validInvalidExecutors: Map[String, String] = invalidHosts.flatMap {	2✔
167	host =>	2✔
168	val invalidExecutorToSizeMapping = executorToCacheSizeMapping.remove(host)	2✔
169	invalidExecutorToSizeMapping.asScala.map {	2✔
170	case (invalidExecutor, size) =>	2✔
171	getLeastLoadedExecutor match {	2✔
172	case Some((reassignedHost, reassignedExecutorId)) =>	2✔
173	val existingExecutorMapping = executorToCacheSizeMapping.get(reassignedHost)	2✔
174	if (existingExecutorMapping != null) {	2✔
175	val existingSize = existingExecutorMapping.get(reassignedExecutorId)	2✔
176	existingExecutorMapping.put(reassignedExecutorId, existingSize + size)	2✔
177	} else {
178	existingExecutorMapping.put(reassignedExecutorId, size)	×
179	}
180	executorToCacheSizeMapping.put(reassignedHost, existingExecutorMapping)	2✔
181	s"${host}_$invalidExecutor" -> s"${ reassignedHost }_$reassignedExecutorId"	2✔
182	case None => "" -> ""	2✔
183	}
184	}
185	}.toMap	2✔
186	updateTableMappingForInvalidExecutors(validInvalidExecutors)	2✔
187	}
188	}
189
190	private def updateTableMappingForInvalidExecutors(validInvalidExecutors: Map[String, String]) {
191	// remove all invalidExecutor mapping from cache.
192	for ((tableName: String, segmentToExecutorMapping) <- tableToExecutorMapping.asScala) {	2✔
193	// find the invalid executor in cache.
194	val newSegmentToExecutorMapping = new ConcurrentHashMap[String, String]()	2✔
195	val existingMapping = tableToExecutorMapping.get(tableName)	2✔
196	segmentToExecutorMapping.asScala.collect {	2✔
197	case (segmentNumber, executorUniqueName) if validInvalidExecutors	2✔
198	.contains(executorUniqueName) =>	2✔
199	val newExecutorId = validInvalidExecutors(executorUniqueName)	2✔
200	// remove mapping for the invalid executor.
201	val executorIdSplits = newExecutorId.split("_")	2✔
202	val (host, executorId) = (executorIdSplits(0), executorIdSplits(1))	2✔
203	// find a new executor for the segment
204	newSegmentToExecutorMapping	2✔
205	.put(segmentNumber, s"${ host }_$executorId")	2✔
206	}
207	existingMapping.putAll(newSegmentToExecutorMapping)	2✔
208	tableToExecutorMapping.put(tableName, existingMapping)	2✔
209	}
210	}
211
212	def invalidateExecutors(invalidExecutors: Seq[String]): Unit = {
213	synchronized {
214	val validInvalidExecutors: Map[String, String] = invalidExecutors.map {	2✔
215	invalidExecutor =>	2✔
216	val executorIdSplits = invalidExecutor.split("_")	2✔
217	val (host, executor) = (executorIdSplits(0), executorIdSplits(1))	2✔
218	val invalidExecutorSize = executorToCacheSizeMapping.get(host).remove(executor)	2✔
219	getLeastLoadedExecutor match {	2✔
220	case Some((reassignedHost, reassignedExecutorId)) =>	2✔
221	val existingExecutorMapping = executorToCacheSizeMapping.get(reassignedHost)	2✔
222	if (existingExecutorMapping != null) {	2✔
223	val existingSize = existingExecutorMapping.get(reassignedExecutorId)	2✔
224	existingExecutorMapping	2✔
225	.put(reassignedExecutorId, existingSize + invalidExecutorSize)	2✔
226	} else {
227	existingExecutorMapping.put(reassignedExecutorId, invalidExecutorSize)	×
228	}
229	executorToCacheSizeMapping.put(reassignedHost, existingExecutorMapping)	2✔
230	invalidExecutor -> s"${ reassignedHost }_$reassignedExecutorId"	2✔
231	case None => "" -> ""	×
232	}
233	}.toMap	2✔
234	updateTableMappingForInvalidExecutors(validInvalidExecutors)	2✔
235	}
236	}
237
238	/**
239	* Sorts the executor cache based on the size each one is handling and returns the least of them.
240	*
241	* @return
242	*/
243	private def getLeastLoadedExecutor: Option[(String, String)] = {
244	val leastHostExecutor = executorToCacheSizeMapping.asScala.flatMap {	2✔
245	case (host, executorToCacheMap) =>	2✔
246	executorToCacheMap.asScala.map {	2✔
247	case (executor, size) =>	2✔
248	(host, executor, size)	2✔
249	}
250	}.toSeq.sortWith(_._3 < _._3).toList	2✔
251	leastHostExecutor match {	2✔
252	case head :: _ =>	2✔
253	Some(head._1, head._2)	2✔
254	case _ => None	2✔
255	}
256	}
257
258	private def checkForUnassignedExecutors(validExecutorIds: Seq[String]): Option[String] = {
259	val usedExecutorIds = executorToCacheSizeMapping.asScala.flatMap {	2✔
260	case (host, executorMap) =>	2✔
261	executorMap.keySet().asScala.map {	2✔
262	executor => s"${ host }_$executor"	2✔
263	}
264	}
265	val unassignedExecutor = validExecutorIds.diff(usedExecutorIds.toSeq)	2✔
266	unassignedExecutor.headOption	2✔
267	}
268
269	/**
270	* Assign a executor for the current segment. If a executor was previously assigned to the
271	* segment then the same would be returned.
272	*
273	* @return
274	*/
275	def assignExecutor(tableUniqueName: String,	2✔
276	segment: Segment,
277	validExecutors: Map[String, Seq[String]]): String = {
278	val segmentMapping = tableToExecutorMapping.get(tableUniqueName)	2✔
279	lazy val executor = segmentMapping.get(segment.getSegmentNo)	2✔
280	if (segmentMapping != null && executor != null) {	2✔
281	s"executor_$executor"	2✔
282	} else {
283	// check if any executor is not assigned. If yes then give priority to that executor
284	// otherwise get the executor which has handled the least size.
285	val validExecutorIds = validExecutors.flatMap {	2✔
286	case (host, executors) => executors.map {	2✔
287	executor => s"${host}_$executor"	2✔
288	}
289	}.toSeq	2✔
290	val unassignedExecutor = checkForUnassignedExecutors(validExecutorIds)	2✔
291	val (newHost, newExecutor) = if (unassignedExecutor.nonEmpty) {	2✔
292	val freeExecutor = unassignedExecutor.get.split("_")	2✔
293	(freeExecutor(0), freeExecutor(1))	2✔
294	} else {
295	getLeastLoadedExecutor match {	2✔
296	case Some((host, executorID)) => (host, executorID)	2✔
297	case None => throw new RuntimeException("Could not find any alive executors.")	×
298	}
299	}
300	val existingExecutorMapping = executorToCacheSizeMapping.get(newHost)	2✔
301	if (existingExecutorMapping != null) {	2✔
302	val existingSize = existingExecutorMapping.get(newExecutor)	2✔
303	if (existingSize != null) {	2✔
304	existingExecutorMapping.put(newExecutor, existingSize + segment.getIndexSize	2✔
305	.toInt)	2✔
306	} else {
307	existingExecutorMapping.put(newExecutor, segment.getIndexSize	×
308	.toInt)	×
309	}
310	} else {
311	val newExecutorMapping = new ConcurrentHashMap[String, Long]()	2✔
312	newExecutorMapping.put(newExecutor, segment.getIndexSize)	2✔
313	executorToCacheSizeMapping.put(newHost, newExecutorMapping)	2✔
314	}
315	val existingSegmentMapping = tableToExecutorMapping.get(tableUniqueName)	2✔
316	if (existingSegmentMapping == null) {	2✔
317	val newSegmentMapping = new ConcurrentHashMap[String, String]()	2✔
318	newSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")	2✔
319	tableToExecutorMapping.put(tableUniqueName, newSegmentMapping)	2✔
320	} else {
321	existingSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")	2✔
322	tableToExecutorMapping.put(tableUniqueName, existingSegmentMapping)	2✔
323	}
324	s"executor_${newHost}_$newExecutor"	2✔
325	}
326	}
327
328	}	2✔

apache / carbondata / 12041

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