DRILL-2719: ValueVector#getBuffers(clear) must consistently clear vec…

…tors & retain buffers

exec/java-exec/src/main/codegen/templates/NullableValueVectors.java

@@ -79,8 +79,11 @@ public void setCurrentValueCount(int count) {
 
   @Override
   public DrillBuf[] getBuffers(boolean clear) {
-    DrillBuf[] buffers = ObjectArrays.concat(bits.getBuffers(clear), values.getBuffers(clear), DrillBuf.class);
+    DrillBuf[] buffers = ObjectArrays.concat(bits.getBuffers(false), values.getBuffers(false), DrillBuf.class);
     if (clear) {
+      for (DrillBuf buffer:buffers) {
+        buffer.retain();
+      }
       clear();
     }
     return buffers;

exec/java-exec/src/main/codegen/templates/RepeatedValueVectors.java

@@ -285,8 +285,11 @@ public void load(SerializedField metadata, DrillBuf buffer) {
 
   @Override
   public DrillBuf[] getBuffers(boolean clear) {
-    DrillBuf[] buffers = ObjectArrays.concat(offsets.getBuffers(clear), values.getBuffers(clear), DrillBuf.class);
+    DrillBuf[] buffers = ObjectArrays.concat(offsets.getBuffers(false), values.getBuffers(false), DrillBuf.class);
     if (clear) {
+      for (DrillBuf buffer:buffers) {
+        buffer.retain();
+      }
       clear();
     }
     return buffers;

exec/java-exec/src/main/codegen/templates/VariableLengthVectors.java

@@ -148,8 +148,12 @@ public void clear() {
 
   @Override
   public DrillBuf[] getBuffers(boolean clear) {
-    DrillBuf[] buffers = ObjectArrays.concat(offsetVector.getBuffers(clear), super.getBuffers(clear), DrillBuf.class);
+    DrillBuf[] buffers = ObjectArrays.concat(offsetVector.getBuffers(false), super.getBuffers(false), DrillBuf.class);
     if (clear) {
+      // does not make much sense but we have to retain buffers even when clear is set. refactor this interface.
+      for (DrillBuf buffer:buffers) {
+        buffer.retain();
+      }
       clear();
     }
     return buffers;

exec/java-exec/src/main/java/org/apache/drill/exec/vector/BaseDataValueVector.java

@@ -23,7 +23,6 @@
 
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.proto.UserBitShared.SerializedField;
-import org.apache.drill.exec.record.DeadBuf;
 import org.apache.drill.exec.record.MaterializedField;
 
 import com.google.common.collect.Iterators;
@@ -36,22 +35,16 @@ public abstract class BaseDataValueVector extends BaseValueVector{
 
   public BaseDataValueVector(MaterializedField field, BufferAllocator allocator) {
     super(field, allocator);
-
+    this.data = allocator.getEmpty();
   }
 
   /**
    * Release the underlying DrillBuf and reset the ValueVector
    */
   @Override
   public void clear() {
-    if (data == null) {
-      data = DeadBuf.DEAD_BUFFER;
-    }
-    if (data != DeadBuf.DEAD_BUFFER) {
-      data.release();
-      data = data.getAllocator().getEmpty();
-      valueCount = 0;
-    }
+    data.release();
+    data = allocator.getEmpty();
   }
 
   public void setCurrentValueCount(int count) {
@@ -66,7 +59,7 @@ public int getCurrentValueCount() {
   @Override
   public DrillBuf[] getBuffers(boolean clear) {
     DrillBuf[] out;
-    if (valueCount == 0) {
+    if (getBufferSize() == 0) {
       out = new DrillBuf[0];
     } else {
       out = new DrillBuf[]{data};
@@ -82,7 +75,7 @@ public DrillBuf[] getBuffers(boolean clear) {
   }
 
   public int getBufferSize() {
-    if (valueCount == 0) {
+    if (getAccessor().getValueCount() == 0) {
       return 0;
     }
     return data.writerIndex();

exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/AbstractMapVector.java

@@ -216,9 +216,15 @@ public VectorWithOrdinal getChildVectorWithOrdinal(String name) {
   public DrillBuf[] getBuffers(boolean clear) {
     List<DrillBuf> buffers = Lists.newArrayList();
 
-    for (ValueVector v : vectors.values()) {
-      for (DrillBuf buf : v.getBuffers(clear)) {
+    for (ValueVector vector : vectors.values()) {
+      for (DrillBuf buf : vector.getBuffers(false)) {
         buffers.add(buf);
+        if (clear) {
+          buf.retain();
+        }
+      }
+      if (clear) {
+        vector.clear();
       }
     }
 

exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/RepeatedListVector.java

@@ -345,7 +345,15 @@ public RepeatedListAccessor getAccessor() {
 
   @Override
   public DrillBuf[] getBuffers(boolean clear) {
-    return ArrayUtils.addAll(offsets.getBuffers(clear), vector.getBuffers(clear));
+    DrillBuf[] buffers = ArrayUtils.addAll(offsets.getBuffers(false), vector.getBuffers(false));
+    if (clear) {
+      // does not make much sense but we have to retain buffers even when clear is set. refactor this interface.
+      for (DrillBuf buffer:buffers) {
+        buffer.retain();
+      }
+      clear();
+    }
+    return buffers;
   }
 
   protected void setVector(ValueVector newVector) {

exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/TestEmptyPopulator.java

@@ -43,11 +43,13 @@ public class TestEmptyPopulator extends ExecTest {
   private EmptyValuePopulator populator;
 
   private final ByteBuffer buffer = ByteBuffer.allocateDirect(BUF_SIZE);
+  private final ByteBuffer empty = ByteBuffer.allocateDirect(0);
 
 
   @Before
   public void initialize() {
     Mockito.when(allocator.buffer(Mockito.anyInt())).thenReturn(DrillBuf.wrapByteBuffer(buffer));
+    Mockito.when(allocator.getEmpty()).thenReturn(DrillBuf.wrapByteBuffer(empty));
     offsets = new UInt4Vector(null, allocator);
     offsets.allocateNewSafe();
     accessor = offsets.getAccessor();