Use bulk BlockBuilder APIs for appending range of blocks

core/trino-main/src/main/java/io/trino/operator/project/MergePages.java

@@ -21,10 +21,6 @@
 import io.trino.spi.Page;
 import io.trino.spi.PageBuilder;
 import io.trino.spi.block.Block;
-import io.trino.spi.block.BlockBuilder;
-import io.trino.spi.block.DictionaryBlock;
-import io.trino.spi.block.RunLengthEncodedBlock;
-import io.trino.spi.block.ValueBlock;
 import io.trino.spi.type.Type;
 
 import java.util.List;
@@ -208,19 +204,8 @@ private void appendPage(Page page)
         {
             pageBuilder.declarePositions(page.getPositionCount());
             for (int channel = 0; channel < types.size(); channel++) {
-                appendRawBlock(
-                        page.getBlock(channel),
-                        pageBuilder.getBlockBuilder(channel));
-            }
-        }
-
-        private void appendRawBlock(Block rawBlock, BlockBuilder blockBuilder)
-        {
-            int length = rawBlock.getPositionCount();
-            switch (rawBlock) {
-                case RunLengthEncodedBlock rleBlock -> blockBuilder.appendRepeated(rleBlock.getValue(), 0, length);
-                case DictionaryBlock dictionaryBlock -> blockBuilder.appendPositions(dictionaryBlock.getDictionary(), dictionaryBlock.getRawIds(), dictionaryBlock.getRawIdsOffset(), length);
-                case ValueBlock valueBlock -> blockBuilder.appendRange(valueBlock, 0, length);
+                Block rawBlock = page.getBlock(channel);
+                pageBuilder.getBlockBuilder(channel).appendBlockRange(rawBlock, 0, rawBlock.getPositionCount());
             }
         }
 

core/trino-main/src/main/java/io/trino/operator/scalar/ArrayConcatFunction.java

@@ -18,7 +18,6 @@
 import io.trino.spi.TrinoException;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BufferedArrayValueBuilder;
-import io.trino.spi.block.ValueBlock;
 import io.trino.spi.function.BoundSignature;
 import io.trino.spi.function.FunctionMetadata;
 import io.trino.spi.function.Signature;
@@ -126,10 +125,7 @@ public static Block concat(Object state, Block[] blocks)
 
         return ((BufferedArrayValueBuilder) state).build(resultPositionCount, elementBuilder -> {
             for (Block block : blocks) {
-                ValueBlock valueBlock = block.getUnderlyingValueBlock();
-                for (int i = 0; i < block.getPositionCount(); i++) {
-                    elementBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-                }
+                elementBuilder.appendBlockRange(block, 0, block.getPositionCount());
             }
         });
     }

core/trino-main/src/main/java/io/trino/operator/scalar/ArrayConcatUtils.java

@@ -17,7 +17,6 @@
 import io.trino.annotation.UsedByGeneratedCode;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
-import io.trino.spi.block.ValueBlock;
 import io.trino.spi.type.Type;
 
 public final class ArrayConcatUtils
@@ -28,10 +27,7 @@ private ArrayConcatUtils() {}
     public static Block appendElement(Type elementType, Block block, long value)
     {
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         elementType.writeLong(blockBuilder, value);
 
@@ -42,10 +38,7 @@ public static Block appendElement(Type elementType, Block block, long value)
     public static Block appendElement(Type elementType, Block block, boolean value)
     {
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         elementType.writeBoolean(blockBuilder, value);
 
@@ -56,10 +49,7 @@ public static Block appendElement(Type elementType, Block block, boolean value)
     public static Block appendElement(Type elementType, Block block, double value)
     {
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         elementType.writeDouble(blockBuilder, value);
 
@@ -70,10 +60,7 @@ public static Block appendElement(Type elementType, Block block, double value)
     public static Block appendElement(Type elementType, Block block, Slice value)
     {
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         elementType.writeSlice(blockBuilder, value);
 
@@ -84,10 +71,7 @@ public static Block appendElement(Type elementType, Block block, Slice value)
     public static Block appendElement(Type elementType, Block block, Object value)
     {
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         elementType.writeObject(blockBuilder, value);
 
@@ -101,10 +85,7 @@ public static Block prependElement(Type elementType, Slice value, Block block)
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
 
         elementType.writeSlice(blockBuilder, value);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         return blockBuilder.build();
     }
@@ -115,10 +96,7 @@ public static Block prependElement(Type elementType, Object value, Block block)
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
 
         elementType.writeObject(blockBuilder, value);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         return blockBuilder.build();
     }
@@ -129,10 +107,7 @@ public static Block prependElement(Type elementType, long value, Block block)
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
 
         elementType.writeLong(blockBuilder, value);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         return blockBuilder.build();
     }
@@ -143,10 +118,7 @@ public static Block prependElement(Type elementType, boolean value, Block block)
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
 
         elementType.writeBoolean(blockBuilder, value);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         return blockBuilder.build();
     }
@@ -157,10 +129,7 @@ public static Block prependElement(Type elementType, double value, Block block)
         BlockBuilder blockBuilder = elementType.createBlockBuilder(null, block.getPositionCount() + 1);
 
         elementType.writeDouble(blockBuilder, value);
-        ValueBlock valueBlock = block.getUnderlyingValueBlock();
-        for (int i = 0; i < block.getPositionCount(); i++) {
-            blockBuilder.append(valueBlock, block.getUnderlyingValuePosition(i));
-        }
+        blockBuilder.appendBlockRange(block, 0, block.getPositionCount());
 
         return blockBuilder.build();
     }

core/trino-main/src/main/java/io/trino/operator/scalar/ArrayFlattenFunction.java

@@ -17,7 +17,6 @@
 import io.trino.metadata.SqlScalarFunction;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
-import io.trino.spi.block.ValueBlock;
 import io.trino.spi.function.BoundSignature;
 import io.trino.spi.function.FunctionMetadata;
 import io.trino.spi.function.Signature;
@@ -76,10 +75,7 @@ public static Block flatten(Type type, Type arrayType, Block array)
         for (int i = 0; i < array.getPositionCount(); i++) {
             if (!array.isNull(i)) {
                 Block subArray = (Block) arrayType.getObject(array, i);
-                ValueBlock subArrayValueBlock = subArray.getUnderlyingValueBlock();
-                for (int j = 0; j < subArray.getPositionCount(); j++) {
-                    builder.append(subArrayValueBlock, subArray.getUnderlyingValuePosition(j));
-                }
+                builder.appendBlockRange(subArray, 0, subArray.getPositionCount());
             }
         }
         return builder.build();

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

@@ -21,7 +21,6 @@
 import static io.airlift.slice.SizeOf.instanceSize;
 import static io.airlift.slice.SizeOf.sizeOf;
 import static io.trino.spi.block.ArrayBlock.createArrayBlockInternal;
-import static io.trino.spi.block.BlockUtil.appendRawBlockRange;
 import static io.trino.spi.block.BlockUtil.calculateNewArraySize;
 import static java.lang.Math.max;
 import static java.util.Objects.checkIndex;
@@ -138,7 +137,7 @@ public void append(ValueBlock block, int position)
         int startOffset = offsets[offsetBase + position];
         int length = offsets[offsetBase + position + 1] - startOffset;
 
-        appendRawBlockRange(arrayBlock.getRawElementBlock(), startOffset, length, values);
+        values.appendBlockRange(arrayBlock.getRawElementBlock(), startOffset, length);
         entryAdded(false);
     }
 
@@ -169,7 +168,7 @@ public void appendRange(ValueBlock block, int offset, int length)
         int startOffset = rawOffsets[rawOffsetBase + offset];
         int endOffset = rawOffsets[rawOffsetBase + offset + length];
 
-        appendRawBlockRange(arrayBlock.getRawElementBlock(), startOffset, endOffset - startOffset, values);
+        values.appendBlockRange(arrayBlock.getRawElementBlock(), startOffset, endOffset - startOffset);
 
         // update offsets for copied data
         for (int i = 0; i < length; i++) {

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

@@ -55,6 +55,19 @@ public interface BlockBuilder
      */
     void appendPositions(ValueBlock block, int[] positions, int offset, int length);
 
+    /**
+     * Append the values in the specified range of a Block, this should be used only when
+     * the Block type can be potentially something other than ValueBlock.
+     */
+    default void appendBlockRange(Block rawBlock, int offset, int length)
+    {
+        switch (rawBlock) {
+            case RunLengthEncodedBlock rleBlock -> appendRepeated(rleBlock.getValue(), 0, length);
+            case DictionaryBlock dictionaryBlock -> appendPositions(dictionaryBlock.getDictionary(), dictionaryBlock.getRawIds(), dictionaryBlock.getRawIdsOffset() + offset, length);
+            case ValueBlock valueBlock -> appendRange(valueBlock, offset, length);
+        }
+    }
+
     /**
      * Appends a null value to the block.
      */

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

@@ -319,15 +319,6 @@ else if (buffer.length < capacity) {
         return buffer;
     }
 
-    static void appendRawBlockRange(Block rawBlock, int offset, int length, BlockBuilder blockBuilder)
-    {
-        switch (rawBlock) {
-            case RunLengthEncodedBlock rleBlock -> blockBuilder.appendRepeated(rleBlock.getValue(), 0, length);
-            case DictionaryBlock dictionaryBlock -> blockBuilder.appendPositions(dictionaryBlock.getDictionary(), dictionaryBlock.getRawIds(), offset, length);
-            case ValueBlock valueBlock -> blockBuilder.appendRange(valueBlock, offset, length);
-        }
-    }
-
     /**
      * Ideally, the underlying nulls array in Block implementations should be a byte array instead of a boolean array.
      * This method is used to perform that conversion until the Block implementations are changed.

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

@@ -23,7 +23,6 @@
 
 import static io.airlift.slice.SizeOf.instanceSize;
 import static io.airlift.slice.SizeOf.sizeOf;
-import static io.trino.spi.block.BlockUtil.appendRawBlockRange;
 import static io.trino.spi.block.BlockUtil.calculateNewArraySize;
 import static io.trino.spi.block.MapBlock.createMapBlockInternal;
 import static io.trino.spi.block.MapHashTables.HASH_MULTIPLIER;
@@ -152,8 +151,8 @@ public void append(ValueBlock block, int position)
         int startOffset = offsets[offsetBase + position];
         int length = offsets[offsetBase + position + 1] - startOffset;
 
-        appendRawBlockRange(mapBlock.getRawKeyBlock(), startOffset, length, keyBlockBuilder);
-        appendRawBlockRange(mapBlock.getRawValueBlock(), startOffset, length, valueBlockBuilder);
+        keyBlockBuilder.appendBlockRange(mapBlock.getRawKeyBlock(), startOffset, length);
+        valueBlockBuilder.appendBlockRange(mapBlock.getRawValueBlock(), startOffset, length);
         entryAdded(false);
     }
 

core/trino-spi/src/main/java/io/trino/spi/type/ArrayType.java

@@ -254,12 +254,8 @@ public Block getObject(Block block, int position)
     public void writeObject(BlockBuilder blockBuilder, Object value)
     {
         Block arrayBlock = (Block) value;
-        ValueBlock valueBlock = arrayBlock.getUnderlyingValueBlock();
-        ((ArrayBlockBuilder) blockBuilder).buildEntry(elementBuilder -> {
-            for (int i = 0; i < arrayBlock.getPositionCount(); i++) {
-                elementBuilder.append(valueBlock, arrayBlock.getUnderlyingValuePosition(i));
-            }
-        });
+        ((ArrayBlockBuilder) blockBuilder).buildEntry(elementBuilder ->
+                elementBuilder.appendBlockRange(arrayBlock, 0, arrayBlock.getPositionCount()));
     }
 
     // FLAT MEMORY LAYOUT

core/trino-spi/src/main/java/io/trino/spi/type/MapType.java

@@ -321,10 +321,8 @@ public void writeObject(BlockBuilder blockBuilder, Object value)
         Block rawValueBlock = sqlMap.getRawValueBlock();
 
         ((MapBlockBuilder) blockBuilder).buildEntry((keyBuilder, valueBuilder) -> {
-            for (int i = 0; i < sqlMap.getSize(); i++) {
-                keyBuilder.append(rawKeyBlock.getUnderlyingValueBlock(), rawKeyBlock.getUnderlyingValuePosition(rawOffset + i));
-                valueBuilder.append(rawValueBlock.getUnderlyingValueBlock(), rawValueBlock.getUnderlyingValuePosition(rawOffset + i));
-            }
+            keyBuilder.appendBlockRange(rawKeyBlock, rawOffset, sqlMap.getSize());
+            valueBuilder.appendBlockRange(rawValueBlock, rawOffset, sqlMap.getSize());
         });
     }