Repository: prestodb/presto
Issue: Division by zero error with empty interleaved block

This change 373c745b30b2db1e96eef844096b3620a11a84c7 added this bit to `InterleavedBlockEncoding.readBlock`:

``` java
byte isEmpty = sliceInput.readByte();
if (isEmpty == 1) {
    return new InterleavedBlock(new Block[0]);
}
```

InterleavedBlock performs divisions using the length of that array (columns), which result in division by zero errors:

``` java
int positionInBlock = position / columns;
```

```
java.sql.SQLException: Query failed (#20160623_104612_01302_fsieu): / by zero
    at com.facebook.presto.jdbc.PrestoResultSet.resultsException(PrestoResultSet.java:1799)
    at com.facebook.presto.jdbc.PrestoResultSet$ResultsPageIterator.computeNext(PrestoResultSet.java:1787)
    at com.facebook.presto.jdbc.PrestoResultSet$ResultsPageIterator.computeNext(PrestoResultSet.java:1756)
    at com.facebook.presto.jdbc.internal.guava.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:143)
    at com.facebook.presto.jdbc.internal.guava.collect.AbstractIterator.hasNext(AbstractIterator.java:138)
    at com.facebook.presto.jdbc.internal.guava.collect.TransformedIterator.hasNext(TransformedIterator.java:43)
    at com.facebook.presto.jdbc.internal.guava.collect.Iterators$5.hasNext(Iterators.java:547)
    at com.facebook.presto.jdbc.PrestoResultSet.next(PrestoResultSet.java:149)
    at com.facebook.presto.verifier.Validator.convertJdbcResultSet(Validator.java:520)
    at com.facebook.presto.verifier.Validator.lambda$getResultSetConverter$4(Validator.java:494)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
    at java.util.concurrent.FutureTask.report(FutureTask.java:122)
    at java.util.concurrent.FutureTask.get(FutureTask.java:206)
    at com.google.common.util.concurrent.SimpleTimeLimiter.callWithTimeout(SimpleTimeLimiter.java:130)
    at com.facebook.presto.verifier.Validator.executeQuery(Validator.java:426)
    at com.facebook.presto.verifier.Validator.executeQueryTest(Validator.java:304)
    at com.facebook.presto.verifier.Validator.validate(Validator.java:218)
    at com.facebook.presto.verifier.Validator.valid(Validator.java:190)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ArithmeticException: / by zero
    at com.facebook.presto.spi.block.AbstractInterleavedBlock.getRegion(AbstractInterleavedBlock.java:260)
    at com.facebook.presto.spi.block.AbstractInterleavedBlock.copyRegion(AbstractInterleavedBlock.java:249)
    at com.facebook.presto.spi.block.AbstractArrayBlock.getSingleValueBlock(AbstractArrayBlock.java:234)
    at com.facebook.presto.operator.TopNOperator$TopNBuilder.getValues(TopNOperator.java:338)
    at com.facebook.presto.operator.TopNOperator$TopNBuilder.addRow(TopNOperator.java:313)
    at com.facebook.presto.operator.TopNOperator$TopNBuilder.mergeWithGlobalCandidates(TopNOperator.java:284)
    at com.facebook.presto.operator.TopNOperator$TopNBuilder.processPage(TopNOperator.java:273)
    at com.facebook.presto.operator.TopNOperator.addInput(TopNOperator.java:202)
    at com.facebook.presto.operator.Driver.processInternal(Driver.java:384)
    at com.facebook.presto.operator.Driver.processFor(Driver.java:301)
    at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:618)
    at com.facebook.presto.execution.TaskExecutor$PrioritizedSplitRunner.process(TaskExecutor.java:529)
    at com.facebook.presto.execution.TaskExecutor$Runner.run(TaskExecutor.java:665)
    ... 3 more
```

This reproduces the issue:

``` java
InterleavedBlock block = new InterleavedBlock(new Block[0]);
block.copyRegion(0, 0);
```

(is that call to copyRegion even valid? If not, there might be another bug further up the stack, too)

Also, the test in `AbstractInterleavedBlock`'s constructor is a bit suspect. The condition checks for "non-negative", but the message says "positive". Moreover, given that methods use that value in division operations, it would seem that we should ensure the value is never 0.

``` java
if (columns < 0) {
    throw new IllegalArgumentException("Number of blocks in InterleavedBlock must be positive");
}
```


Changed files:
presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockEncoding.java
presto-main/src/test/java/com/facebook/presto/block/TestInterleavedBlock.java
presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockBuilder.java
presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractInterleavedBlock.java
presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlock.java

diff --git a/presto-main/src/test/java/com/facebook/presto/block/TestInterleavedBlock.java b/presto-main/src/test/java/com/facebook/presto/block/TestInterleavedBlock.java
index e3380eca5a..3bedad2d57 100644
--- a/presto-main/src/test/java/com/facebook/presto/block/TestInterleavedBlock.java
+++ b/presto-main/src/test/java/com/facebook/presto/block/TestInterleavedBlock.java
@@ -134,16 +134,17 @@ public class TestInterleavedBlock
         Type type = TYPES.get(position % TYPES.size());
         assertInterleavedPosition(ImmutableList.of(type), block.getSingleValueBlock(position), 0, expectedValue);
 
-        assertInterleavedPosition(ImmutableList.of(type), block.getRegion(position, 1), 0, expectedValue);
-        assertInterleavedPosition(TYPES, block.getRegion(0, position + 1), position, expectedValue);
-        assertInterleavedPosition(ImmutableList.of(type), block.getRegion(position, block.getPositionCount() - position), 0, expectedValue);
+        int alignedPosition = position - position % COLUMN_COUNT;
 
-        assertInterleavedPosition(ImmutableList.of(type), block.copyRegion(position, 1), 0, expectedValue);
-        assertInterleavedPosition(TYPES, block.copyRegion(0, position + 1), position, expectedValue);
-        assertInterleavedPosition(ImmutableList.of(type), block.copyRegion(position, block.getPositionCount() - position), 0, expectedValue);
+        assertInterleavedPosition(ImmutableList.of(type), block.getRegion(alignedPosition, COLUMN_COUNT), position - alignedPosition, expectedValue);
+        assertInterleavedPosition(TYPES, block.getRegion(0, alignedPosition + COLUMN_COUNT), position, expectedValue);
+        assertInterleavedPosition(ImmutableList.of(type), block.getRegion(alignedPosition, block.getPositionCount() - alignedPosition), position - alignedPosition, expectedValue);
 
-        int positionFloored = position / COLUMN_COUNT * COLUMN_COUNT;
-        assertInterleavedPosition(TYPES, block.copyPositions(IntStream.range(positionFloored, positionFloored + COLUMN_COUNT).boxed().collect(Collectors.toList())), position % COLUMN_COUNT, expectedValue);
+        assertInterleavedPosition(ImmutableList.of(type), block.copyRegion(alignedPosition, COLUMN_COUNT), position - alignedPosition, expectedValue);
+        assertInterleavedPosition(TYPES, block.copyRegion(0, alignedPosition + COLUMN_COUNT), position, expectedValue);
+        assertInterleavedPosition(ImmutableList.of(type), block.copyRegion(alignedPosition, block.getPositionCount() - alignedPosition), position - alignedPosition, expectedValue);
+
+        assertInterleavedPosition(TYPES, block.copyPositions(IntStream.range(alignedPosition, alignedPosition + COLUMN_COUNT).boxed().collect(Collectors.toList())), position % COLUMN_COUNT, expectedValue);
     }
 
     private <T> void assertInterleavedPosition(List<Type> types, Block block, int position, T expectedValue)
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractInterleavedBlock.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractInterleavedBlock.java
index 80dc9daef9..0fdc5cb081 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractInterleavedBlock.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractInterleavedBlock.java
@@ -32,7 +32,7 @@ public abstract class AbstractInterleavedBlock
 
     protected AbstractInterleavedBlock(int columns)
     {
-        if (columns < 0) {
+        if (columns <= 0) {
             throw new IllegalArgumentException("Number of blocks in InterleavedBlock must be positive");
         }
         this.columns = columns;
@@ -43,12 +43,14 @@ public abstract class AbstractInterleavedBlock
         return columns;
     }
 
-    AbstractInterleavedBlock semiCompact()
+    Block[] computeSerializableSubBlocks()
     {
-        // the returned InterleavedBlock is guaranteed to have
-        // * start equal to 0
-        // * positionCount equal to the sum of positionCount of column blocks
-        return getRegionAlwaysReturnInterleavedBlock(0, getPositionCount(), false);
+        InterleavedBlock interleavedBlock = (InterleavedBlock) sliceRange(0, getPositionCount(), false);
+        Block[] result = new Block[interleavedBlock.getBlockCount()];
+        for (int i = 0; i < result.length; i++) {
+            result[i] = interleavedBlock.getBlock(i);
+        }
+        return result;
     }
 
     /**
@@ -246,40 +248,30 @@ public abstract class AbstractInterleavedBlock
     @Override
     public Block copyRegion(int position, int length)
     {
-        return getRegion(position, length, true);
+        validateRange(position, length);
+        return sliceRange(position, length, true);
     }
 
-    protected Block getRegion(int position, int length, boolean compact)
+    protected void validateRange(int position, int length)
     {
         int positionCount = getPositionCount();
-        if (position < 0 || length < 0 || position + length > positionCount) {
-            throw new IndexOutOfBoundsException("Invalid position (" + position + "), length (" + length + ") in block with " + positionCount + " positions");
-        }
-        if (length <= 1) {
-            position = computePosition(position);
-            int positionInBlock = position / columns;
-            if (compact) {
-                return getBlock(position % columns).copyRegion(positionInBlock, length);
-            }
-            else {
-                return getBlock(position % columns).getRegion(positionInBlock, length);
-            }
+        if (position < 0 || length < 0 || position + length > positionCount || position % columns != 0 || length % columns != 0) {
+            throw new IndexOutOfBoundsException("Invalid position (" + position + "), length (" + length + ") in InterleavedBlock with " + positionCount + " positions and " + columns + " columns");
         }
-        return getRegionAlwaysReturnInterleavedBlock(position, length, compact);
     }
 
-    private InterleavedBlock getRegionAlwaysReturnInterleavedBlock(int position, int length, boolean compact)
+    protected Block sliceRange(int position, int length, boolean compact)
     {
         position = computePosition(position);
-        Block[] resultBlocks = new Block[Math.min(columns, length)];
-        for (int newBlockIndex = 0; newBlockIndex < resultBlocks.length; newBlockIndex++) {
-            int positionInBlock = (position + newBlockIndex) / columns;
-            int subBlockLength = (length + columns - 1 - newBlockIndex) / columns;
+        Block[] resultBlocks = new Block[columns];
+        int positionInBlock = position / columns;
+        int subBlockLength = length / columns;
+        for (int blockIndex = 0; blockIndex < columns; blockIndex++) {
             if (compact) {
-                resultBlocks[newBlockIndex] = getBlock((newBlockIndex + position) % columns).copyRegion(positionInBlock, subBlockLength);
+                resultBlocks[blockIndex] = getBlock((blockIndex + position) % columns).copyRegion(positionInBlock, subBlockLength);
             }
             else {
-                resultBlocks[newBlockIndex] = getBlock((newBlockIndex + position) % columns).getRegion(positionInBlock, subBlockLength);
+                resultBlocks[blockIndex] = getBlock((blockIndex + position) % columns).getRegion(positionInBlock, subBlockLength);
             }
         }
         return new InterleavedBlock(resultBlocks);
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlock.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlock.java
index d5a104133f..2d53a1c711 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlock.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlock.java
@@ -35,37 +35,17 @@ public class InterleavedBlock
         super(blocks.length);
         this.blocks = blocks;
 
-        // Aside from calculating sizeInBytes, retainedSizeInBytes, and positionCount,
-        // the loop below verifies that the position count of sub-blocks in the InterleavedBlock
-        // * differs by at most one
-        // * is non-ascending
         int sizeInBytes = 0;
         int retainedSizeInBytes = INSTANCE_SIZE;
         int positionCount = 0;
-        if (blocks.length != 0) {
-            int firstSubBlockPositionCount = blocks[0].getPositionCount();
-            boolean subBlockHasDifferentSize = false;
-            for (int i = 0; i < getBlockCount(); i++) {
-                sizeInBytes += blocks[i].getSizeInBytes();
-                retainedSizeInBytes += blocks[i].getRetainedSizeInBytes();
-                positionCount += blocks[i].getPositionCount();
-
-                if (subBlockHasDifferentSize) {
-                    if (firstSubBlockPositionCount - 1 != blocks[i].getPositionCount()) {
-                        throw new IllegalArgumentException(
-                                "length of sub blocks differ by at least 2 or is not non-ascending: block 0: " + firstSubBlockPositionCount + ", block " + i + ": " + blocks[i].getPositionCount());
-                    }
-                }
-                else {
-                    if (firstSubBlockPositionCount != blocks[i].getPositionCount()) {
-                        if (firstSubBlockPositionCount - 1 == blocks[i].getPositionCount()) {
-                            subBlockHasDifferentSize = true;
-                        }
-                        else {
-                            throw new IllegalArgumentException("length of sub blocks differ by at least 2: block 0: " + firstSubBlockPositionCount + ", block " + i + ": " + blocks[i].getPositionCount());
-                        }
-                    }
-                }
+        int firstSubBlockPositionCount = blocks[0].getPositionCount();
+        for (int i = 0; i < getBlockCount(); i++) {
+            sizeInBytes += blocks[i].getSizeInBytes();
+            retainedSizeInBytes += blocks[i].getRetainedSizeInBytes();
+            positionCount += blocks[i].getPositionCount();
+
+            if (firstSubBlockPositionCount != blocks[i].getPositionCount()) {
+                throw new IllegalArgumentException("length of sub blocks differ: block 0: " + firstSubBlockPositionCount + ", block " + i + ": " + blocks[i].getPositionCount());
             }
         }
 
@@ -90,9 +70,7 @@ public class InterleavedBlock
     @Override
     public Block getRegion(int position, int length)
     {
-        if (position < 0 || length < 0 || position + length > positionCount) {
-            throw new IndexOutOfBoundsException("Invalid position (" + position + "), length (" + length + ") in block with " + positionCount + " positions");
-        }
+        validateRange(position, length);
         return new InterleavedBlock(blocks, computePosition(position), length, retainedSizeInBytes, blockEncoding);
     }
 
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockBuilder.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockBuilder.java
index 2c30b96f5e..395ac8548a 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockBuilder.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockBuilder.java
@@ -238,7 +238,8 @@ public class InterleavedBlockBuilder
     @Override
     public Block getRegion(int position, int length)
     {
-        return getRegion(position, length, false);
+        validateRange(position, length);
+        return sliceRange(position, length, false);
     }
 
     @Override
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockEncoding.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockEncoding.java
index a62b514079..a595b8e92a 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockEncoding.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/InterleavedBlockEncoding.java
@@ -39,25 +39,16 @@ public class InterleavedBlockEncoding
     @Override
     public void writeBlock(SliceOutput sliceOutput, Block block)
     {
-        if (block.getPositionCount() == 0) {
-            sliceOutput.appendByte(1);
-            return;
-        }
-        else {
-            sliceOutput.appendByte(0);
-        }
-
         AbstractInterleavedBlock interleavedBlock = (AbstractInterleavedBlock) block;
 
-        interleavedBlock = interleavedBlock.semiCompact();
-
         if (interleavedBlock.getBlockCount() != individualBlockEncodings.length) {
             throw new IllegalArgumentException(
                     "argument block differs in length (" + interleavedBlock.getBlockCount() + ") with this encoding (" + individualBlockEncodings.length + ")");
         }
 
-        for (int i = 0; i < individualBlockEncodings.length; i++) {
-            individualBlockEncodings[i].writeBlock(sliceOutput, interleavedBlock.getBlock(i));
+        Block[] subBlocks = interleavedBlock.computeSerializableSubBlocks();
+        for (int i = 0; i < subBlocks.length; i++) {
+            individualBlockEncodings[i].writeBlock(sliceOutput, subBlocks[i]);
         }
     }
 
@@ -71,11 +62,6 @@ public class InterleavedBlockEncoding
     @Override
     public Block readBlock(SliceInput sliceInput)
     {
-        byte isEmpty = sliceInput.readByte();
-        if (isEmpty == 1) {
-            return new InterleavedBlock(new Block[0]);
-        }
-
         Block[] individualBlocks = new Block[individualBlockEncodings.length];
         for (int i = 0; i < individualBlockEncodings.length; i++) {
             individualBlocks[i] = individualBlockEncodings[i].readBlock(sliceInput);