Remove null vector from ArrayBlock, MapBlock, and RowBlock when all v…

…alues are not null

This reverts commit d7256c5.

presto-hive/src/main/java/com/facebook/presto/hive/HivePageSource.java

@@ -514,7 +514,7 @@ public Block apply(Block block)
                 valueIsNull[i] = arrayBlock.isNull(i);
                 offsets[i + 1] = offsets[i] + arrayBlock.getLength(i);
             }
-            return ArrayBlock.fromElementBlock(arrayBlock.getPositionCount(), valueIsNull, offsets, elementsBlock);
+            return ArrayBlock.fromElementBlock(arrayBlock.getPositionCount(), Optional.of(valueIsNull), offsets, elementsBlock);
         }
     }
 
@@ -550,7 +550,7 @@ public Block apply(Block block)
                 valueIsNull[i] = mapBlock.isNull(i);
                 offsets[i + 1] = offsets[i] + mapBlock.getEntryCount(i);
             }
-            return ((MapType) toType).createBlockFromKeyValue(valueIsNull, offsets, keysBlock, valuesBlock);
+            return ((MapType) toType).createBlockFromKeyValue(Optional.of(valueIsNull), offsets, keysBlock, valuesBlock);
         }
     }
 
@@ -600,7 +600,7 @@ else if (i < rowBlock.getFieldCount()) {
             for (int i = 0; i < rowBlock.getPositionCount(); i++) {
                 valueIsNull[i] = rowBlock.isNull(i);
             }
-            return RowBlock.fromFieldBlocks(valueIsNull, fields);
+            return RowBlock.fromFieldBlocks(valueIsNull.length, Optional.of(valueIsNull), fields);
         }
     }
 

presto-hive/src/main/java/com/facebook/presto/hive/parquet/reader/ParquetReader.java

@@ -144,7 +144,7 @@ private ColumnChunk readArray(GroupField field)
         BooleanList valueIsNull = new BooleanArrayList();
 
         calculateCollectionOffsets(field, offsets, valueIsNull, columnChunk.getDefinitionLevels(), columnChunk.getRepetitionLevels());
-        Block arrayBlock = ArrayBlock.fromElementBlock(valueIsNull.size(), valueIsNull.toBooleanArray(), offsets.toIntArray(), columnChunk.getBlock());
+        Block arrayBlock = ArrayBlock.fromElementBlock(valueIsNull.size(), Optional.of(valueIsNull.toBooleanArray()), offsets.toIntArray(), columnChunk.getBlock());
         return new ColumnChunk(arrayBlock, columnChunk.getDefinitionLevels(), columnChunk.getRepetitionLevels());
     }
 
@@ -161,7 +161,7 @@ private ColumnChunk readMap(GroupField field)
         IntList offsets = new IntArrayList();
         BooleanList valueIsNull = new BooleanArrayList();
         calculateCollectionOffsets(field, offsets, valueIsNull, columnChunk.getDefinitionLevels(), columnChunk.getRepetitionLevels());
-        Block mapBlock = ((MapType) field.getType()).createBlockFromKeyValue(valueIsNull.toBooleanArray(), offsets.toIntArray(), blocks[0], blocks[1]);
+        Block mapBlock = ((MapType) field.getType()).createBlockFromKeyValue(Optional.of(valueIsNull.toBooleanArray()), offsets.toIntArray(), blocks[0], blocks[1]);
         return new ColumnChunk(mapBlock, columnChunk.getDefinitionLevels(), columnChunk.getRepetitionLevels());
     }
 
@@ -185,7 +185,8 @@ private ColumnChunk readStruct(GroupField field)
             }
         }
         BooleanList structIsNull = ParquetStructColumnReader.calculateStructOffsets(field, columnChunk.getDefinitionLevels(), columnChunk.getRepetitionLevels());
-        Block rowBlock = RowBlock.fromFieldBlocks(structIsNull.toBooleanArray(), blocks);
+        boolean[] structIsNullVector = structIsNull.toBooleanArray();
+        Block rowBlock = RowBlock.fromFieldBlocks(structIsNullVector.length, Optional.of(structIsNullVector), blocks);
         return new ColumnChunk(rowBlock, columnChunk.getDefinitionLevels(), columnChunk.getRepetitionLevels());
     }
 

presto-main/src/test/java/com/facebook/presto/block/TestArrayBlock.java

@@ -180,11 +180,11 @@ public void testCompactBlock()
         int[] offsets = {0, 1, 1, 2, 4, 8, 16};
         boolean[] valueIsNull = {false, true, false, false, false, false};
 
-        testCompactBlock(fromElementBlock(0, new boolean[0], new int[1], emptyValueBlock));
-        testCompactBlock(fromElementBlock(valueIsNull.length, valueIsNull, offsets, compactValueBlock));
-        testIncompactBlock(fromElementBlock(valueIsNull.length - 1, valueIsNull, offsets, compactValueBlock));
+        testCompactBlock(fromElementBlock(0, Optional.empty(), new int[1], emptyValueBlock));
+        testCompactBlock(fromElementBlock(valueIsNull.length, Optional.of(valueIsNull), offsets, compactValueBlock));
+        testIncompactBlock(fromElementBlock(valueIsNull.length - 1, Optional.of(valueIsNull), offsets, compactValueBlock));
         // underlying value block is not compact
-        testIncompactBlock(fromElementBlock(valueIsNull.length, valueIsNull, offsets, inCompactValueBlock));
+        testIncompactBlock(fromElementBlock(valueIsNull.length, Optional.of(valueIsNull), offsets, inCompactValueBlock));
     }
 
     private static BlockBuilder createBlockBuilderWithValues(long[][][] expectedValues)

presto-main/src/test/java/com/facebook/presto/block/TestMapBlock.java

@@ -72,12 +72,12 @@ public void testCompactBlock()
         int[] offsets = {0, 1, 1, 2, 4, 8, 16};
         boolean[] mapIsNull = {false, true, false, false, false, false};
 
-        testCompactBlock(mapType(TINYINT, TINYINT).createBlockFromKeyValue(new boolean[0], new int[1], emptyBlock, emptyBlock));
-        testCompactBlock(mapType(TINYINT, TINYINT).createBlockFromKeyValue(mapIsNull, offsets, compactKeyBlock, compactValueBlock));
+        testCompactBlock(mapType(TINYINT, TINYINT).createBlockFromKeyValue(Optional.empty(), new int[1], emptyBlock, emptyBlock));
+        testCompactBlock(mapType(TINYINT, TINYINT).createBlockFromKeyValue(Optional.of(mapIsNull), offsets, compactKeyBlock, compactValueBlock));
         // TODO: Add test case for a sliced MapBlock
 
         // underlying key/value block is not compact
-        testIncompactBlock(mapType(TINYINT, TINYINT).createBlockFromKeyValue(mapIsNull, offsets, inCompactKeyBlock, inCompactValueBlock));
+        testIncompactBlock(mapType(TINYINT, TINYINT).createBlockFromKeyValue(Optional.of(mapIsNull), offsets, inCompactKeyBlock, inCompactValueBlock));
     }
 
     private Map<String, Long>[] createTestMap(int... entryCounts)
@@ -158,7 +158,7 @@ private Block createBlockWithValuesFromKeyValueBlock(Map<String, Long>[] maps)
                 offsets[i + 1] = offsets[i] + map.size();
             }
         }
-        return mapType(VARCHAR, BIGINT).createBlockFromKeyValue(mapIsNull, offsets, createStringsBlock(keys), createLongsBlock(values));
+        return mapType(VARCHAR, BIGINT).createBlockFromKeyValue(Optional.of(mapIsNull), offsets, createStringsBlock(keys), createLongsBlock(values));
     }
 
     private void createBlockBuilderWithValues(Map<String, Long> map, BlockBuilder mapBlockBuilder)

presto-main/src/test/java/com/facebook/presto/block/TestRowBlock.java

@@ -88,13 +88,13 @@ public void testCompactBlock()
         Block incompactFiledBlock2 = new ByteArrayBlock(5, Optional.empty(), createExpectedValue(6).getBytes());
         boolean[] rowIsNull = {false, true, false, false, false, false};
 
-        assertCompact(fromFieldBlocks(new boolean[0], new Block[] {emptyBlock, emptyBlock}));
-        assertCompact(fromFieldBlocks(rowIsNull, new Block[] {compactFieldBlock1, compactFieldBlock2}));
+        assertCompact(fromFieldBlocks(0, Optional.empty(), new Block[] {emptyBlock, emptyBlock}));
+        assertCompact(fromFieldBlocks(rowIsNull.length, Optional.of(rowIsNull), new Block[] {compactFieldBlock1, compactFieldBlock2}));
         // TODO: add test case for a sliced RowBlock
 
         // underlying field blocks are not compact
-        testIncompactBlock(fromFieldBlocks(rowIsNull, new Block[] {incompactFiledBlock1, incompactFiledBlock2}));
-        testIncompactBlock(fromFieldBlocks(rowIsNull, new Block[] {incompactFiledBlock1, incompactFiledBlock2}));
+        testIncompactBlock(fromFieldBlocks(rowIsNull.length, Optional.of(rowIsNull), new Block[] {incompactFiledBlock1, incompactFiledBlock2}));
+        testIncompactBlock(fromFieldBlocks(rowIsNull.length, Optional.of(rowIsNull), new Block[] {incompactFiledBlock1, incompactFiledBlock2}));
     }
 
     private void testWith(List<Type> fieldTypes, List<Object>[] expectedValues)

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkArraySubscript.java

@@ -164,7 +164,7 @@ private static Block createArrayBlock(int positionCount, Block elementsBlock)
             for (int i = 0; i < offsets.length; i++) {
                 offsets[i] = arraySize * i;
             }
-            return ArrayBlock.fromElementBlock(positionCount, new boolean[positionCount], offsets, elementsBlock);
+            return ArrayBlock.fromElementBlock(positionCount, Optional.empty(), offsets, elementsBlock);
         }
 
         private static Block createFixWidthValueBlock(int positionCount, int mapSize)

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkMapConcat.java

@@ -164,7 +164,7 @@ private static Block createMapBlock(MapType mapType, int positionCount, Block ke
             for (int i = 0; i < offsets.length; i++) {
                 offsets[i] = mapSize * i;
             }
-            return mapType.createBlockFromKeyValue(new boolean[positionCount], offsets, keyBlock, valueBlock);
+            return mapType.createBlockFromKeyValue(Optional.empty(), offsets, keyBlock, valueBlock);
         }
 
         private static Block createKeyBlock(int positionCount, List<String> keys)

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkMapSubscript.java

@@ -175,7 +175,7 @@ private static Block createMapBlock(MapType mapType, int positionCount, Block ke
             for (int i = 0; i < offsets.length; i++) {
                 offsets[i] = mapSize * i;
             }
-            return mapType.createBlockFromKeyValue(new boolean[positionCount], offsets, keyBlock, valueBlock);
+            return mapType.createBlockFromKeyValue(Optional.empty(), offsets, keyBlock, valueBlock);
         }
 
         private static Block createKeyBlock(int positionCount, List<String> keys)

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkMapToMapCast.java

@@ -108,7 +108,7 @@ private static Block createMapBlock(MapType mapType, int positionCount, Block ke
             for (int i = 0; i < offsets.length; i++) {
                 offsets[i] = mapSize * i;
             }
-            return mapType.createBlockFromKeyValue(new boolean[positionCount], offsets, keyBlock, valueBlock);
+            return mapType.createBlockFromKeyValue(Optional.empty(), offsets, keyBlock, valueBlock);
         }
 
         private static Block createKeyBlock(int positionCount, int mapSize)

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkRowToRowCast.java

@@ -99,7 +99,7 @@ public void setup()
             Block[] fieldBlocks = fromFieldTypes.stream()
                     .map(type -> createBlock(POSITION_COUNT, type))
                     .toArray(Block[]::new);
-            Block rowBlock = fromFieldBlocks(new boolean[POSITION_COUNT], fieldBlocks);
+            Block rowBlock = fromFieldBlocks(POSITION_COUNT, Optional.empty(), fieldBlocks);
 
             page = new Page(rowBlock);
         }

presto-orc/src/main/java/com/facebook/presto/orc/reader/ListStreamReader.java

@@ -34,6 +34,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.List;
+import java.util.Optional;
 
 import static com.facebook.presto.orc.metadata.Stream.StreamKind.LENGTH;
 import static com.facebook.presto.orc.metadata.Stream.StreamKind.PRESENT;
@@ -106,14 +107,15 @@ public Block readBlock(Type type)
         // We will use the offsetVector as the buffer to read the length values from lengthStream,
         // and the length values will be converted in-place to an offset vector.
         int[] offsetVector = new int[nextBatchSize + 1];
-        boolean[] nullVector = new boolean[nextBatchSize];
+        boolean[] nullVector = null;
         if (presentStream == null) {
             if (lengthStream == null) {
                 throw new OrcCorruptionException(streamDescriptor.getOrcDataSourceId(), "Value is not null but data stream is not present");
             }
             lengthStream.nextIntVector(nextBatchSize, offsetVector, 0);
         }
         else {
+            nullVector = new boolean[nextBatchSize];
             int nullValues = presentStream.getUnsetBits(nextBatchSize, nullVector);
             if (nullValues != nextBatchSize) {
                 if (lengthStream == null) {
@@ -143,7 +145,7 @@ public Block readBlock(Type type)
         else {
             elements = elementType.createBlockBuilder(null, 0).build();
         }
-        Block arrayBlock = ArrayBlock.fromElementBlock(nextBatchSize, nullVector, offsetVector, elements);
+        Block arrayBlock = ArrayBlock.fromElementBlock(nextBatchSize, Optional.ofNullable(nullVector), offsetVector, elements);
 
         readOffset = 0;
         nextBatchSize = 0;

presto-orc/src/main/java/com/facebook/presto/orc/reader/MapStreamReader.java

@@ -35,6 +35,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.List;
+import java.util.Optional;
 
 import static com.facebook.presto.orc.metadata.Stream.StreamKind.LENGTH;
 import static com.facebook.presto.orc.metadata.Stream.StreamKind.PRESENT;
@@ -110,7 +111,7 @@ public Block readBlock(Type type)
         // We will use the offsetVector as the buffer to read the length values from lengthStream,
         // and the length values will be converted in-place to an offset vector.
         int[] offsetVector = new int[nextBatchSize + 1];
-        boolean[] nullVector = new boolean[nextBatchSize];
+        boolean[] nullVector = null;
 
         if (presentStream == null) {
             if (lengthStream == null) {
@@ -119,6 +120,7 @@ public Block readBlock(Type type)
             lengthStream.nextIntVector(nextBatchSize, offsetVector, 0);
         }
         else {
+            nullVector = new boolean[nextBatchSize];
             int nullValues = presentStream.getUnsetBits(nextBatchSize, nullVector);
             if (nullValues != nextBatchSize) {
                 if (lengthStream == null) {
@@ -165,7 +167,7 @@ public Block readBlock(Type type)
         readOffset = 0;
         nextBatchSize = 0;
 
-        return mapType.createBlockFromKeyValue(nullVector, offsetVector, keyValueBlock[0], keyValueBlock[1]);
+        return mapType.createBlockFromKeyValue(Optional.ofNullable(nullVector), offsetVector, keyValueBlock[0], keyValueBlock[1]);
     }
 
     private static Block[] createKeyValueBlock(int positionCount, Block keys, Block values, int[] lengths)

presto-orc/src/main/java/com/facebook/presto/orc/reader/StructStreamReader.java

@@ -99,13 +99,14 @@ public Block readBlock(Type type)
             }
         }
 
-        boolean[] nullVector = new boolean[nextBatchSize];
+        boolean[] nullVector = null;
         Block[] blocks;
 
         if (presentStream == null) {
             blocks = getBlocksForType(type, nextBatchSize);
         }
         else {
+            nullVector = new boolean[nextBatchSize];
             int nullValues = presentStream.getUnsetBits(nextBatchSize, nullVector);
             if (nullValues != nextBatchSize) {
                 blocks = getBlocksForType(type, nextBatchSize - nullValues);
@@ -125,7 +126,7 @@ public Block readBlock(Type type)
                 .count() == 1);
 
         // Struct is represented as a row block
-        Block rowBlock = RowBlock.fromFieldBlocks(nullVector, blocks);
+        Block rowBlock = RowBlock.fromFieldBlocks(nextBatchSize, Optional.ofNullable(nullVector), blocks);
 
         readOffset = 0;
         nextBatchSize = 0;

presto-orc/src/test/java/com/facebook/presto/orc/TestStructStreamReader.java

@@ -234,7 +234,7 @@ private void write(TempFile tempFile, Type writerType, List<String> data) throws
             fieldBlocks[i] = blockBuilder.build();
             blockBuilder = blockBuilder.newBlockBuilderLike(null);
         }
-        Block rowBlock = RowBlock.fromFieldBlocks(rowIsNull, fieldBlocks);
+        Block rowBlock = RowBlock.fromFieldBlocks(rowIsNull.length, Optional.of(rowIsNull), fieldBlocks);
         writer.write(new Page(rowBlock));
         writer.close();
     }

presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractArrayBlock.java

@@ -111,9 +111,10 @@ public Block copyRegion(int position, int length)
         Block newValues = getRawElementBlock().copyRegion(startValueOffset, endValueOffset - startValueOffset);
 
         int[] newOffsets = compactOffsets(getOffsets(), position + getOffsetBase(), length);
-        boolean[] newValueIsNull = compactArray(getValueIsNull(), position + getOffsetBase(), length);
+        boolean[] valueIsNull = getValueIsNull();
+        boolean[] newValueIsNull = valueIsNull == null ? null : compactArray(valueIsNull, position + getOffsetBase(), length);
 
-        if (newValues == getRawElementBlock() && newOffsets == getOffsets() && newValueIsNull == getValueIsNull()) {
+        if (newValues == getRawElementBlock() && newOffsets == getOffsets() && newValueIsNull == valueIsNull) {
             return this;
         }
         return createArrayBlockInternal(0, length, newValueIsNull, newOffsets, newValues);
@@ -180,7 +181,8 @@ public long getEstimatedDataSizeForStats(int position)
     public boolean isNull(int position)
     {
         checkReadablePosition(position);
-        return getValueIsNull()[position + getOffsetBase()];
+        boolean[] valueIsNull = getValueIsNull();
+        return valueIsNull == null ? false : valueIsNull[position + getOffsetBase()];
     }
 
     public <T> T apply(ArrayBlockFunction<T> function, int position)

presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractMapBlock.java

@@ -18,6 +18,7 @@
 
 import java.lang.invoke.MethodHandle;
 import java.util.Arrays;
+import java.util.Optional;
 
 import static com.facebook.presto.spi.block.BlockUtil.checkArrayRange;
 import static com.facebook.presto.spi.block.BlockUtil.checkValidRegion;
@@ -121,7 +122,7 @@ public Block copyPositions(int[] positions, int offset, int length)
 
         Block newKeys = getRawKeyBlock().copyPositions(entriesPositions.elements(), 0, entriesPositions.size());
         Block newValues = getRawValueBlock().copyPositions(entriesPositions.elements(), 0, entriesPositions.size());
-        return createMapBlockInternal(0, length, newMapIsNull, newOffsets, newKeys, newValues, newHashTable, keyType, keyBlockNativeEquals, keyNativeHashCode);
+        return createMapBlockInternal(0, length, Optional.of(newMapIsNull), newOffsets, newKeys, newValues, newHashTable, keyType, keyBlockNativeEquals, keyNativeHashCode);
     }
 
     @Override
@@ -133,7 +134,7 @@ public Block getRegion(int position, int length)
         return createMapBlockInternal(
                 position + getOffsetBase(),
                 length,
-                getMapIsNull(),
+                Optional.ofNullable(getMapIsNull()),
                 getOffsets(),
                 getRawKeyBlock(),
                 getRawValueBlock(),
@@ -171,16 +172,17 @@ public Block copyRegion(int position, int length)
         Block newValues = getRawValueBlock().copyRegion(startValueOffset, endValueOffset - startValueOffset);
 
         int[] newOffsets = compactOffsets(getOffsets(), position + getOffsetBase(), length);
-        boolean[] newMapIsNull = compactArray(getMapIsNull(), position + getOffsetBase(), length);
+        boolean[] mapIsNull = getMapIsNull();
+        boolean[] newMapIsNull = mapIsNull == null ? null : compactArray(mapIsNull, position + getOffsetBase(), length);
         int[] newHashTable = compactArray(getHashTables(), startValueOffset * HASH_MULTIPLIER, (endValueOffset - startValueOffset) * HASH_MULTIPLIER);
 
-        if (newKeys == getRawKeyBlock() && newValues == getRawValueBlock() && newOffsets == getOffsets() && newMapIsNull == getMapIsNull() && newHashTable == getHashTables()) {
+        if (newKeys == getRawKeyBlock() && newValues == getRawValueBlock() && newOffsets == getOffsets() && newMapIsNull == mapIsNull && newHashTable == getHashTables()) {
             return this;
         }
         return createMapBlockInternal(
                 0,
                 length,
-                newMapIsNull,
+                Optional.ofNullable(newMapIsNull),
                 newOffsets,
                 newKeys,
                 newValues,
@@ -233,7 +235,7 @@ public Block getSingleValueBlock(int position)
         return createMapBlockInternal(
                 0,
                 1,
-                new boolean[] {isNull(position)},
+                Optional.of(new boolean[] {isNull(position)}),
                 new int[] {0, valueLength},
                 newKeys,
                 newValues,
@@ -269,7 +271,8 @@ public long getEstimatedDataSizeForStats(int position)
     public boolean isNull(int position)
     {
         checkReadablePosition(position);
-        return getMapIsNull()[position + getOffsetBase()];
+        boolean[] mapIsNull = getMapIsNull();
+        return mapIsNull != null && mapIsNull[position + getOffsetBase()];
     }
 
     private void checkReadablePosition(int position)