Avoid direct usage of DictionaryBlock constructor

core/trino-main/src/test/java/io/trino/block/TestBlockBuilder.java

@@ -131,7 +131,7 @@ private static void assertInvalidPosition(Block block, int[] positions, int offs
     {
         assertThatThrownBy(() -> block.getPositions(positions, offset, length).getLong(0, 0))
                 .isInstanceOfAny(IllegalArgumentException.class, IndexOutOfBoundsException.class)
-                .hasMessage("Invalid position %d in block with %d positions", positions[0], block.getPositionCount());
+                .hasMessage("Invalid position %d and length %d in block with %d positions", positions[0], length, block.getPositionCount());
     }
 
     private static void assertInvalidOffset(Block block, int[] positions, int offset, int length)

core/trino-main/src/test/java/io/trino/block/TestDictionaryBlock.java

@@ -41,6 +41,19 @@
 public class TestDictionaryBlock
         extends AbstractTestBlock
 {
+    @Test
+    public void testConstructionIdsOffset()
+    {
+        Slice[] expectedValues = createExpectedValues(10);
+        Block dictionary = createSlicesBlock(expectedValues);
+
+        Block block = DictionaryBlock.create(1, 2, dictionary, new int[] {1, 3, 6, 8});
+        assertThat(block).isInstanceOf(DictionaryBlock.class);
+        assertThat(block.getPositionCount()).isEqualTo(2);
+        assertThat(block.getSlice(0, 0, block.getSliceLength(0))).isEqualTo(expectedValues[3]);
+        assertThat(block.getSlice(1, 0, block.getSliceLength(1))).isEqualTo(expectedValues[6]);
+    }
+
     @Test
     public void testConstructionNoPositions()
     {
@@ -64,6 +77,18 @@ public void testConstructionOnePositions()
         assertThat(block.getSlice(0, 0, block.getSliceLength(0))).isEqualTo(expectedValues[1]);
     }
 
+    @Test
+    public void testConstructionOnePositionsIdsOffset()
+    {
+        Slice[] expectedValues = createExpectedValues(10);
+        Block dictionary = createSlicesBlock(expectedValues);
+
+        Block block = DictionaryBlock.create(2, 1, dictionary, new int[] {1, 5, 9});
+        assertThat(block).isInstanceOf(VariableWidthBlock.class);
+        assertThat(block.getPositionCount()).isEqualTo(1);
+        assertThat(block.getSlice(0, 0, block.getSliceLength(0))).isEqualTo(expectedValues[9]);
+    }
+
     @Test
     public void testConstructionUnnestDictionary()
     {
@@ -79,6 +104,21 @@ public void testConstructionUnnestDictionary()
         assertThat(actualDictionary).isSameAs(innerDictionary);
     }
 
+    @Test
+    public void testConstructionUnnestDictionaryIdsOffset()
+    {
+        Slice[] expectedValues = createExpectedValues(10);
+        Block innerDictionary = createSlicesBlock(expectedValues);
+        DictionaryBlock dictionary = (DictionaryBlock) DictionaryBlock.create(4, innerDictionary, new int[] {1, 3, 5, 7});
+
+        Block block = DictionaryBlock.create(1, 2, dictionary, new int[] {0, 1, 3});
+        assertThat(block).isInstanceOf(DictionaryBlock.class);
+        assertBlock(block, new Slice[] {expectedValues[3], expectedValues[7]});
+
+        Block actualDictionary = ((DictionaryBlock) block).getDictionary();
+        assertThat(actualDictionary).isSameAs(innerDictionary);
+    }
+
     @Test
     public void testSizeInBytes()
     {

core/trino-main/src/test/java/io/trino/operator/join/TestLookupJoinPageBuilder.java

@@ -20,6 +20,7 @@
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
 import io.trino.spi.block.DictionaryBlock;
+import io.trino.spi.block.LongArrayBlock;
 import io.trino.spi.type.Type;
 import org.testng.annotations.Test;
 
@@ -100,7 +101,7 @@ public void testDifferentPositions()
         JoinProbe probe = joinProbeFactory.createJoinProbe(page);
         Page output = lookupJoinPageBuilder.build(probe);
         assertEquals(output.getChannelCount(), 2);
-        assertTrue(output.getBlock(0) instanceof DictionaryBlock);
+        assertTrue(output.getBlock(0) instanceof LongArrayBlock);
         assertEquals(output.getPositionCount(), 0);
         lookupJoinPageBuilder.reset();
 

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

@@ -244,7 +244,7 @@ default Block getPositions(int[] positions, int offset, int length)
     {
         checkArrayRange(positions, offset, length);
 
-        return new DictionaryBlock(offset, length, this, positions);
+        return DictionaryBlock.create(offset, length, this, positions);
     }
 
     /**

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

@@ -100,7 +100,7 @@ private static ColumnarRow toColumnarRowFromDictionaryWithoutNulls(DictionaryBlo
         Block[] fields = new Block[columnarRow.getFieldCount()];
         for (int i = 0; i < fields.length; i++) {
             // Reuse the dictionary ids array directly since no nulls are present
-            fields[i] = new DictionaryBlock(
+            fields[i] = DictionaryBlock.create(
                     dictionaryBlock.getRawIdsOffset(),
                     dictionaryBlock.getPositionCount(),
                     columnarRow.getField(i),

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

@@ -54,24 +54,29 @@ public class DictionaryBlock
 
     public static Block create(int positionCount, Block dictionary, int[] ids)
     {
-        return createInternal(positionCount, dictionary, ids, randomDictionaryId());
+        return create(0, positionCount, dictionary, ids);
+    }
+
+    public static Block create(int idsOffset, int positionCount, Block dictionary, int[] ids)
+    {
+        return createInternal(idsOffset, positionCount, dictionary, ids, randomDictionaryId());
     }
 
     /**
      * This should not only be used when creating a projection of another dictionary block.
      */
     public static Block createProjectedDictionaryBlock(int positionCount, Block dictionary, int[] ids, DictionaryId dictionarySourceId)
     {
-        return createInternal(positionCount, dictionary, ids, dictionarySourceId);
+        return createInternal(0, positionCount, dictionary, ids, dictionarySourceId);
     }
 
-    private static Block createInternal(int positionCount, Block dictionary, int[] ids, DictionaryId dictionarySourceId)
+    private static Block createInternal(int idsOffset, int positionCount, Block dictionary, int[] ids, DictionaryId dictionarySourceId)
     {
         if (positionCount == 0) {
             return dictionary.copyRegion(0, 0);
         }
         if (positionCount == 1) {
-            return dictionary.getRegion(ids[0], 1);
+            return dictionary.getRegion(ids[idsOffset], 1);
         }
 
         // if dictionary is an RLE then this can just be a new RLE
@@ -83,18 +88,14 @@ private static Block createInternal(int positionCount, Block dictionary, int[] i
         if (dictionary instanceof DictionaryBlock dictionaryBlock) {
             int[] newIds = new int[positionCount];
             for (int position = 0; position < positionCount; position++) {
-                newIds[position] = dictionaryBlock.getId(ids[position]);
+                newIds[position] = dictionaryBlock.getId(ids[idsOffset + position]);
             }
             dictionary = dictionaryBlock.getDictionary();
             dictionarySourceId = randomDictionaryId();
             ids = newIds;
+            idsOffset = 0;
         }
-        return new DictionaryBlock(0, positionCount, dictionary, ids, false, false, dictionarySourceId);
-    }
-
-    DictionaryBlock(int idsOffset, int positionCount, Block dictionary, int[] ids)
-    {
-        this(idsOffset, positionCount, dictionary, ids, false, false, randomDictionaryId());
+        return new DictionaryBlock(idsOffset, positionCount, dictionary, ids, false, false, dictionarySourceId);
     }
 
     private DictionaryBlock(int idsOffset, int positionCount, Block dictionary, int[] ids, boolean dictionaryIsCompacted, boolean isSequentialIds, DictionaryId dictionarySourceId)

plugin/trino-hive/src/main/java/io/trino/plugin/hive/MergeFileWriter.java

@@ -62,6 +62,8 @@
 public class MergeFileWriter
         implements FileWriter
 {
+    private static final Page EMPTY_PAGE = new Page(0);
+
     // The bucketPath looks like this: /root/dir/delta_nnnnnnn_mmmmmmm_ssss/bucket_bbbbb(_aaaa)?
     private static final Pattern BUCKET_PATH_MATCHER = Pattern.compile("(?s)(?<rootDir>.*)/(?<dirStart>delta_\\d+_\\d+)_(?<statementId>\\d+)/(?<filenameBase>bucket_(?<bucketNumber>\\d+))(?<attemptId>_\\d+)?$");
 
@@ -228,8 +230,11 @@ public PartitionUpdateAndMergeResults getPartitionUpdateAndMergeResults(Partitio
     private Page buildDeletePage(Block rowIds, long writeId)
     {
         ColumnarRow columnarRow = toColumnarRow(rowIds);
-        checkArgument(!columnarRow.mayHaveNull(), "The rowIdsRowBlock may not have null rows");
         int positionCount = rowIds.getPositionCount();
+        if (positionCount == 0) {
+            return EMPTY_PAGE;
+        }
+        checkArgument(!columnarRow.mayHaveNull(), "The rowIdsRowBlock may not have null rows");
         // We've verified that the rowIds block has no null rows, so it's okay to get the field blocks
         Block[] blockArray = {
                 RunLengthEncodedBlock.create(DELETE_OPERATION_BLOCK, positionCount),