From d44b889ffebbb58939110776795489c052aac786 Mon Sep 17 00:00:00 2001
From: Amit Hadke <amit.hadke@gmail.com>
Date: Tue, 24 Nov 2015 16:00:16 -0800
Subject: [PATCH] DRILL-4109 Fix NPE in RecordIterator.

---
 .../apache/drill/exec/record/RecordIterator.java | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java
index faa4d830d16..7b5ec288976 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java
@@ -93,7 +93,7 @@ public void mark() {
     // Release all batches before current batch. [0 to startBatchPosition).
     final Map<Range<Long>,RecordBatchData> oldBatches = batches.subRangeMap(Range.closedOpen(0l, startBatchPosition)).asMapOfRanges();
     for (Range<Long> range : oldBatches.keySet()) {
-      oldBatches.get(range.lowerEndpoint()).clear();
+      oldBatches.get(range).clear();
     }
     batches.remove(Range.closedOpen(0l, startBatchPosition));
     markedInnerPosition = innerPosition;
@@ -113,8 +113,9 @@ public void reset() {
       }
       innerPosition = markedInnerPosition;
       outerPosition = markedOuterPosition;
-      startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
-      innerRecordCount = (int)(batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition);
+      final Range<Long> markedBatchRange = batches.getEntry(outerPosition).getKey();
+      startBatchPosition = markedBatchRange.lowerEndpoint();
+      innerRecordCount = (int)(markedBatchRange.upperEndpoint() - startBatchPosition);
       markedInnerPosition = -1;
       markedOuterPosition = -1;
     }
@@ -133,9 +134,10 @@ public void forward(long delta) {
     // Get vectors from new position.
     container.transferIn(rbdNew.getContainer());
     outerPosition = nextOuterPosition;
-    startBatchPosition = batches.getEntry(outerPosition).getKey().lowerEndpoint();
+    final Range<Long> markedBatchRange = batches.getEntry(outerPosition).getKey();
+    startBatchPosition = markedBatchRange.lowerEndpoint();
     innerPosition = (int)(outerPosition - startBatchPosition);
-    innerRecordCount = (int)(batches.getEntry(outerPosition).getKey().upperEndpoint() - startBatchPosition);
+    innerRecordCount = (int)(markedBatchRange.upperEndpoint() - startBatchPosition);
   }
 
   /**
@@ -239,7 +241,9 @@ public long getOuterPosition() {
 
   public int getCurrentPosition() {
     Preconditions.checkArgument(initialized);
-    Preconditions.checkArgument(innerPosition >= 0 && innerPosition < innerRecordCount);
+    Preconditions.checkArgument(innerPosition >= 0 && innerPosition < innerRecordCount,
+      String.format("innerPosition:%d, outerPosition:%d, innerRecordCount:%d, totalRecordCount:%d",
+        innerPosition, outerPosition, innerRecordCount, totalRecordCount));
     return innerPosition;
   }