From c18ee19c128f2917ca4b5040e3db52e939cc23b5 Mon Sep 17 00:00:00 2001
From: Meisam Fathi Salmi <meisam.fathi@gmail.com>
Date: Thu, 23 Oct 2014 14:54:58 -0400
Subject: [PATCH] Extracting a new function for refilling currentChunk.

This change make it possible to override guaranteeFill in subclasses

#14: Refactor GpuFilteredRDD and FilteredChunkItrator

Task-Url: http://github.com/meisam/spark/issues/issue/14
---
 core/src/main/scala/org/apache/spark/rdd/GpuRDD.scala | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/rdd/GpuRDD.scala b/core/src/main/scala/org/apache/spark/rdd/GpuRDD.scala
index 200a99c4cdcf1..44fce8bb91583 100644
--- a/core/src/main/scala/org/apache/spark/rdd/GpuRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/GpuRDD.scala
@@ -172,12 +172,16 @@ class ChunkIterator[T <: Product : ClassTag]
   protected val currentChunk: RDDChunk[T] = new RDDChunk[T](columnTypes, chunkCapacity)
 
   override def next(): T = {
+    guaranteeFill
+    val t: T = currentChunk(currentPosition)
+    currentPosition += 1
+    t
+  }
+
+  def guaranteeFill {
     if (currentPosition >= currentChunk.size || currentPosition < 0) {
       currentChunk.fill(itr)
       currentPosition = 0
     }
-    val t: T = currentChunk(currentPosition)
-    currentPosition += 1
-    t
   }
 }
\ No newline at end of file