[SQL-DS-CACHE-190]put plasma detector in seperate object to avoid unn…

…ecessary dependency of arrow (#191)
oap-project · Aug 2, 2021 · f225d0a · f225d0a
1 parent 6bc51ce
commit f225d0a
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 22 deletions.
diff --git a/...1/main/scala/org/apache/spark/sql/execution/datasources/oap/filecache/MemoryManager.scala b/...1/main/scala/org/apache/spark/sql/execution/datasources/oap/filecache/MemoryManager.scala
@@ -31,7 +31,6 @@ import org.apache.spark.internal.config.ConfigEntry
 import org.apache.spark.memory.MemoryMode
 import org.apache.spark.sql.execution.datasources.OapException
 import org.apache.spark.sql.execution.datasources.oap.filecache.FiberType.FiberType
-import org.apache.spark.sql.execution.datasources.oap.filecache.OapCache.plasmaServerDetect
 import org.apache.spark.sql.execution.datasources.oap.utils.PersistentMemoryConfigUtils
 import org.apache.spark.sql.internal.oap.OapConf
 import org.apache.spark.storage.{BlockManager, TestBlockId}
@@ -135,7 +134,7 @@ private[sql] object MemoryManager extends Logging {
       case "tmp" => new TmpDramMemoryManager(sparkEnv)
       case "kmem" => new DaxKmemMemoryManager(sparkEnv)
       case "plasma" =>
-        if (plasmaServerDetect(sparkEnv)) {
+        if (ExternalCacheDetector.plasmaServerDetect(sparkEnv)) {
           new PlasmaMemoryManager(sparkEnv)
         } else {
           new OffHeapMemoryManager(sparkEnv)
@@ -164,7 +163,7 @@ private[sql] object MemoryManager extends Logging {
       case "noevict" => new HybridMemoryManager(sparkEnv)
       case "vmem" => new TmpDramMemoryManager(sparkEnv)
       case "external" =>
-        if (plasmaServerDetect(sparkEnv)) {
+        if (ExternalCacheDetector.plasmaServerDetect(sparkEnv)) {
           new PlasmaMemoryManager(sparkEnv)
         } else {
           new OffHeapMemoryManager(sparkEnv)

diff --git a/...-3.1.1/main/scala/org/apache/spark/sql/execution/datasources/oap/filecache/OapCache.scala b/...-3.1.1/main/scala/org/apache/spark/sql/execution/datasources/oap/filecache/OapCache.scala
@@ -217,24 +217,6 @@ private[filecache] class CacheGuardian(maxMemory: Long) extends Thread with Logg
 }
 
 private[filecache] object OapCache extends Logging {
-  def plasmaServerDetect(sparkEnv: SparkEnv): Boolean = {
-    val socket = sparkEnv.conf.get(OapConf.OAP_EXTERNAL_CACHE_SOCKET_PATH)
-    try {
-      System.loadLibrary("plasma_java")
-    } catch {
-      case e: Exception => logError(s"load plasma jni lib failed " + e.getMessage)
-    }
-    var plasmaDetected: Boolean = true;
-    try {
-      val conn: plasma.PlasmaClient = new plasma.PlasmaClient(socket, "", 0)
-    } catch {
-      case e: PlasmaClientException =>
-        logWarning("External cache strategy requires plasma-store-server launched, " +
-          "failed to detect plasma-store-server, will fallback to simpleCache." + e.getMessage)
-        plasmaDetected = false;
-    }
-    plasmaDetected
-  }
   def cacheFallBackDetect(sparkEnv: SparkEnv,
                           fallBackEnabled: Boolean = true,
                           fallBackRes: Boolean = true): Boolean = {
@@ -307,7 +289,7 @@ private[filecache] object OapCache extends Logging {
 
     oapCacheOpt match {
       case "external" =>
-        if (plasmaServerDetect(sparkEnv)) new ExternalCache(fiberType)
+        if (ExternalCacheDetector.plasmaServerDetect(sparkEnv)) new ExternalCache(fiberType)
         else new SimpleOapCache()
       case "guava" =>
         if (cacheFallBackDetect(sparkEnv, fallBackEnabled.toBoolean, fallBackRes.toBoolean)) {
@@ -962,6 +944,27 @@ class MixCache(dataCacheMemory: Long,
   }
 }
 
+object ExternalCacheDetector extends Logging {
+  def plasmaServerDetect(sparkEnv: SparkEnv): Boolean = {
+    val socket = sparkEnv.conf.get(OapConf.OAP_EXTERNAL_CACHE_SOCKET_PATH)
+    try {
+      System.loadLibrary("plasma_java")
+    } catch {
+      case e: Exception => logError(s"load plasma jni lib failed " + e.getMessage)
+    }
+    var plasmaDetected: Boolean = true;
+    try {
+      val conn: plasma.PlasmaClient = new plasma.PlasmaClient(socket, "", 0)
+    } catch {
+      case e: plasma.exceptions.PlasmaClientException =>
+        logWarning("External cache strategy requires plasma-store-server launched, " +
+          "failed to detect plasma-store-server, will fallback to simpleCache." + e.getMessage)
+        plasmaDetected = false;
+    }
+    plasmaDetected
+  }
+}
+
 class ExternalCache(fiberType: FiberType) extends OapCache with Logging {
   private val conf = SparkEnv.get.conf
   private val externalStoreCacheSocket: String =