[SNAP-2366] row buffer fault-in, forced rollover, merge small batches

cluster/src/dunit/scala/io/snappydata/cluster/SplitSnappyClusterDUnitTest.scala

@@ -261,13 +261,13 @@ class SplitSnappyClusterDUnitTest(s: String)
     // added in SNAP-2012
     StoreUtils.TEST_RANDOM_BUCKETID_ASSIGNMENT = true
     try {
-      ColumnUpdateDeleteTests.testBasicUpdate(session)
-      ColumnUpdateDeleteTests.testDeltaStats(session)
-      ColumnUpdateDeleteTests.testBasicDelete(session)
-      ColumnUpdateDeleteTests.testSNAP1925(session)
-      ColumnUpdateDeleteTests.testSNAP1926(session)
-      ColumnUpdateDeleteTests.testConcurrentOps(session)
-      ColumnUpdateDeleteTests.testSNAP2124(session)
+      ColumnUpdateDeleteTests.testBasicUpdate(session, redundancy = 1)
+      ColumnUpdateDeleteTests.testDeltaStats(session, redundancy = 1)
+      ColumnUpdateDeleteTests.testBasicDelete(session, redundancy = 1)
+      ColumnUpdateDeleteTests.testSNAP1925(session, redundancy = 1)
+      ColumnUpdateDeleteTests.testSNAP1926(session, redundancy = 1)
+      ColumnUpdateDeleteTests.testConcurrentOps(session, redundancy = 1)
+      ColumnUpdateDeleteTests.testSNAP2124(session, redundancy = 1)
     } finally {
       StoreUtils.TEST_RANDOM_BUCKETID_ASSIGNMENT = false
     }

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

@@ -36,19 +36,19 @@ import org.apache.spark.sql.execution.columnar.impl.ColumnFormatRelation
 class SnappyStorageEvictor extends Logging {
 
   private def getAllRegionList(offHeap: Boolean,
-      hasOffHeap: Boolean): ArrayBuffer[LocalRegion] = {
+      hasOffHeap: Boolean, onlyRowBuffers: Boolean): ArrayBuffer[LocalRegion] = {
     val cache = GemFireCacheImpl.getExisting
     val allRegionList = new ArrayBuffer[LocalRegion]()
     val irm: InternalResourceManager = cache.getResourceManager
     for (listener <- irm.getResourceListeners(
       SnappyStorageEvictor.resourceType).asScala) listener match {
       case pr: PartitionedRegion =>
-        if (includePartitionedRegion(pr, offHeap, hasOffHeap)) {
+        if (includePartitionedRegion(pr, offHeap, hasOffHeap, onlyRowBuffers)) {
           allRegionList ++= pr.getDataStore.getAllLocalBucketRegions.asScala
         }
       // no off-heap local regions yet in SnappyData
       case lr: LocalRegion =>
-        if (!offHeap && includeLocalRegion(lr)) {
+        if (!offHeap && !onlyRowBuffers && includeLocalRegion(lr)) {
           allRegionList += lr
         }
       case _ =>
@@ -64,7 +64,8 @@ class SnappyStorageEvictor extends Logging {
   }
 
   @throws(classOf[Exception])
-  def evictRegionData(bytesRequired: Long, offHeap: Boolean): Long = {
+  def evictRegionData(bytesRequired: Long, offHeap: Boolean,
+      onlyRowBuffers: Boolean = false): Long = {
     val cache = GemFireCacheImpl.getInstance()
     if (cache eq null) return 0L
 
@@ -76,7 +77,7 @@ class SnappyStorageEvictor extends Logging {
     val stats = cache.getCachePerfStats
     stats.incEvictorJobsStarted()
     var totalBytesEvicted: Long = 0
-    val regionSet = Random.shuffle(getAllRegionList(offHeap, hasOffHeap))
+    val regionSet = Random.shuffle(getAllRegionList(offHeap, hasOffHeap, onlyRowBuffers))
     val start = CachePerfStats.getStatTime
     try {
       while (regionSet.nonEmpty) {
@@ -119,19 +120,26 @@ class SnappyStorageEvictor extends Logging {
       }
       stats.incEvictorJobsCompleted()
     }
-    totalBytesEvicted
+    // evict row-buffers as the last resort
+    if (!onlyRowBuffers && !offHeap && totalBytesEvicted < bytesRequired) {
+      totalBytesEvicted + evictRegionData(bytesRequired - totalBytesEvicted,
+        offHeap, onlyRowBuffers = true)
+    } else totalBytesEvicted
   }
 
   protected def includePartitionedRegion(region: PartitionedRegion,
-      offHeap: Boolean, hasOffHeap: Boolean): Boolean = {
-    val hasLRU = (region.getEvictionAttributes.getAlgorithm.isLRUHeap
-      && (region.getDataStore != null)
-      && !region.getAttributes.getEnableOffHeapMemory && !region.isRowBuffer())
+      offHeap: Boolean, hasOffHeap: Boolean, onlyRowBuffers: Boolean): Boolean = {
+    var hasLRU = (region.getEvictionAttributes.getAlgorithm.isLRUHeap
+        && (region.getDataStore != null)
+        && !region.getAttributes.getEnableOffHeapMemory)
+    val isRowBuffer = region.isRowBuffer
+    if (onlyRowBuffers) return hasLRU && isRowBuffer
+
+    hasLRU &&= !isRowBuffer
     if (hasOffHeap) {
       // when off-heap is enabled then all column tables use off-heap
-      val regionPath = Misc.getFullTableNameFromRegionPath(region.getFullPath)
-      if (offHeap) hasLRU && ColumnFormatRelation.isColumnTable(regionPath)
-      else hasLRU && !ColumnFormatRelation.isColumnTable(regionPath)
+      hasLRU && offHeap == ColumnFormatRelation.isColumnTable(
+        Misc.getFullTableNameFromRegionPath(region.getFullPath))
     } else {
       assert(!offHeap,
         "unexpected invocation for hasOffHeap=false and offHeap=true")
@@ -141,7 +149,7 @@ class SnappyStorageEvictor extends Logging {
 
   protected def includeLocalRegion(region: LocalRegion): Boolean = {
     (region.getEvictionAttributes.getAlgorithm.isLRUHeap
-      && !region.getAttributes.getEnableOffHeapMemory)
+        && !region.getAttributes.getEnableOffHeapMemory)
   }
 }
 

cluster/src/test/scala/io/snappydata/QueryTest.scala

@@ -97,11 +97,19 @@ class QueryTest extends SnappyFunSuite {
 
     val query = "select k, v from t1 inner join t2 where t1.id = t2.k order by k, v"
     val df = session.sql(query)
-    val result1 = df.collect().mkString(" ")
+    var result1 = df.collect().mkString(" ")
     val result2 = spark.sql(query).collect().mkString(" ")
     if (result1 != result2) {
       fail(s"Expected result: $result2\nGot: $result1")
     }
+
+    // force run stats so that small batches have been merged repeatedly
+    SnappyEmbeddedTableStatsProviderService.publishColumnTableRowCountStats()
+    Thread.sleep(10000)
+    result1 = df.collect().mkString(" ")
+    if (result1 != result2) {
+      fail(s"Expected result: $result2\nGot: $result1")
+    }
   }
 
   /**

core/src/dunit/scala/io/snappydata/cluster/SplitClusterDUnitTestBase.scala

@@ -209,13 +209,13 @@ trait SplitClusterDUnitTestBase extends Logging {
         // using random bucket assignment for cases like SNAP-2175
         StoreUtils.TEST_RANDOM_BUCKETID_ASSIGNMENT = true
         try {
-          ColumnUpdateDeleteTests.testBasicUpdate(session)
-          ColumnUpdateDeleteTests.testDeltaStats(session)
-          ColumnUpdateDeleteTests.testBasicDelete(session)
-          ColumnUpdateDeleteTests.testSNAP1925(session)
-          ColumnUpdateDeleteTests.testSNAP1926(session)
-          ColumnUpdateDeleteTests.testConcurrentOps(session)
-          ColumnUpdateDeleteTests.testSNAP2124(session)
+          ColumnUpdateDeleteTests.testBasicUpdate(session, redundancy = 1)
+          ColumnUpdateDeleteTests.testDeltaStats(session, redundancy = 1)
+          ColumnUpdateDeleteTests.testBasicDelete(session, redundancy = 1)
+          ColumnUpdateDeleteTests.testSNAP1925(session, redundancy = 1)
+          ColumnUpdateDeleteTests.testSNAP1926(session, redundancy = 1)
+          ColumnUpdateDeleteTests.testConcurrentOps(session, redundancy = 1)
+          ColumnUpdateDeleteTests.testSNAP2124(session, redundancy = 1)
         } finally {
           StoreUtils.TEST_RANDOM_BUCKETID_ASSIGNMENT = false
         }