carbondata git commit: [HOTFIX] Fixed Query performance issue

[ra...@apache.org]

Repository: carbondata
Updated Branches:
  refs/heads/master 8affab843 -> b5609b073


[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/b5609b07
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/b5609b07
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/b5609b07

Branch: refs/heads/master
Commit: b5609b07331437145218b638ba00c16ec96afad6
Parents: 8affab8
Author: kumarvishal09 <ku...@gmail.com>
Authored: Wed Aug 8 23:25:41 2018 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Aug 9 19:06:47 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/b5609b07/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/b5609b07/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(