DRILL-4858: REPEATED_COUNT on an array of maps and an array of arrays is not implemented

- Implemented 'repeated_count' function for repeated MAP and repeated LIST;
- Updated RepeatedListReader and RepeatedMapReader implementations to return correct value from size() method
- Moved repeated_count to freemarker template and added support for more repeated types for the function
closes #1641

9 files changed, 412 insertions, 204 deletions

diff --git a/exec/java-exec/src/main/codegen/templates/RepeatedCountFunctions.java b/exec/java-exec/src/main/codegen/templates/RepeatedCountFunctions.java
new file mode 100644
index 000000000..85c7b9096
--- /dev/null
+++ b/exec/java-exec/src/main/codegen/templates/RepeatedCountFunctions.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+<@pp.dropOutputFile />
+
+<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/RepeatedCountFunctions.java" />
+
+<#include "/@includes/license.ftl" />
+package org.apache.drill.exec.expr.fn.impl;
+
+import org.apache.drill.exec.expr.DrillSimpleFunc;
+import org.apache.drill.exec.expr.annotations.FunctionTemplate;
+import org.apache.drill.exec.expr.annotations.Output;
+import org.apache.drill.exec.expr.annotations.Param;
+import org.apache.drill.exec.expr.annotations.Workspace;
+import org.apache.drill.exec.expr.holders.*;
+
+/*
+ * This class is generated using freemarker and the ${.template_name} template.
+ */
+public class RepeatedCountFunctions {
+
+  private RepeatedCountFunctions() {
+  }
+
+  <#list countAggrTypes.countFunctionsInput as type>
+    <#if type?starts_with("Repeated")>
+  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
+  public static class RepeatedCount${type} implements DrillSimpleFunc {
+
+    @Param
+    ${type}Holder input;
+    @Output
+    IntHolder out;
+
+    @Override
+    public void setup() {
+    }
+
+    @Override
+    public void eval() {
+      out.value = input.reader.size();
+    }
+  }
+
+    </#if>
+  </#list>
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
index e13a9adb5..6f797480f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
@@ -479,6 +479,14 @@ public class EvaluationVisitor {
 
       if (!hasReadPath && !complex) {
         JBlock eval = new JBlock();
+
+        if (repeated) {
+          JExpression expr = vv1.invoke("getReader");
+          // Set correct position to the reader
+          eval.add(expr.invoke("reset"));
+          eval.add(expr.invoke("setPosition").arg(recordIndex));
+        }
+
         GetSetVectorHelper.read(e.getMajorType(),  vv1, eval, out, generator.getModel(), recordIndex);
         generator.getEvalBlock().add(eval);
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/SimpleRepeatedFunctions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/SimpleRepeatedFunctions.java
index 9de39288d..ea84d455f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/SimpleRepeatedFunctions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/SimpleRepeatedFunctions.java
@@ -42,153 +42,6 @@ public class SimpleRepeatedFunctions {
   private SimpleRepeatedFunctions() {
   }
 
-  // TODO - replace with a freemarker template and fill out the rest of the types
-  // focused on getting functions defined for JSON types as this is the primary format
-  // users are extracting repeated data out of currently
-
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthBigInt implements DrillSimpleFunc {
-
-    @Param
-    RepeatedBigIntHolder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthTinyInt implements DrillSimpleFunc {
-
-    @Param
-    RepeatedTinyIntHolder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
-  /*
-  // TODO - need to confirm that these work   SMP: They do not
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthMap implements DrillSimpleFunc {
-
-    @Param
-    RepeatedMapHolder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
-  // TODO - need to confirm that these work   SMP: They do not
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthList implements DrillSimpleFunc {
-
-    @Param
-    RepeatedListHolder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-  */
-
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthBit implements DrillSimpleFunc {
-
-    @Param
-    RepeatedBitHolder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthFloat4 implements DrillSimpleFunc {
-
-    @Param
-    RepeatedFloat4Holder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthFloat8 implements DrillSimpleFunc {
-
-    @Param
-    RepeatedFloat8Holder input;
-    @Output
-    IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthInt implements DrillSimpleFunc {
-
-    @Param RepeatedIntHolder input;
-    @Output IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
-  @FunctionTemplate(name = "repeated_count", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class RepeatedLengthVarChar implements DrillSimpleFunc {
-
-    @Param RepeatedVarCharHolder input;
-    @Output IntHolder out;
-
-    public void setup() {
-    }
-
-    public void eval() {
-      out.value = input.end - input.start;
-    }
-  }
-
   @FunctionTemplate(name = "repeated_contains", scope = FunctionTemplate.FunctionScope.SIMPLE)
   public static class ContainsBigInt implements DrillSimpleFunc {
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewSimpleRepeatedFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewSimpleRepeatedFunctions.java
index 552346da0..2181385a4 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewSimpleRepeatedFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewSimpleRepeatedFunctions.java
@@ -17,36 +17,309 @@
  */
 package org.apache.drill.exec.fn.impl;
 
-import org.apache.drill.test.BaseTestQuery;
 import org.apache.drill.categories.SqlFunctionTest;
+import org.apache.drill.test.ClusterFixture;
+import org.apache.drill.test.ClusterTest;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
+import java.nio.file.Paths;
+
 @Category(SqlFunctionTest.class)
-public class TestNewSimpleRepeatedFunctions extends BaseTestQuery {
+public class TestNewSimpleRepeatedFunctions extends ClusterTest {
+
+  private static final String REPEATED_TYPES_PARQUET_TABLE = "cp.`store/parquet/complex/repeated_types.parquet`";
+  private static final String REPEATED_TYPES_JSON_TABLE = "cp.`parquet/alltypes_repeated.json`";
+  private static final String SELECT_REPEATED_COUNT_LIST = "select repeated_count(array) from dfs.`functions/repeated/repeated_list.json`";
+  private static final String SELECT_REPEATED_COUNT_MAP = "select repeated_count(mapArray) from dfs.`functions/repeated/repeated_map.json`";
+  private static final String SELECT_REPEATED_COUNT_QUERY = "select repeated_count(%s) from %s";
+  private static final String COLUMN_NAME = "EXPR$0";
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    dirTestWatcher.copyResourceToRoot(Paths.get("functions", "repeated"));
+    startCluster(ClusterFixture.builder(dirTestWatcher));
+  }
+
   @Test
   public void testRepeatedContainsForWildCards() throws Exception {
-     testBuilder().
-       sqlQuery("select repeated_contains(topping, 'Choc*') from cp.`testRepeatedWrite.json`")
-       .ordered()
-       .baselineColumns("EXPR$0")
-       .baselineValues(true)
-       .baselineValues(true)
-       .baselineValues(true)
-       .baselineValues(true)
-       .baselineValues(false)
-       .build().run();
-
-     testBuilder().
-       sqlQuery("select repeated_contains(topping, 'Pow*') from cp.`testRepeatedWrite.json`")
-       .ordered()
-       .baselineColumns("EXPR$0")
-       .baselineValues(true)
-       .baselineValues(false)
-       .baselineValues(false)
-       .baselineValues(true)
-       .baselineValues(false)
-       .build().run();
-
-  }
-}
\ No newline at end of file
+    String query = "select repeated_contains(topping, '%s*') from cp.`testRepeatedWrite.json`";
+    testBuilder()
+        .sqlQuery(query, "Choc")
+        .unOrdered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(true, true, true, true, false)
+        .go();
+
+    testBuilder()
+        .sqlQuery(query, "Pow")
+        .unOrdered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(true, false, false, true, false)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountVarCharJSON() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "str_list", "cp.`store/json/json_basic_repeated_varchar.json`")
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(5, 1, 3, 1)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountIntJSON() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int_col", REPEATED_TYPES_JSON_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(12, 4, 4, 4)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedFloatJSON() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "float4_col", REPEATED_TYPES_JSON_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(7, 4, 4, 4)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedBitJSON() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "bit_col", REPEATED_TYPES_JSON_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(7, 7, 5, 3)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountDate() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "date_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(3, 1, 0)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountTime() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "time_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(1, 4, 3)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountTimestamp() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "timestamp_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(0, 2, 5)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountInterval() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "interval_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(2, 5, 0)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountVarChar() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "string_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(0, 3, 10)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountInt() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int8_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(3, 1, 3)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountInt_2() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int16_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(3, 2, 4)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountInt_3() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int32_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(1, 9, 3)
+        .go();
+  }
+  @Test
+  public void testRepeatedCountUInt8() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "uint8_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(1, 7, 1)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountUInt16() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "uint16_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(0, 3, 1)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountUInt32() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "uint32_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(4, 1, 4)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountBigInt() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int64_raw_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(4, 1, 2)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountBigInt_2() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int64_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(9, 2, 4)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountBigInt_3() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "uint64_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(0, 1, 3)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountVarDecimal() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "decimal_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(7, 3, 6)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountVarBinary() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_QUERY, "int96_raw_list", REPEATED_TYPES_PARQUET_TABLE)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(0, 1, 9)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountRepeatedMap() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_MAP)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(2, 2, 3, 0, 1, 5, 2)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountRepeatedMapInWhere() throws Exception {
+    String query = SELECT_REPEATED_COUNT_MAP + " where repeated_count(mapArray) > 2";
+    testBuilder()
+        .sqlQuery(query)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(3, 5)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountRepeatedMapInHaving() throws Exception {
+    String query = SELECT_REPEATED_COUNT_MAP + " group by 1 having repeated_count(mapArray) < 3";
+    testBuilder()
+        .sqlQuery(query)
+        .unOrdered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(2, 0, 1)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountRepeatedList() throws Exception {
+    testBuilder()
+        .sqlQuery(SELECT_REPEATED_COUNT_LIST)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(3, 0, 2, 5, 9, 4, 0, 3)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountRepeatedListInWhere() throws Exception {
+    String query = SELECT_REPEATED_COUNT_LIST + " where repeated_count(array) > 4";
+    testBuilder()
+        .sqlQuery(query)
+        .ordered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(5, 9)
+        .go();
+  }
+
+  @Test
+  public void testRepeatedCountRepeatedListInHaving() throws Exception {
+    String query = SELECT_REPEATED_COUNT_LIST + " group by 1 having repeated_count(array) < 4";
+    testBuilder()
+        .sqlQuery(query)
+        .unOrdered()
+        .baselineColumns(COLUMN_NAME)
+        .baselineValuesForSingleColumn(3, 0, 2)
+        .go();
+  }
+}
diff --git a/exec/java-exec/src/test/resources/functions/repeated/repeated_list.json b/exec/java-exec/src/test/resources/functions/repeated/repeated_list.json
new file mode 100644
index 000000000..cc93832b7
--- /dev/null
+++ b/exec/java-exec/src/test/resources/functions/repeated/repeated_list.json
@@ -0,0 +1,8 @@
+{"id": 1, "array": [[1, 2], [1, 3], [2, 3]]}
+{"id": 2, "array": []}
+{"id": 3, "array": [[2, 3], [1, 3, 4]]}
+{"id": 4, "array": [[1], [2], [3, 4], [5], [6]]}
+{"id": 5, "array": [[1, 2, 3], [4, 5], [6], [7], [8, 9], [2, 3], [2, 3], [2, 3], [2]]}
+{"id": 6, "array": [[1, 2], [3], [4], [5]]}
+{"id": 7, "array": []}
+{"id": 8, "array": [[1], [2], [3]]}
diff --git a/exec/java-exec/src/test/resources/functions/repeated/repeated_map.json b/exec/java-exec/src/test/resources/functions/repeated/repeated_map.json
new file mode 100644
index 000000000..5e9127b87
--- /dev/null
+++ b/exec/java-exec/src/test/resources/functions/repeated/repeated_map.json
@@ -0,0 +1,7 @@
+{"mapArray": [{"field1": 1, "field2": "val1"}, {"field1": 2}]}
+{"mapArray": [{"field1": 1}, {"field1": 2}]}
+{"mapArray": [{"field2": "val2"}, {"field1": 2}, {"field1": 2}]}
+{"mapArray": []}
+{"mapArray": [{"field1": 1, "field2": "val3"}]}
+{"mapArray": [{"field1": 1, "field2": "val4"}, {"field1": 2}, {"field1": 2}, {"field2": "val1"}, {"field1": 2}]}
+{"mapArray": [{"field1": 1, "field2": "val3"}, {"field1": 2}]}
diff --git a/exec/java-exec/src/test/resources/store/parquet/complex/repeated_types.parquet b/exec/java-exec/src/test/resources/store/parquet/complex/repeated_types.parquet
new file mode 100644
index 000000000..e17b14123
--- /dev/null
+++ b/exec/java-exec/src/test/resources/store/parquet/complex/repeated_types.parquet
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedListReaderImpl.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedListReaderImpl.java
index eb2ce20cb..a98132a66 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedListReaderImpl.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedListReaderImpl.java
@@ -34,6 +34,8 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
   private final String name;
   private final RepeatedListVector container;
   private FieldReader reader;
+  private int currentOffset;
+  private int maxOffset;
 
   public RepeatedListReaderImpl(String name, RepeatedListVector container) {
     super();
@@ -48,7 +50,7 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
 
   @Override
   public void copyAsValue(ListWriter writer) {
-    if (currentOffset == NO_VALUES) {
+    if (isEmpty()) {
       return;
     }
     RepeatedListWriter impl = (RepeatedListWriter) writer;
@@ -57,16 +59,13 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
 
   @Override
   public void copyAsField(String name, MapWriter writer) {
-    if (currentOffset == NO_VALUES) {
+    if (isEmpty()) {
       return;
     }
     RepeatedListWriter impl = (RepeatedListWriter) writer.list(name);
     impl.container.copyFromSafe(idx(), impl.idx(), container);
   }
 
-  private int currentOffset;
-  private int maxOffset;
-
   @Override
   public void reset() {
     super.reset();
@@ -80,7 +79,7 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
 
   @Override
   public int size() {
-    return maxOffset - currentOffset;
+    return isEmpty() ? 0 : maxOffset - currentOffset;
   }
 
   @Override
@@ -96,9 +95,9 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
     if (h.start == h.end) {
       currentOffset = NO_VALUES;
     } else {
-      currentOffset = h.start-1;
-      maxOffset = h.end;
-      if(reader != null) {
+      currentOffset = h.start - 1;
+      maxOffset = h.end - 1;
+      if (reader != null) {
         reader.setPosition(currentOffset);
       }
     }
@@ -106,7 +105,7 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
 
   @Override
   public boolean next() {
-    if (currentOffset +1 < maxOffset) {
+    if (currentOffset < maxOffset) {
       currentOffset++;
       if (reader != null) {
         reader.setPosition(currentOffset);
@@ -137,8 +136,11 @@ public class RepeatedListReaderImpl extends AbstractFieldReader{
     return reader;
   }
 
+  public boolean isEmpty() {
+    return currentOffset == NO_VALUES;
+  }
+
   public boolean isSet() {
     return true;
   }
-
 }
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedMapReaderImpl.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedMapReaderImpl.java
index 156c6b8e8..58cb4f32b 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedMapReaderImpl.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/RepeatedMapReaderImpl.java
@@ -34,17 +34,13 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
 
   private final RepeatedMapVector vector;
   private final Map<String, FieldReader> fields = Maps.newHashMap();
+  private int currentOffset;
+  private int maxOffset;
 
   public RepeatedMapReaderImpl(RepeatedMapVector vector) {
     this.vector = vector;
   }
 
-  private void setChildrenPosition(int index) {
-    for (FieldReader r : fields.values()) {
-      r.setPosition(index);
-    }
-  }
-
   @Override
   public FieldReader reader(String name) {
     FieldReader reader = fields.get(name);
@@ -63,7 +59,7 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
 
   @Override
   public FieldReader reader() {
-    if (currentOffset == NO_VALUES) {
+    if (isNull()) {
       return NullReader.INSTANCE;
     }
 
@@ -71,9 +67,6 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
     return new SingleLikeRepeatedMapReaderImpl(vector, this);
   }
 
-  private int currentOffset;
-  private int maxOffset;
-
   @Override
   public void reset() {
     super.reset();
@@ -87,10 +80,7 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
 
   @Override
   public int size() {
-    if (isNull()) {
-      return 0;
-    }
-    return maxOffset - (currentOffset < 0 ? 0 : currentOffset);
+    return isNull() ? 0 : maxOffset - currentOffset;
   }
 
   @Override
@@ -106,8 +96,8 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
     if (h.start == h.end) {
       currentOffset = NO_VALUES;
     } else {
-      currentOffset = h.start-1;
-      maxOffset = h.end;
+      currentOffset = h.start - 1;
+      maxOffset = h.end - 1;
       setChildrenPosition(currentOffset);
     }
   }
@@ -129,7 +119,7 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
 
   @Override
   public boolean next() {
-    if (currentOffset +1 < maxOffset) {
+    if (currentOffset < maxOffset) {
       setChildrenPosition(++currentOffset);
       return true;
     } else {
@@ -164,7 +154,7 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
 
   @Override
   public void copyAsValue(MapWriter writer) {
-    if (currentOffset == NO_VALUES) {
+    if (isNull()) {
       return;
     }
     RepeatedMapWriter impl = (RepeatedMapWriter) writer;
@@ -172,7 +162,7 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
   }
 
   public void copyAsValueSingle(MapWriter writer) {
-    if (currentOffset == NO_VALUES) {
+    if (isNull()) {
       return;
     }
     SingleMapWriter impl = (SingleMapWriter) writer;
@@ -181,11 +171,16 @@ public class RepeatedMapReaderImpl extends AbstractFieldReader{
 
   @Override
   public void copyAsField(String name, MapWriter writer) {
-    if (currentOffset == NO_VALUES) {
+    if (isNull()) {
       return;
     }
     RepeatedMapWriter impl = (RepeatedMapWriter) writer.map(name);
     impl.container.copyFromSafe(idx(), impl.idx(), vector);
   }
 
+  private void setChildrenPosition(int index) {
+    for (FieldReader r : fields.values()) {
+      r.setPosition(index);
+    }
+  }
 }