[CARBONDATA-3459] Fixed id based distribution for showcache command

Problem: Currently tasks are not being fired based on the executor ID because getPrefferedLocation was not overridden. Solution: override getPreferredLocations in the ShowCache and InvalidateCacheRDD to fire tasks at the appropriate location This closes #3315
apache · Jul 15, 2019 · a682f98 · a682f98
1 parent b017253
commit a682f98
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 4 deletions.
diff --git a/...gration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.scala b/...gration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.scala
@@ -316,10 +316,10 @@ object DistributedRDDUtils {
       if (existingSegmentMapping == null) {
         val newSegmentMapping = new ConcurrentHashMap[String, String]()
         newSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")
-        tableToExecutorMapping.put(tableUniqueName, newSegmentMapping)
+        tableToExecutorMapping.putIfAbsent(tableUniqueName, newSegmentMapping)
       } else {
-        existingSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")
-        tableToExecutorMapping.put(tableUniqueName, existingSegmentMapping)
+        existingSegmentMapping.putIfAbsent(segment.getSegmentNo, s"${newHost}_$newExecutor")
+        tableToExecutorMapping.putIfAbsent(tableUniqueName, existingSegmentMapping)
       }
       s"executor_${newHost}_$newExecutor"
     }

diff --git a/...ion/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.scala b/...ion/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.scala
@@ -37,6 +37,14 @@ class DistributedShowCacheRDD(@transient private val ss: SparkSession, tableName
       }
   }.toArray
 
+  override protected def getPreferredLocations(split: Partition): Seq[String] = {
+    if (split.asInstanceOf[DataMapRDDPartition].getLocations != null) {
+      split.asInstanceOf[DataMapRDDPartition].getLocations.toSeq
+    } else {
+      Seq()
+    }
+  }
+
   override protected def internalGetPartitions: Array[Partition] = {
     executorsList.zipWithIndex.map {
       case (executor, idx) =>

diff --git a/...n/spark2/src/main/scala/org/apache/carbondata/indexserver/InvalidateSegmentCacheRDD.scala b/...n/spark2/src/main/scala/org/apache/carbondata/indexserver/InvalidateSegmentCacheRDD.scala
@@ -30,14 +30,27 @@ import org.apache.carbondata.spark.rdd.CarbonRDD
 class InvalidateSegmentCacheRDD(@transient private val ss: SparkSession, carbonTable: CarbonTable,
     invalidSegmentIds: List[String]) extends CarbonRDD[String](ss, Nil) {
 
-  val executorsList: Array[String] = DistributionUtil.getNodeList(ss.sparkContext)
+  val executorsList: Array[String] = DistributionUtil.getExecutors(ss.sparkContext).flatMap {
+    case (host, executors) =>
+      executors.map {
+        executor => s"executor_${host}_$executor"
+      }
+  }.toArray
 
   override def internalCompute(split: Partition,
       context: TaskContext): Iterator[String] = {
     DataMapStoreManager.getInstance().clearInvalidSegments(carbonTable, invalidSegmentIds.asJava)
     Iterator.empty
   }
 
+  override protected def getPreferredLocations(split: Partition): Seq[String] = {
+    if (split.asInstanceOf[DataMapRDDPartition].getLocations != null) {
+      split.asInstanceOf[DataMapRDDPartition].getLocations.toSeq
+    } else {
+      Seq()
+    }
+  }
+
   override protected def internalGetPartitions: Array[Partition] = {
     if (invalidSegmentIds.isEmpty) {
       Array()