From 4c3f77347fd3116c5becaaf1ddf44974105cdd6d Mon Sep 17 00:00:00 2001
From: "hongze.zhz" <hongze.zhz@alibaba-inc.com>
Date: Fri, 18 Nov 2016 20:11:38 +0800
Subject: [PATCH] DRILL-5051: Fix incorrect computation of 'fetch' in
 LimitRecordBatch when 'offset' is specified

---
 .../physical/impl/limit/LimitRecordBatch.java | 35 ++++---------------
 .../java/org/apache/drill/TestBugFixes.java   | 10 ++++++
 2 files changed, 17 insertions(+), 28 deletions(-)

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
index 08ffc0b7787..254a297ef0e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
@@ -139,18 +139,13 @@ protected IterOutcome doWork() {
       skipBatch = true;
     } else {
       outgoingSv.allocateNew(recordCount);
-      if(incomingSv != null) {
-       limitWithSV(recordCount);
-      } else {
-       limitWithNoSV(recordCount);
-      }
+      limit(recordCount);
     }
 
     return IterOutcome.OK;
   }
 
-  // These two functions are identical except for the computation of the index; merge
-  private void limitWithNoSV(int recordCount) {
+  private void limit(int recordCount) {
     final int offset = Math.max(0, Math.min(recordCount - 1, recordsToSkip));
     recordsToSkip -= offset;
     int fetch;
@@ -164,27 +159,11 @@ private void limitWithNoSV(int recordCount) {
 
     int svIndex = 0;
     for(int i = offset; i < fetch; svIndex++, i++) {
-      outgoingSv.setIndex(svIndex, (char) i);
-    }
-    outgoingSv.setRecordCount(svIndex);
-  }
-
-  private void limitWithSV(int recordCount) {
-    final int offset = Math.max(0, Math.min(recordCount - 1, recordsToSkip));
-    recordsToSkip -= offset;
-    int fetch;
-
-    if(noEndLimit) {
-      fetch = recordCount;
-    } else {
-      fetch = Math.min(recordCount, recordsLeft);
-      recordsLeft -= Math.max(0, fetch - offset);
-    }
-
-    int svIndex = 0;
-    for(int i = offset; i < fetch; svIndex++, i++) {
-      final char index = incomingSv.getIndex(i);
-      outgoingSv.setIndex(svIndex, index);
+      if (incomingSv != null) {
+        outgoingSv.setIndex(svIndex, incomingSv.getIndex(i));
+      } else {
+        outgoingSv.setIndex(svIndex, (char) i);
+      }
     }
     outgoingSv.setRecordCount(svIndex);
   }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java b/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
index 03b1b61b8bd..a9fc5d0ae5c 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
@@ -221,4 +221,14 @@ public void testDRILL4884() throws Exception {
             .baselineRecords(baseline)
             .go();
   }
+
+  @Test
+  public void testDRILL5051() throws Exception {
+    testBuilder()
+        .sqlQuery("select count(1) as cnt from (select l_orderkey from (select l_orderkey from cp.`tpch/lineitem.parquet` limit 2) limit 1 offset 1)")
+        .unOrdered()
+        .baselineColumns("cnt")
+        .baselineValues(1L)
+        .go();
+  }
 }