Use disk to store and transport large results

- push results from CachedDataFrame to disk in case the result exceeds a limit (4MB by
  default)
- to reduce number of disk files keep appending to the same file till a higher limit of 8X
  the initial size (i.e. 32MB by default)
- use a separately generated broadcast ID to generate BroadBlockIds used for registering
  these disk blocks against Spark's BlockManager; the broadcast ID starts at Long.Max/2 to
  avoid any overlaps with the real broadcasted blocks
- each compressed block (4MB size by default) is separate one having its size at the start
  so that the reader can separately decompress those blocks independently since there is no
  clean way to switch streams from underneath a compressed stream or append to existing
  compressed data
- cleanup is done in four places: a) after end reader consumes a disk block, b) in case
  there is an exception while generating the results, c) registered with SnappySession to be
  cleared with session clear() and in its finalizer, d) in case there is a top-level
  failure in CachedDataFrame.collectWithHandler execution
- increase the default stack depth shown in UI to 50 from 20
- refactored and cleaned up CompressionUtils
- updated the docs to correct the defaults mentioned for storageMaxFraction and other
  related properties, and expanded the help text on those properties a bit
- updated the docs about VSD and where to find it

Author: Sumedh Wale

aqp

@@ -44,8 +44,9 @@ import io.snappydata.{Constant, Property, QueryHint}
 import org.apache.spark.serializer.{KryoSerializerPool, StructTypeSerializer}
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.expressions
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
-import org.apache.spark.sql.collection.Utils
+import org.apache.spark.sql.collection.{GenerateFlatIterator, Utils}
 import org.apache.spark.sql.types._
 import org.apache.spark.storage.RDDBlockId
 import org.apache.spark.util.SnappyUtils
@@ -129,13 +130,12 @@ class SparkSQLExecuteImpl(
       //   shipping/compilation etc and lack of proper BlockManager usage in
       //   messaging + server-side final processing, so do it selectively)
       val partitionBlocks = df match {
-        case cdf: CachedDataFrame => cdf.collectWithHandler(CachedDataFrame,
-           CachedDataFrame.localBlockStoreResultHandler(rddId, bm),
+        case cdf: CachedDataFrame => cdf.collectWithHandler(CachedDataFrame(session),
+           CachedDataFrame.localBlockStoreResultHandler(rddId, bm, session),
            CachedDataFrame.localBlockStoreDecoder(querySchema.length, bm))
-        case dataFrame: DataFrame => {
+        case dataFrame: DataFrame =>
           Iterator(CachedDataFrame(null,
-            dataFrame.queryExecution.executedPlan.executeCollect().iterator)._1)
-        }
+            dataFrame.queryExecution.executedPlan.executeCollect().iterator, -1L)._1)
       }
       hdos.clearForReuse()
       SparkSQLExecuteImpl.writeMetaData(srh, hdos, tableNames, nullability, getColumnNames,
@@ -146,12 +146,14 @@ class SparkSQLExecuteImpl(
         block match {
           case null => // skip but still id has to be incremented
           case data: Array[Byte] => if (data.length > 0) {
+            hdos.writeInt(data.length)
             hdos.write(data)
           }
           case p: RDDBlockId =>
             val partitionData = Utils.getPartitionData(p, bm)
             // remove the block once a local handle to it has been obtained
             bm.removeBlock(p, tellMaster = false)
+            hdos.writeInt(partitionData.remaining())
             hdos.write(partitionData)
         }
         logTrace(s"Writing data for partition ID = $id: $block")
@@ -408,8 +410,14 @@ object SparkSQLExecuteImpl {
     }
     val execRow = new ValueRow(dvds)
     val numFields = types.length
-    val unsafeRows = CachedDataFrame.decodeUnsafeRows(numFields,
-      input.array(), input.position(), input.available())
+    val unsafeRows = new GenerateFlatIterator[UnsafeRow, ByteArrayDataInput](input => {
+      if (input.available() > 0) {
+        val size = input.readInt()
+        val pos = input.position()
+        input.setPosition(pos + size)
+        (CachedDataFrame.decodeUnsafeRows(numFields, input.array(), pos, size), input)
+      } else (GenerateFlatIterator.TERMINATE, input)
+    }, input)
     unsafeRows.map { row =>
       var index = 0
       var refTypeIndex = 0

cluster/src/main/scala/io/snappydata/gemxd/SparkSQLExecuteImpl.scala

@@ -224,7 +224,7 @@ object SnappyInterpreterExecute {
           "scala code execution", "", "ComputeDB", "Cluster")
       }
     }
-    val id: Long = session.getUniqueIdForExecScala()
+    val id: Long = session.getUniqueIdForExecScala
     val intpHelper = SnappyInterpreterExecute.getOrCreateStateHolder(id, user, authToken, group)
     try {
       intpHelper.interpret(code.split("\n"), options) match {

cluster/src/main/scala/io/snappydata/remote/interpreter/SnappyInterpreterExecute.scala

@@ -22,11 +22,13 @@ import java.util.concurrent.ThreadFactory
 import java.util.concurrent.atomic.AtomicInteger
 
 import scala.collection.mutable
+
 import com.gemstone.gemfire.internal.tcp.ConnectionTable
 import com.gemstone.gemfire.{CancelException, SystemFailure}
 import com.google.common.cache.{CacheBuilder, CacheLoader}
 import com.pivotal.gemfirexd.internal.engine.Misc
 import com.pivotal.gemfirexd.internal.engine.distributed.utils.GemFireXDUtils
+
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.serializer.KryoSerializerPool
 import org.apache.spark.sql.internal.ContextJarUtils
@@ -42,7 +44,6 @@ class SnappyExecutor(
     isLocal: Boolean = false)
     extends Executor(executorId, executorHostname, env, userClassPath, isLocal) {
 
-
   {
     // set a thread-factory for the thread pool for cleanup
     val threadGroup: ThreadGroup = Thread.currentThread().getThreadGroup
@@ -76,12 +77,12 @@ class SnappyExecutor(
     Thread.setDefaultUncaughtExceptionHandler(exceptionHandler)
   }
 
-  private val classLoaderCache = {
+  private[this] val classLoaderCache = {
     val loader = new CacheLoader[ClassLoaderKey, ClassLoader]() {
       override def load(key: ClassLoaderKey): ClassLoader = {
         if (key.isReplPath) return mutableLoaderWithRepl(key)
         logInfo(s"Creating ClassLoader for key = $key" +
-          s" with appTime = ${key.appTime} and appName = ${key.appName}")
+            s" with appTime = ${key.appTime} and appName = ${key.appName}")
         val appName = key.appName // appName = "schemaname.functionname"
         val appNameAndJars = key.appNameAndJars
         lazy val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
@@ -134,11 +135,10 @@ class SnappyExecutor(
       private def mutableLoaderWithRepl(key: ClassLoaderKey): ClassLoader = {
         try {
           val klass = Utils.classForName("org.apache.spark.repl.SnappyExecutorClassLoader")
-            .asInstanceOf[Class[_ <: ClassLoader]]
+              .asInstanceOf[Class[_ <: ClassLoader]]
           val constructor = klass.getConstructor(classOf[SparkConf], classOf[SparkEnv],
             classOf[String], classOf[ClassLoader], classOf[Boolean])
-          val cl = constructor.newInstance(conf, env, key.appName, replClassLoader, java.lang.Boolean.TRUE)
-          cl
+          constructor.newInstance(conf, env, key.appName, replClassLoader, java.lang.Boolean.TRUE)
         } catch {
           case _: ClassNotFoundException =>
             logError("Could not find org.apache.spark.repl.ExecutorClassLoader on classpath!")
@@ -148,7 +148,7 @@ class SnappyExecutor(
       }
     }
     // Keeping 500 as cache size. Can revisit the number
-    CacheBuilder.newBuilder().maximumSize(500).build(loader)
+    CacheBuilder.newBuilder().maximumSize(500).build[ClassLoaderKey, ClassLoader](loader)
   }
 
   class ClassLoaderKey(val appName: String,
@@ -170,7 +170,8 @@ class SnappyExecutor(
     }
 
     var userClasspathFirst: Boolean = false
-    def setUserClassPathFirst(userClassPathFirst: Boolean) = {
+
+    def setUserClassPathFirst(userClassPathFirst: Boolean): Unit = {
       this.userClasspathFirst = userClassPathFirst
     }
   }
@@ -270,10 +271,10 @@ class SnappyExecutor(
     Utils.getLocalDir(conf)
   }
 
-  def invalidateReplLoader(replDir: String) = try {
+  def invalidateReplLoader(replDir: String): Unit = try {
     classLoaderCache.invalidate(replDir)
   } catch {
-    case npe: NullPointerException => // ignore
+    case _: NullPointerException => // ignore
   }
 }
 
@@ -305,7 +306,7 @@ private class SnappyUncaughtExceptionHandler(
     val executorBackend: SnappyCoarseGrainedExecutorBackend)
     extends Thread.UncaughtExceptionHandler with Logging {
 
-  override def uncaughtException(thread: Thread, exception: Throwable) {
+  override def uncaughtException(thread: Thread, exception: Throwable): Unit = {
     try {
       // Make it explicit that uncaught exceptions are thrown when container is shutting down.
       // It will help users when they analyze the executor logs

cluster/src/main/scala/org/apache/spark/executor/SnappyExecutor.scala

@@ -793,10 +793,9 @@ class SnappyUnifiedMemoryManager private[memory](
       numBytes: Long,
       memoryMode: MemoryMode): Unit = {
     // if UMM lock is already held, then release inline else enqueue and be done with it
-    if (Thread.holdsLock(this)) synchronized {
-      releaseStorageMemoryForObject_(objectName, numBytes, memoryMode)
-    } else {
-      pendingStorageMemoryReleases.put((objectName, numBytes, memoryMode))
+    if (Thread.holdsLock(this) || !pendingStorageMemoryReleases.offer(
+      (objectName, numBytes, memoryMode), 15, TimeUnit.SECONDS)) {
+      synchronized(releaseStorageMemoryForObject_(objectName, numBytes, memoryMode))
     }
   }
 
@@ -962,13 +961,14 @@ object SnappyUnifiedMemoryManager extends Logging {
           val clazz = Utils.classForName(
             "com.gemstone.gemfire.internal.cache.store.ManagedDirectBufferAllocator")
           clazz.getDeclaredMethod("instance").invoke(null)
+          size
         } catch {
           case NonFatal(e) =>
             logError("Failed to load managed buffer allocator in SnappyData OSS." +
                 s"Temporary scan buffers will be unaccounted DirectByteBuffers: $e")
+            0
         }
-      }
-      size
+      } else 0
     }
     if (memorySize > 0) {
       // set Spark's off-heap properties

cluster/src/main/scala/org/apache/spark/memory/SnappyUnifiedMemoryManager.scala

@@ -615,7 +615,7 @@ class SnappyMemoryAccountingSuite extends MemoryFunSuite {
     val taskContext =
       new TaskContextImpl(0, 0, taskAttemptId = 1, 0, taskMemoryManager, new Properties, null)
     try {
-      CachedDataFrame(taskContext, Seq(unsafeRow).iterator)
+      CachedDataFrame(taskContext, Seq(unsafeRow).iterator, -1L)
       assert(false , "Should not have obtained memory")
     } catch {
       case lme : LowMemoryException => // Success

cluster/src/test/scala/org/apache/spark/memory/SnappyMemoryAccountingSuite.scala

@@ -161,6 +161,17 @@ object Property extends Enumeration {
         "configuration level or as system property (latter on all the servers) and cannot be " +
         "changed at the session level. Default is 0.2", Some(0.2))
 
+  val MaxMemoryResultSize: SparkValue[String] = Val[String](
+    s"${Constant.SPARK_PREFIX}sql.maxMemoryResultSize",
+    "Maximum size of results from a JDBC/ODBC query in a partition that will be held " +
+        "in memory beyond which the results will be written to disk " +
+        "(value in bytes or k/m/g suffixes for unit, min 1k). Default is 4MB.", Some("4m"))
+
+  val ResultPersistenceTimeout: SparkValue[Long] = Val[Long](
+    s"${Constant.SPARK_PREFIX}sql.ResultPersistenceTimeout",
+    s"Maximum duration in seconds for which results larger than ${MaxMemoryResultSize.name}" +
+        s"are held on disk after which they are cleaned up. Default is 3600s (1h).", Some(3600L))
+
   val DisableHashJoin: SQLValue[Boolean] = SQLVal[Boolean](
     s"${Constant.PROPERTY_PREFIX}sql.disableHashJoin",
     "Disable hash joins completely including those for replicated row tables. Default is false.",

core/src/main/scala/io/snappydata/Literals.scala

@@ -97,6 +97,8 @@ object ServiceUtils {
       for ((hiveVar, dirName) <- HiveClientUtil.HIVE_DEFAULT_SETTINGS) {
         sysProps.putIfAbsent(hiveVar.varname, dirName)
       }
+      // increase call stack depth
+      sysProps.putIfAbsent("spark.callstack.depth", "50")
     }
     // set default member-timeout higher for GC pauses (SNAP-1777)
     storeProps.putIfAbsent(DistributionConfig.MEMBER_TIMEOUT_NAME, "30000")