Introduce startOffset to RowBlock

Author: SongChujun

core/trino-main/src/main/java/io/trino/operator/output/RowPositionsAppender.java

@@ -21,7 +21,6 @@
 import it.unimi.dsi.fastutil.ints.IntArrayList;
 
 import java.util.Arrays;
-import java.util.List;
 import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -79,9 +78,25 @@ public void append(IntArrayList positions, ValueBlock block)
         }
         ensureCapacity(positions.size());
         RowBlock sourceRowBlock = (RowBlock) block;
+        int startOffset = sourceRowBlock.getOffsetBase();
 
-        for (int i = 0; i < fieldAppenders.length; i++) {
-            fieldAppenders[i].append(positions, sourceRowBlock.getFieldBlock(i));
+        Block[] rawFieldBlocks = sourceRowBlock.getRawFieldBlocks();
+
+        if (startOffset == 0) {
+            for (int i = 0; i < fieldAppenders.length; i++) {
+                fieldAppenders[i].append(positions, rawFieldBlocks[i]);
+            }
+        }
+        else {
+            int[] adjustedPositions = new int[positions.size()];
+            for (int i = 0; i < positions.size(); i++) {
+                adjustedPositions[i] = startOffset + positions.getInt(i);
+            }
+            IntArrayList adjustedPositionsList = IntArrayList.wrap(adjustedPositions);
+
+            for (int i = 0; i < fieldAppenders.length; i++) {
+                fieldAppenders[i].append(adjustedPositionsList, rawFieldBlocks[i]);
+            }
         }
 
         if (sourceRowBlock.mayHaveNull()) {
@@ -108,9 +123,11 @@ public void appendRle(ValueBlock value, int rlePositionCount)
         ensureCapacity(rlePositionCount);
         RowBlock sourceRowBlock = (RowBlock) value;
 
-        List<Block> fieldBlocks = sourceRowBlock.getFieldBlocks();
+        Block[] rawFieldBlocks = sourceRowBlock.getRawFieldBlocks();
+        int startOffset = sourceRowBlock.getOffsetBase();
+
         for (int i = 0; i < fieldAppenders.length; i++) {
-            fieldAppenders[i].appendRle(fieldBlocks.get(i).getSingleValueBlock(0), rlePositionCount);
+            fieldAppenders[i].appendRle(rawFieldBlocks[i].getSingleValueBlock(startOffset), rlePositionCount);
         }
 
         if (sourceRowBlock.isNull(0)) {
@@ -134,9 +151,11 @@ public void append(int position, ValueBlock value)
         ensureCapacity(1);
         RowBlock sourceRowBlock = (RowBlock) value;
 
-        List<Block> fieldBlocks = sourceRowBlock.getFieldBlocks();
+        Block[] rawFieldBlocks = sourceRowBlock.getRawFieldBlocks();
+        int startOffset = sourceRowBlock.getOffsetBase();
+
         for (int i = 0; i < fieldAppenders.length; i++) {
-            fieldAppenders[i].append(position, fieldBlocks.get(i));
+            fieldAppenders[i].append(startOffset + position, rawFieldBlocks[i]);
         }
 
         if (sourceRowBlock.isNull(position)) {

core/trino-spi/pom.xml

@@ -804,6 +804,20 @@
                                     <code>java.method.removed</code>
                                     <old>method io.trino.spi.connector.CatalogHandle io.trino.spi.catalog.CatalogProperties::catalogHandle()</old>
                                 </item>
+                                <item>
+                                    <ignore>true</ignore>
+                                    <code>java.method.visibilityIncreased</code>
+                                    <old>method io.trino.spi.block.Block[] io.trino.spi.block.RowBlock::getRawFieldBlocks()</old>
+                                    <new>method io.trino.spi.block.Block[] io.trino.spi.block.RowBlock::getRawFieldBlocks()</new>
+                                    <oldVisibility>package</oldVisibility>
+                                    <newVisibility>public</newVisibility>
+                                    <justification>getRawFieldBlocks need to be accessed from trino-main</justification>
+                                </item>
+                                <item>
+                                    <ignore>true</ignore>
+                                    <code>java.method.removed</code>
+                                    <old>method io.trino.spi.block.Block io.trino.spi.block.RowBlock::getFieldBlock(int)</old>
+                                </item>
                             </differences>
                         </revapi.differences>
                     </analysisConfiguration>