[40/47] carbondata git commit: [HOTFIX] Fixed Query performance issue

[ra...@apache.org]

[HOTFIX] Fixed Query performance issue

Problem: Query performance is slower than 1.4
Root cause: In CarbonScanRDD it is getting TaskContext onCompleteCallbacks Field using reflection to check whether already InsertTaskCompletionListener is already added or not to avoid adding two taskcompletion listener in case of insert into. This Scala reflection call is taking ~2 seconds for each query task. Because of this query is slower.
Solution: Now added Java reflection to get the onCompleteCallbacks field from TaskContextImpl class
Query Time:
Old: ~59 seconds
New: ~33 seconds

This closes #2622


Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/ca460831
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/ca460831
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/ca460831

Branch: refs/heads/branch-1.4
Commit: ca46083118f82aca42a0f3ea3de53df2f0e74023
Parents: 887489c
Author: kumarvishal09 <ku...@gmail.com>
Authored: Wed Aug 8 23:25:41 2018 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Aug 9 23:51:36 2018 +0530

----------------------------------------------------------------------
 .../org/apache/carbondata/spark/rdd/CarbonScanRDD.scala     | 3 ++-
 .../scala/org/apache/spark/util/CarbonReflectionUtils.scala | 9 +++++----
 2 files changed, 7 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/ca460831/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index 6b43999..38062a4 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -463,9 +463,10 @@ class CarbonScanRDD[T: ClassTag](
       model.setStatisticsRecorder(recorder)
 
       // TODO: rewrite this logic to call free memory in FailureListener on failures. On success,
-      // no memory leak should be there, resources should be freed on success completion.
+      // TODO: no memory leak should be there, resources should be freed on success completion.
       val listeners = CarbonReflectionUtils.getField("onCompleteCallbacks", context)
         .asInstanceOf[ArrayBuffer[TaskCompletionListener]]
+
       val isAdded = listeners.exists(p => p.isInstanceOf[InsertTaskCompletionListener])
       model.setFreeUnsafeMemory(!isAdded)
       // add task completion before calling initialize as initialize method will internally call

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ca460831/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
index 4264aa1..1061e98 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.util
 
+import java.lang.reflect.Field
+
 import scala.reflect.runtime._
 import scala.reflect.runtime.universe._
 
@@ -50,10 +52,9 @@ object CarbonReflectionUtils {
    * @return
    */
   def getField[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
-    val im = rm.reflect(obj)
-
-    im.symbol.typeSignature.members.find(_.name.toString.equals(name))
-      .map(l => im.reflectField(l.asTerm).get).getOrElse(null)
+    val field = obj.getClass.getDeclaredField(name)
+    field.setAccessible(true)
+    field.get(obj)
   }
 
   def getUnresolvedRelation(