Produce RunLengthEncodedBlock in RowBlockBuilder when all values are null

core/trino-spi/src/main/java/io/trino/spi/block/AbstractRowBlock.java

@@ -72,7 +72,7 @@ public String getEncodingName()
     }
 
     @Override
-    public final Block copyPositions(int[] positions, int offset, int length)
+    public Block copyPositions(int[] positions, int offset, int length)
     {
         checkArrayRange(positions, offset, length);
 

core/trino-spi/src/main/java/io/trino/spi/block/RowBlockBuilder.java

@@ -25,6 +25,8 @@
 
 import static io.airlift.slice.SizeOf.sizeOf;
 import static io.trino.spi.block.BlockUtil.calculateBlockResetSize;
+import static io.trino.spi.block.BlockUtil.checkArrayRange;
+import static io.trino.spi.block.BlockUtil.checkValidRegion;
 import static io.trino.spi.block.RowBlock.createRowBlockInternal;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -46,6 +48,7 @@ public class RowBlockBuilder
 
     private boolean currentEntryOpened;
     private boolean hasNullRow;
+    private boolean hasNonNullRow;
 
     public RowBlockBuilder(List<Type> fieldTypes, BlockBuilderStatus blockBuilderStatus, int expectedEntries)
     {
@@ -201,6 +204,7 @@ private void entryAdded(boolean isNull)
         }
         rowIsNull[positionCount] = isNull;
         hasNullRow |= isNull;
+        hasNonNullRow |= !isNull;
         positionCount++;
 
         for (int i = 0; i < numFields; i++) {
@@ -220,6 +224,9 @@ public Block build()
         if (currentEntryOpened) {
             throw new IllegalStateException("Current entry must be closed before the block can be built");
         }
+        if (!hasNonNullRow) {
+            return nullRle(positionCount);
+        }
         Block[] fieldBlocks = new Block[numFields];
         for (int i = 0; i < numFields; i++) {
             fieldBlocks[i] = fieldBlockBuilders[i].build();
@@ -243,4 +250,50 @@ public BlockBuilder newBlockBuilderLike(BlockBuilderStatus blockBuilderStatus)
         }
         return new RowBlockBuilder(blockBuilderStatus, newBlockBuilders, new int[newSize + 1], new boolean[newSize]);
     }
+
+    @Override
+    public Block copyPositions(int[] positions, int offset, int length)
+    {
+        checkArrayRange(positions, offset, length);
+
+        if (!hasNonNullRow) {
+            return nullRle(length);
+        }
+        return super.copyPositions(positions, offset, length);
+    }
+
+    @Override
+    public Block getRegion(int position, int length)
+    {
+        int positionCount = getPositionCount();
+        checkValidRegion(positionCount, position, length);
+
+        if (!hasNonNullRow) {
+            return nullRle(length);
+        }
+        return super.getRegion(position, length);
+    }
+
+    @Override
+    public Block copyRegion(int position, int length)
+    {
+        int positionCount = getPositionCount();
+        checkValidRegion(positionCount, position, length);
+
+        if (!hasNonNullRow) {
+            return nullRle(length);
+        }
+        return super.copyRegion(position, length);
+    }
+
+    private RunLengthEncodedBlock nullRle(int length)
+    {
+        Block[] fieldBlocks = new Block[numFields];
+        for (int i = 0; i < numFields; i++) {
+            fieldBlocks[i] = fieldBlockBuilders[i].newBlockBuilderLike(null).build();
+        }
+
+        RowBlock nullRowBlock = createRowBlockInternal(0, 1, new boolean[] {true}, new int[] {0, 0}, fieldBlocks);
+        return new RunLengthEncodedBlock(nullRowBlock, length);
+    }
 }

core/trino-spi/src/test/java/io/trino/spi/block/TestRowBlockBuilder.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed 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.
+ */
+package io.trino.spi.block;
+
+import com.google.common.collect.ImmutableList;
+import org.testng.annotations.Test;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+public class TestRowBlockBuilder
+{
+    @Test
+    public void testBuilderProducesNullRleForNullRows()
+    {
+        // empty block
+        assertIsNullRle(blockBuilder().build(), 0);
+
+        // single null
+        assertIsNullRle(blockBuilder().appendNull().build(), 1);
+
+        // multiple nulls
+        assertIsNullRle(blockBuilder().appendNull().appendNull().build(), 2);
+
+        BlockBuilder blockBuilder = blockBuilder().appendNull().appendNull();
+        assertIsNullRle(blockBuilder.copyPositions(new int[] {0}, 0, 1), 1);
+        assertIsNullRle(blockBuilder.getRegion(0, 1), 1);
+        assertIsNullRle(blockBuilder.copyRegion(0, 1), 1);
+    }
+
+    private static BlockBuilder blockBuilder()
+    {
+        return new RowBlockBuilder(ImmutableList.of(BIGINT), null, 10);
+    }
+
+    private void assertIsNullRle(Block block, int expectedPositionCount)
+    {
+        assertEquals(block.getPositionCount(), expectedPositionCount);
+        assertEquals(block.getClass(), RunLengthEncodedBlock.class);
+        if (expectedPositionCount > 0) {
+            assertTrue(block.isNull(0));
+        }
+    }
+}