Add value block abstraction

core/trino-main/src/main/java/io/trino/metadata/FunctionManager.java

@@ -23,6 +23,7 @@
 import io.trino.spi.TrinoException;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.block.ValueBlock;
 import io.trino.spi.connector.CatalogHandle;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.function.AggregationImplementation;
@@ -240,6 +241,11 @@ private static void verifyMethodHandleSignature(BoundSignature boundSignature, S
                     verifyFunctionSignature(parameterType.equals(Block.class) && methodType.parameterType(parameterIndex + 1).equals(int.class),
                             "Expected %s argument types to be Block and int".formatted(argumentConvention));
                     break;
+                case VALUE_BLOCK_POSITION:
+                case VALUE_BLOCK_POSITION_NOT_NULL:
+                    verifyFunctionSignature(ValueBlock.class.isAssignableFrom(parameterType) && methodType.parameterType(parameterIndex + 1).equals(int.class),
+                            "Expected %s argument types to be ValueBlock and int".formatted(argumentConvention));
+                    break;
                 case FLAT:
                     verifyFunctionSignature(parameterType.equals(byte[].class) &&
                                     methodType.parameterType(parameterIndex + 1).equals(int.class) &&

core/trino-main/src/main/java/io/trino/operator/BucketPartitionFunction.java

@@ -35,7 +35,7 @@ public BucketPartitionFunction(BucketFunction bucketFunction, int[] bucketToPart
     }
 
     @Override
-    public int getPartitionCount()
+    public int partitionCount()
     {
         return partitionCount;
     }

core/trino-main/src/main/java/io/trino/operator/ChangeOnlyUpdatedColumnsMergeProcessor.java

@@ -68,7 +68,11 @@ public Page transformPage(Page inputPage)
         checkArgument(positionCount > 0, "positionCount should be > 0, but is %s", positionCount);
 
         ColumnarRow mergeRow = toColumnarRow(inputPage.getBlock(mergeRowChannel));
-        checkArgument(!mergeRow.mayHaveNull(), "The mergeRow may not have null rows");
+        if (mergeRow.mayHaveNull()) {
+            for (int position = 0; position < positionCount; position++) {
+                checkArgument(!mergeRow.isNull(position), "The mergeRow may not have null rows");
+            }
+        }
 
         // We've verified that the mergeRow block has no null rows, so it's okay to get the field blocks
 

core/trino-main/src/main/java/io/trino/operator/JoinDomainBuilder.java

@@ -18,6 +18,9 @@
 import io.airlift.units.DataSize;
 import io.trino.spi.TrinoException;
 import io.trino.spi.block.Block;
+import io.trino.spi.block.DictionaryBlock;
+import io.trino.spi.block.RunLengthEncodedBlock;
+import io.trino.spi.block.ValueBlock;
 import io.trino.spi.predicate.Domain;
 import io.trino.spi.predicate.ValueSet;
 import io.trino.spi.type.Type;
@@ -32,8 +35,8 @@
 import static io.trino.operator.VariableWidthData.EMPTY_CHUNK;
 import static io.trino.operator.VariableWidthData.POINTER_SIZE;
 import static io.trino.spi.StandardErrorCode.GENERIC_INSUFFICIENT_RESOURCES;
-import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.BLOCK_POSITION_NOT_NULL;
 import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.FLAT;
+import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.VALUE_BLOCK_POSITION_NOT_NULL;
 import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.FAIL_ON_NULL;
 import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.FLAT_RETURN;
 import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.NULLABLE_RETURN;
@@ -88,8 +91,8 @@ public class JoinDomainBuilder
     private int distinctSize;
     private int distinctMaxFill;
 
-    private Block minValue;
-    private Block maxValue;
+    private ValueBlock minValue;
+    private ValueBlock maxValue;
 
     private boolean collectDistinctValues = true;
     private boolean collectMinMax;
@@ -116,15 +119,15 @@ public JoinDomainBuilder(
         MethodHandle readOperator = typeOperators.getReadValueOperator(type, simpleConvention(NULLABLE_RETURN, FLAT));
         readOperator = readOperator.asType(readOperator.type().changeReturnType(Object.class));
         this.readFlat = readOperator;
-        this.writeFlat = typeOperators.getReadValueOperator(type, simpleConvention(FLAT_RETURN, BLOCK_POSITION_NOT_NULL));
+        this.writeFlat = typeOperators.getReadValueOperator(type, simpleConvention(FLAT_RETURN, VALUE_BLOCK_POSITION_NOT_NULL));
 
         this.hashFlat = typeOperators.getHashCodeOperator(type, simpleConvention(FAIL_ON_NULL, FLAT));
-        this.hashBlock = typeOperators.getHashCodeOperator(type, simpleConvention(FAIL_ON_NULL, BLOCK_POSITION_NOT_NULL));
+        this.hashBlock = typeOperators.getHashCodeOperator(type, simpleConvention(FAIL_ON_NULL, VALUE_BLOCK_POSITION_NOT_NULL));
         this.distinctFlatFlat = typeOperators.getDistinctFromOperator(type, simpleConvention(FAIL_ON_NULL, FLAT, FLAT));
-        this.distinctFlatBlock = typeOperators.getDistinctFromOperator(type, simpleConvention(FAIL_ON_NULL, FLAT, BLOCK_POSITION_NOT_NULL));
+        this.distinctFlatBlock = typeOperators.getDistinctFromOperator(type, simpleConvention(FAIL_ON_NULL, FLAT, VALUE_BLOCK_POSITION_NOT_NULL));
         if (collectMinMax) {
             this.compareFlatFlat = typeOperators.getComparisonUnorderedLastOperator(type, simpleConvention(FAIL_ON_NULL, FLAT, FLAT));
-            this.compareBlockBlock = typeOperators.getComparisonUnorderedLastOperator(type, simpleConvention(FAIL_ON_NULL, BLOCK_POSITION_NOT_NULL, BLOCK_POSITION_NOT_NULL));
+            this.compareBlockBlock = typeOperators.getComparisonUnorderedLastOperator(type, simpleConvention(FAIL_ON_NULL, VALUE_BLOCK_POSITION_NOT_NULL, VALUE_BLOCK_POSITION_NOT_NULL));
         }
         else {
             this.compareFlatFlat = null;
@@ -157,9 +160,24 @@ public boolean isCollecting()
 
     public void add(Block block)
     {
+        block = block.getLoadedBlock();
         if (collectDistinctValues) {
-            for (int position = 0; position < block.getPositionCount(); ++position) {
-                add(block, position);
+            if (block instanceof ValueBlock valueBlock) {
+                for (int position = 0; position < block.getPositionCount(); position++) {
+                    add(valueBlock, position);
+                }
+            }
+            else if (block instanceof RunLengthEncodedBlock rleBlock) {
+                add(rleBlock.getValue(), 0);
+            }
+            else if (block instanceof DictionaryBlock dictionaryBlock) {
+                ValueBlock dictionary = dictionaryBlock.getDictionary();
+                for (int i = 0; i < dictionaryBlock.getPositionCount(); i++) {
+                    add(dictionary, dictionaryBlock.getId(i));
+                }
+            }
+            else {
+                throw new IllegalArgumentException("Unsupported block type: " + block.getClass().getSimpleName());
             }
 
             // if the distinct size is too large, fall back to min max, and drop the distinct values
@@ -207,8 +225,10 @@ else if (collectMinMax) {
             int minValuePosition = -1;
             int maxValuePosition = -1;
 
-            for (int position = 0; position < block.getPositionCount(); ++position) {
-                if (block.isNull(position)) {
+            ValueBlock valueBlock = block.getUnderlyingValueBlock();
+            for (int i = 0; i < block.getPositionCount(); i++) {
+                int position = block.getUnderlyingValuePosition(i);
+                if (valueBlock.isNull(position)) {
                     continue;
                 }
                 if (minValuePosition == -1) {
@@ -217,10 +237,10 @@ else if (collectMinMax) {
                     maxValuePosition = position;
                     continue;
                 }
-                if (valueCompare(block, position, block, minValuePosition) < 0) {
+                if (valueCompare(valueBlock, position, valueBlock, minValuePosition) < 0) {
                     minValuePosition = position;
                 }
-                else if (valueCompare(block, position, block, maxValuePosition) > 0) {
+                else if (valueCompare(valueBlock, position, valueBlock, maxValuePosition) > 0) {
                     maxValuePosition = position;
                 }
             }
@@ -231,18 +251,18 @@ else if (valueCompare(block, position, block, maxValuePosition) > 0) {
             }
 
             if (minValue == null) {
-                minValue = block.getSingleValueBlock(minValuePosition);
-                maxValue = block.getSingleValueBlock(maxValuePosition);
+                minValue = valueBlock.getSingleValueBlock(minValuePosition);
+                maxValue = valueBlock.getSingleValueBlock(maxValuePosition);
                 return;
             }
-            if (valueCompare(block, minValuePosition, minValue, 0) < 0) {
+            if (valueCompare(valueBlock, minValuePosition, minValue, 0) < 0) {
                 retainedSizeInBytes -= minValue.getRetainedSizeInBytes();
-                minValue = block.getSingleValueBlock(minValuePosition);
+                minValue = valueBlock.getSingleValueBlock(minValuePosition);
                 retainedSizeInBytes += minValue.getRetainedSizeInBytes();
             }
-            if (valueCompare(block, maxValuePosition, maxValue, 0) > 0) {
+            if (valueCompare(valueBlock, maxValuePosition, maxValue, 0) > 0) {
                 retainedSizeInBytes -= maxValue.getRetainedSizeInBytes();
-                maxValue = block.getSingleValueBlock(maxValuePosition);
+                maxValue = valueBlock.getSingleValueBlock(maxValuePosition);
                 retainedSizeInBytes += maxValue.getRetainedSizeInBytes();
             }
         }
@@ -289,7 +309,7 @@ public Domain build()
         return Domain.all(type);
     }
 
-    private void add(Block block, int position)
+    private void add(ValueBlock block, int position)
     {
         // Inner and right join doesn't match rows with null key column values.
         if (block.isNull(position)) {
@@ -343,7 +363,7 @@ private int matchInVector(byte[] otherValues, VariableWidthData otherVariableWid
         return -1;
     }
 
-    private int matchInVector(Block block, int position, int vectorStartBucket, long repeated, long controlVector)
+    private int matchInVector(ValueBlock block, int position, int vectorStartBucket, long repeated, long controlVector)
     {
         long controlMatches = match(controlVector, repeated);
         while (controlMatches != 0) {
@@ -367,7 +387,7 @@ private int findEmptyInVector(long vector, int vectorStartBucket)
         return bucket(vectorStartBucket + slot);
     }
 
-    private void insert(int index, Block block, int position, byte hashPrefix)
+    private void insert(int index, ValueBlock block, int position, byte hashPrefix)
     {
         setControl(index, hashPrefix);
 
@@ -512,7 +532,7 @@ private Object readValueToObject(int position)
         }
     }
 
-    private Block readValueToBlock(int position)
+    private ValueBlock readValueToBlock(int position)
     {
         return writeNativeValue(type, readValueToObject(position));
     }
@@ -538,7 +558,7 @@ private long valueHashCode(byte[] values, int position)
         }
     }
 
-    private long valueHashCode(Block right, int rightPosition)
+    private long valueHashCode(ValueBlock right, int rightPosition)
     {
         try {
             return (long) hashBlock.invokeExact(right, rightPosition);
@@ -549,7 +569,7 @@ private long valueHashCode(Block right, int rightPosition)
         }
     }
 
-    private boolean valueNotDistinctFrom(int leftPosition, Block right, int rightPosition)
+    private boolean valueNotDistinctFrom(int leftPosition, ValueBlock right, int rightPosition)
     {
         byte[] leftFixedRecordChunk = distinctRecords;
         int leftRecordOffset = getRecordOffset(leftPosition);
@@ -603,7 +623,7 @@ private boolean valueNotDistinctFrom(int leftPosition, byte[] rightValues, Varia
         }
     }
 
-    private int valueCompare(Block left, int leftPosition, Block right, int rightPosition)
+    private int valueCompare(ValueBlock left, int leftPosition, ValueBlock right, int rightPosition)
     {
         try {
             return (int) (long) compareBlockBlock.invokeExact(

core/trino-main/src/main/java/io/trino/operator/PartitionFunction.java

@@ -17,7 +17,7 @@
 
 public interface PartitionFunction
 {
-    int getPartitionCount();
+    int partitionCount();
 
     /**
      * @param page the arguments to bucketing function in order (no extra columns)

core/trino-main/src/main/java/io/trino/operator/aggregation/AbstractMapAggregationState.java

@@ -17,9 +17,9 @@
 import com.google.common.primitives.Ints;
 import io.trino.operator.VariableWidthData;
 import io.trino.spi.TrinoException;
-import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
 import io.trino.spi.block.MapBlockBuilder;
+import io.trino.spi.block.ValueBlock;
 import io.trino.spi.type.Type;
 import jakarta.annotation.Nullable;
 
@@ -285,7 +285,7 @@ private void serializeEntry(BlockBuilder keyBuilder, BlockBuilder valueBuilder,
         }
     }
 
-    protected void add(int groupId, Block keyBlock, int keyPosition, Block valueBlock, int valuePosition)
+    protected void add(int groupId, ValueBlock keyBlock, int keyPosition, ValueBlock valueBlock, int valuePosition)
     {
         checkArgument(!keyBlock.isNull(keyPosition), "key must not be null");
         checkArgument(groupId == 0 || groupRecordIndex != null, "groupId must be zero when grouping is not enabled");
@@ -322,7 +322,7 @@ protected void add(int groupId, Block keyBlock, int keyPosition, Block valueBloc
         }
     }
 
-    private int matchInVector(int groupId, Block block, int position, int vectorStartBucket, long repeated, long controlVector)
+    private int matchInVector(int groupId, ValueBlock block, int position, int vectorStartBucket, long repeated, long controlVector)
     {
         long controlMatches = match(controlVector, repeated);
         while (controlMatches != 0) {
@@ -346,7 +346,7 @@ private int findEmptyInVector(long vector, int vectorStartBucket)
         return bucket(vectorStartBucket + slot);
     }
 
-    private void insert(int index, int groupId, Block keyBlock, int keyPosition, Block valueBlock, int valuePosition, byte hashPrefix)
+    private void insert(int index, int groupId, ValueBlock keyBlock, int keyPosition, ValueBlock valueBlock, int valuePosition, byte hashPrefix)
     {
         setControl(index, hashPrefix);
 
@@ -499,7 +499,7 @@ private long keyHashCode(int groupId, byte[] records, int index)
         }
     }
 
-    private long keyHashCode(int groupId, Block right, int rightPosition)
+    private long keyHashCode(int groupId, ValueBlock right, int rightPosition)
     {
         try {
             long valueHash = (long) keyHashBlock.invokeExact(right, rightPosition);
@@ -511,7 +511,7 @@ private long keyHashCode(int groupId, Block right, int rightPosition)
         }
     }
 
-    private boolean keyNotDistinctFrom(int leftPosition, Block right, int rightPosition, int rightGroupId)
+    private boolean keyNotDistinctFrom(int leftPosition, ValueBlock right, int rightPosition, int rightGroupId)
     {
         byte[] leftRecords = getRecords(leftPosition);
         int leftRecordOffset = getRecordOffset(leftPosition);