Remove LazyBlock

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

@@ -21,7 +21,6 @@
 import io.trino.spi.block.Fixed12BlockEncoding;
 import io.trino.spi.block.Int128ArrayBlockEncoding;
 import io.trino.spi.block.IntArrayBlockEncoding;
-import io.trino.spi.block.LazyBlockEncoding;
 import io.trino.spi.block.LongArrayBlockEncoding;
 import io.trino.spi.block.MapBlockEncoding;
 import io.trino.spi.block.RowBlockEncoding;
@@ -57,7 +56,6 @@ public BlockEncodingManager()
         addBlockEncoding(new MapBlockEncoding());
         addBlockEncoding(new RowBlockEncoding());
         addBlockEncoding(new RunLengthBlockEncoding());
-        addBlockEncoding(new LazyBlockEncoding());
     }
 
     public BlockEncoding getBlockEncodingByName(String encodingName)

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

@@ -65,7 +65,7 @@ public Page transformPage(Page inputPage)
         int positionCount = inputPage.getPositionCount();
         checkArgument(positionCount > 0, "positionCount should be > 0, but is %s", positionCount);
 
-        Block mergeRow = inputPage.getBlock(mergeRowChannel).getLoadedBlock();
+        Block mergeRow = inputPage.getBlock(mergeRowChannel);
         List<Block> fields = getRowFieldsFromBlock(mergeRow);
         List<Block> builder = new ArrayList<>(dataColumnChannels.size() + 4);
         for (int channel : dataColumnChannels) {

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

@@ -14,13 +14,11 @@
 package io.trino.operator;
 
 import com.google.common.base.Throwables;
-import com.google.common.base.VerifyException;
 import com.google.common.collect.ImmutableList;
 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.LazyBlock;
 import io.trino.spi.block.RunLengthEncodedBlock;
 import io.trino.spi.block.ValueBlock;
 import io.trino.spi.predicate.Domain;
@@ -162,7 +160,6 @@ public boolean isCollecting()
 
     public void add(Block block)
     {
-        block = block.getLoadedBlock();
         if (collectDistinctValues) {
             switch (block) {
                 case ValueBlock valueBlock -> {
@@ -177,7 +174,6 @@ public void add(Block block)
                         add(dictionary, dictionaryBlock.getId(i));
                     }
                 }
-                case LazyBlock _ -> throw new VerifyException("Did not expect LazyBlock after loading " + block.getClass().getSimpleName());
             }
 
             // if the distinct size is too large, fall back to min max, and drop the distinct values

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

@@ -65,7 +65,6 @@ public LoadCursor add(Page page, int startingPosition)
             // Initiate additional actions on close
             checkState(currentCursor != null);
             pageAccounting.unlockPage();
-            pageAccounting.loadPageLoadIfNeeded();
             // Account for page size after lazy loading (which can change the page size)
             pageBytes += pageAccounting.sizeOf();
             currentCursor = null;
@@ -237,7 +236,6 @@ private final class PageAccounting
 
         private final int pageId;
         private Page page;
-        private boolean isPageLoaded;
         private long[] rowIds;
         // Start off locked to give the caller time to declare which rows to reference
         private boolean lockedPage = true;
@@ -315,14 +313,6 @@ public boolean isCompactionEligible()
             return !lockedPage && activePositions * COMPACTION_MIN_FILL_MULTIPLIER < page.getPositionCount();
         }
 
-        public void loadPageLoadIfNeeded()
-        {
-            if (!isPageLoaded && activePositions > 0) {
-                page = page.getLoadedPage();
-                isPageLoaded = true;
-            }
-        }
-
         public void compact()
         {
             checkState(!lockedPage, "Should not attempt compaction when page is locked");
@@ -331,8 +321,6 @@ public void compact()
                 return;
             }
 
-            loadPageLoadIfNeeded();
-
             int newIndex = 0;
             int[] positionsToKeep = new int[activePositions];
             long[] newRowIds = new long[activePositions];
@@ -354,9 +342,7 @@ public void compact()
 
         public long sizeOf()
         {
-            // Getting the size of a page forces a lazy page to be loaded, so only provide the size after an explicit decision to load
-            long loadedPageSize = isPageLoaded ? page.getSizeInBytes() : 0;
-            return PAGE_ACCOUNTING_INSTANCE_SIZE + loadedPageSize + SizeOf.sizeOf(rowIds);
+            return PAGE_ACCOUNTING_INSTANCE_SIZE + page.getSizeInBytes() + SizeOf.sizeOf(rowIds);
         }
     }
 

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

@@ -295,8 +295,7 @@ private void processInput(Page page)
                     startPosition = nextGroupStart;
                 }
                 else {
-                    // late materialization requires that page being locally stored is materialized before the next one is fetched
-                    currentGroup = page.getRegion(page.getPositionCount() - 1, 1).getLoadedPage();
+                    currentGroup = page.getRegion(page.getPositionCount() - 1, 1);
                     return;
                 }
             }

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

@@ -270,8 +270,6 @@ public Page getOutput()
         Page page = null;
         if (sourcePage != null) {
             page = sourcePage.getPage();
-            // assure the page is in memory before handing to another operator
-            page = page.getLoadedPage();
 
             // update operator stats
             long endCompletedBytes = source.getCompletedBytes();

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

@@ -66,7 +66,6 @@ public WorkProcessorSourceOperatorAdapter(OperatorContext operatorContext, WorkP
                         operatorContext.getDriverContext().getYieldSignal(),
                         WorkProcessor.create(splitBuffer));
         this.pages = sourceOperator.getOutputPages()
-                .map(Page::getLoadedPage)
                 .withProcessStateMonitor(state -> updateOperatorStats())
                 .finishWhen(() -> operatorFinishing);
         operatorContext.setInfoSupplier(() -> sourceOperator.getOperatorInfo().orElse(null));

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

@@ -29,7 +29,6 @@
 import io.trino.spi.block.Block;
 import io.trino.spi.block.ByteArrayBlock;
 import io.trino.spi.block.DictionaryBlock;
-import io.trino.spi.block.LazyBlock;
 import io.trino.spi.block.RunLengthEncodedBlock;
 import io.trino.spi.block.ValueBlock;
 
@@ -50,14 +49,12 @@
 import static io.airlift.bytecode.expression.BytecodeExpressions.constantFalse;
 import static io.airlift.bytecode.expression.BytecodeExpressions.constantInt;
 import static io.airlift.bytecode.expression.BytecodeExpressions.constantNull;
-import static io.airlift.bytecode.expression.BytecodeExpressions.constantString;
 import static io.airlift.bytecode.expression.BytecodeExpressions.equal;
 import static io.airlift.bytecode.expression.BytecodeExpressions.invokeStatic;
 import static io.airlift.bytecode.expression.BytecodeExpressions.isNotNull;
 import static io.airlift.bytecode.expression.BytecodeExpressions.isNull;
 import static io.airlift.bytecode.expression.BytecodeExpressions.lessThan;
 import static io.airlift.bytecode.expression.BytecodeExpressions.newArray;
-import static io.airlift.bytecode.expression.BytecodeExpressions.newInstance;
 import static io.airlift.bytecode.expression.BytecodeExpressions.not;
 import static io.airlift.bytecode.expression.BytecodeExpressions.notEqual;
 import static io.airlift.bytecode.expression.BytecodeExpressions.or;
@@ -104,11 +101,6 @@ public static Constructor<? extends AggregationMaskBuilder> generateAggregationM
                 .ifTrue(invokeStatic(AggregationMask.class, "createSelectNone", AggregationMask.class, positionCount).ret()));
 
         Variable maskBlock = scope.declareVariable("maskBlock", body, maskBlockParameter.invoke("orElse", Object.class, constantNull(Object.class)).cast(Block.class));
-        body.append(new IfStatement()
-                .condition(maskBlock.instanceOf(LazyBlock.class))
-                .ifTrue(new BytecodeBlock()
-                        .append(newInstance(IllegalArgumentException.class, constantString("mask block must not be a LazyBlock")))
-                        .throwObject()));
         Variable hasMaskBlock = scope.declareVariable("hasMaskBlock", body, isNotNull(maskBlock));
         Variable maskBlockMayHaveNull = scope.declareVariable(
                 "maskBlockMayHaveNull",

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

@@ -82,15 +82,13 @@ public void processPage(int groupCount, int[] groupIds, Page page)
             Page arguments = page.getColumns(inputChannels);
             Optional<Block> maskBlock = Optional.empty();
             if (maskChannel.isPresent()) {
-                maskBlock = Optional.of(page.getBlock(maskChannel.getAsInt()).getLoadedBlock());
+                maskBlock = Optional.of(page.getBlock(maskChannel.getAsInt()));
             }
             AggregationMask mask = maskBuilder.buildAggregationMask(arguments, maskBlock);
 
             if (mask.isSelectNone()) {
                 return;
             }
-            // Unwrap any LazyBlock values before evaluating the accumulator
-            arguments = arguments.getLoadedPage();
             long start = System.nanoTime();
             accumulator.addInput(groupIds, arguments, mask);
             metrics.recordAccumulatorUpdateTimeSince(start);

core/trino-main/src/main/java/io/trino/operator/aggregation/builder/InMemoryHashAggregationBuilder.java

@@ -154,10 +154,10 @@ public void close() {}
     public Work<?> processPage(Page page)
     {
         if (groupedAggregators.isEmpty()) {
-            return new MeasuredGroupByHashWork<>(groupByHash.addPage(page.getLoadedPage(groupByChannels)), aggregationMetrics);
+            return new MeasuredGroupByHashWork<>(groupByHash.addPage(page.getColumns(groupByChannels)), aggregationMetrics);
         }
         return new TransformWork<>(
-                new MeasuredGroupByHashWork<>(groupByHash.getGroupIds(page.getLoadedPage(groupByChannels)), aggregationMetrics),
+                new MeasuredGroupByHashWork<>(groupByHash.getGroupIds(page.getColumns(groupByChannels)), aggregationMetrics),
                 groupByIdBlock -> {
                     int groupCount = groupByHash.getGroupCount();
                     for (GroupedAggregator groupedAggregator : groupedAggregators) {

core/trino-main/src/main/java/io/trino/operator/exchange/PageChannelSelector.java

@@ -23,9 +23,6 @@
 public class PageChannelSelector
         implements Function<Page, Page>
 {
-    // No channels need to be remapped, only ensure that all page blocks are loaded
-    private static final Function<Page, Page> GET_LOADED_PAGE = Page::getLoadedPage;
-
     private final int[] channels;
 
     public PageChannelSelector(int... channels)
@@ -37,12 +34,6 @@ public PageChannelSelector(int... channels)
     @Override
     public Page apply(Page page)
     {
-        // Ensure the channels that are emitted are fully loaded and in the correct order
-        return page.getLoadedPage(channels);
-    }
-
-    public static Function<Page, Page> identitySelection()
-    {
-        return GET_LOADED_PAGE;
+        return page.getColumns(channels);
     }
 }

core/trino-main/src/main/java/io/trino/operator/join/JoinProbe.java

@@ -41,8 +41,8 @@ public JoinProbeFactory(int[] probeOutputChannels, List<Integer> probeJoinChanne
 
         public JoinProbe createJoinProbe(Page page)
         {
-            Page probePage = page.getLoadedPage(probeJoinChannels);
-            return new JoinProbe(probeOutputChannels, page, probePage, probeHashChannel >= 0 ? page.getBlock(probeHashChannel).getLoadedBlock() : null);
+            Page probePage = page.getColumns(probeJoinChannels);
+            return new JoinProbe(probeOutputChannels, page, probePage, probeHashChannel >= 0 ? page.getBlock(probeHashChannel) : null);
         }
     }
 

core/trino-main/src/main/java/io/trino/operator/join/LookupJoinPageBuilder.java

@@ -111,7 +111,7 @@ public Page build(JoinProbe probe)
         if (!isSequentialProbeIndices || outputPositions == 0) {
             int[] probeIndices = probeIndexBuilder.toIntArray();
             for (int i = 0; i < probeOutputChannels.length; i++) {
-                blocks[i] = unwrapLoadedBlock(probePage.getBlock(probeOutputChannels[i]).getPositions(probeIndices, 0, outputPositions));
+                blocks[i] = probePage.getBlock(probeOutputChannels[i]).getPositions(probeIndices, 0, outputPositions);
             }
         }
         else {
@@ -127,7 +127,7 @@ public Page build(JoinProbe probe)
                     // only a subregion of the block should be output
                     block = block.getRegion(startRegion, outputPositions);
                 }
-                blocks[i] = unwrapLoadedBlock(block);
+                blocks[i] = block;
             }
         }
 
@@ -148,13 +148,6 @@ public String toString()
                 .toString();
     }
 
-    private static Block unwrapLoadedBlock(Block filteredProbeBlock)
-    {
-        // Lazy blocks (e.g. used in filter condition) could be loaded during filter evaluation.
-        // Unwrap them to reduce overhead of further processing.
-        return filteredProbeBlock.isLoaded() ? filteredProbeBlock.getLoadedBlock() : filteredProbeBlock;
-    }
-
     private void appendProbeIndex(JoinProbe probe)
     {
         int position = probe.getPosition();
@@ -206,7 +199,6 @@ private long getEstimatedProbeRowSize(JoinProbe probe)
         for (int index : probe.getOutputChannels()) {
             Block block = probe.getPage().getBlock(index);
             // Estimate the size of the probe row
-            // TODO: improve estimation for unloaded blocks by making it similar as in PageProcessor
             estimatedProbeRowSize += block.getSizeInBytes() / block.getPositionCount();
         }
 

core/trino-main/src/main/java/io/trino/operator/join/unspilled/JoinProbe.java

@@ -51,8 +51,8 @@ public JoinProbeFactory(List<Integer> probeOutputChannels, List<Integer> probeJo
 
         public JoinProbe createJoinProbe(Page page, LookupSource lookupSource)
         {
-            Page probePage = page.getLoadedPage(probeJoinChannels);
-            return new JoinProbe(probeOutputChannels, page, probePage, lookupSource, probeHashChannel >= 0 ? page.getBlock(probeHashChannel).getLoadedBlock() : null, hasFilter);
+            Page probePage = page.getColumns(probeJoinChannels);
+            return new JoinProbe(probeOutputChannels, page, probePage, lookupSource, probeHashChannel >= 0 ? page.getBlock(probeHashChannel) : null, hasFilter);
         }
     }
 

core/trino-main/src/main/java/io/trino/operator/join/unspilled/LookupJoinPageBuilder.java

@@ -131,7 +131,7 @@ public Page build(JoinProbe probe)
         if (!isSequentialProbeIndices || outputPositions == 0) {
             int[] probeIndices = probeIndexBuilder.toIntArray();
             for (int i = 0; i < probeOutputChannels.length; i++) {
-                blocks[i] = unwrapLoadedBlock(probePage.getBlock(probeOutputChannels[i]).getPositions(probeIndices, 0, outputPositions));
+                blocks[i] = probePage.getBlock(probeOutputChannels[i]).getPositions(probeIndices, 0, outputPositions);
             }
         }
         else {
@@ -147,7 +147,7 @@ public Page build(JoinProbe probe)
                     // only a subregion of the block should be output
                     block = block.getRegion(startRegion, outputPositions);
                 }
-                blocks[i] = unwrapLoadedBlock(block);
+                blocks[i] = block;
             }
         }
 
@@ -194,13 +194,6 @@ public String toString()
                 .toString();
     }
 
-    private static Block unwrapLoadedBlock(Block filteredProbeBlock)
-    {
-        // Lazy blocks (e.g. used in filter condition) could be loaded during filter evaluation.
-        // Unwrap them to reduce overhead of further processing.
-        return filteredProbeBlock.isLoaded() ? filteredProbeBlock.getLoadedBlock() : filteredProbeBlock;
-    }
-
     private void appendProbeIndex(JoinProbe probe)
     {
         int position = probe.getPosition();
@@ -252,7 +245,6 @@ private long getEstimatedProbeRowSize(JoinProbe probe)
         for (int index : probe.getOutputChannels()) {
             Block block = probe.getPage().getBlock(index);
             // Estimate the size of the probe row
-            // TODO: improve estimation for unloaded blocks by making it similar as in PageProcessor
             estimatedProbeRowSize += block.getSizeInBytes() / block.getPositionCount();
         }
 

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

@@ -15,7 +15,6 @@
 
 import io.trino.spi.block.Block;
 import io.trino.spi.block.DictionaryBlock;
-import io.trino.spi.block.LazyBlock;
 import io.trino.spi.block.RunLengthEncodedBlock;
 import io.trino.spi.block.ValueBlock;
 import io.trino.type.BlockTypeOperators.BlockPositionIsIdentical;
@@ -103,7 +102,6 @@ else if (state == State.DICTIONARY && this.dictionary == dictionary) {
                 transitionToDirect();
                 delegate.append(positions, valueBlock);
             }
-            case LazyBlock ignore -> throw new IllegalArgumentException("Unsupported block type: " + source.getClass().getSimpleName());
         }
     }
 
@@ -149,7 +147,6 @@ public void append(int position, Block source)
             case RunLengthEncodedBlock runLengthEncodedBlock -> delegate.append(0, runLengthEncodedBlock.getValue());
             case DictionaryBlock dictionaryBlock -> delegate.append(dictionaryBlock.getId(position), dictionaryBlock.getDictionary());
             case ValueBlock valueBlock -> delegate.append(position, valueBlock);
-            case LazyBlock ignore -> throw new IllegalArgumentException("Unsupported block type: " + source.getClass().getSimpleName());
         }
     }
 

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

@@ -56,7 +56,7 @@ public InputChannels getInputChannels()
     @Override
     public SelectedPositions filter(ConnectorSession session, SourcePage page)
     {
-        Block block = page.getBlock(0).getLoadedBlock();
+        Block block = page.getBlock(0);
 
         if (block instanceof RunLengthEncodedBlock runLengthEncodedBlock) {
             Block value = runLengthEncodedBlock.getValue();

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

@@ -180,8 +180,6 @@ public Block getResult()
 
         private void setupDictionaryBlockProjection()
         {
-            block = block.getLoadedBlock();
-
             Optional<Block> dictionary = Optional.empty();
             if (block instanceof RunLengthEncodedBlock) {
                 dictionary = Optional.of(((RunLengthEncodedBlock) block).getValue());

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

@@ -97,7 +97,7 @@ private InputChannelsSourcePage(SourcePage sourcePage, int[] channels, @Nullable
             if (eagerlyLoad != null) {
                 for (int channel = 0; channel < eagerlyLoad.length; channel++) {
                     if (eagerlyLoad[channel]) {
-                        this.blocks[channel] = sourcePage.getBlock(channels[channel]).getLoadedBlock();
+                        this.blocks[channel] = sourcePage.getBlock(channels[channel]);
                     }
                 }
             }