diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/ArrayBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/ArrayBlockEncoding.java
index 8033a9fbbfc8..805fe735290a 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/ArrayBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/ArrayBlockEncoding.java
@@ -19,6 +19,7 @@
 import static io.trino.spi.block.ArrayBlock.createArrayBlockInternal;
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
+import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
 
 public class ArrayBlockEncoding
         implements BlockEncoding
@@ -67,7 +68,12 @@ public ArrayBlock readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput sl
         int positionCount = sliceInput.readInt();
         int[] offsets = new int[positionCount + 1];
         sliceInput.readInts(offsets);
-        boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElse(null);
+        byte[] valueIsNullPacked = retrieveNullBits(sliceInput, positionCount);
+        if (valueIsNullPacked == null) {
+            return createArrayBlockInternal(0, positionCount, null, offsets, values);
+        }
+
+        boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
         return createArrayBlockInternal(0, positionCount, valueIsNull, offsets, values);
     }
 }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/ByteArrayBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/ByteArrayBlockEncoding.java
index cc03f3f18dfc..969a662d7137 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/ByteArrayBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/ByteArrayBlockEncoding.java
@@ -21,7 +21,6 @@
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
 import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
-import static java.lang.System.arraycopy;
 import static java.util.Objects.checkFromIndexSize;
 
 public class ByteArrayBlockEncoding
@@ -56,20 +55,7 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
 
         encodeNullsAsBits(sliceOutput, isNull, rawOffset, positionCount);
 
-        if (isNull == null) {
-            sliceOutput.writeBytes(rawValues, rawOffset, positionCount);
-        }
-        else {
-            byte[] valuesWithoutNull = new byte[positionCount];
-            int nonNullPositionCount = 0;
-            for (int i = 0; i < positionCount; i++) {
-                valuesWithoutNull[nonNullPositionCount] = rawValues[i + rawOffset];
-                nonNullPositionCount += isNull[i + rawOffset] ? 0 : 1;
-            }
-
-            sliceOutput.writeInt(nonNullPositionCount);
-            sliceOutput.writeBytes(valuesWithoutNull, 0, nonNullPositionCount);
-        }
+        sliceOutput.writeBytes(rawValues, rawOffset, positionCount);
     }
 
     @Override
@@ -79,42 +65,13 @@ public ByteArrayBlock readBlock(BlockEncodingSerde blockEncodingSerde, SliceInpu
 
         byte[] valueIsNullPacked = retrieveNullBits(sliceInput, positionCount);
         byte[] values = new byte[positionCount];
+        sliceInput.readBytes(Slices.wrappedBuffer(values));
 
         if (valueIsNullPacked == null) {
-            sliceInput.readBytes(Slices.wrappedBuffer(values));
             return new ByteArrayBlock(0, positionCount, null, values);
         }
         boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
 
-        int nonNullPositionCount = sliceInput.readInt();
-        sliceInput.readBytes(Slices.wrappedBuffer(values, 0, nonNullPositionCount));
-        int position = nonNullPositionCount - 1;
-
-        // Handle Last (positionCount % 8) values
-        for (int i = positionCount - 1; i >= (positionCount & ~0b111) && position >= 0; i--) {
-            values[i] = values[position];
-            if (!valueIsNull[i]) {
-                position--;
-            }
-        }
-
-        // Handle the remaining positions.
-        for (int i = (positionCount & ~0b111) - 8; i >= 0 && position >= 0; i -= 8) {
-            byte packed = valueIsNullPacked[i >>> 3];
-            if (packed == 0) { // Only values
-                arraycopy(values, position - 7, values, i, 8);
-                position -= 8;
-            }
-            else if (packed != -1) { // At least one non-null
-                for (int j = i + 7; j >= i && position >= 0; j--) {
-                    values[j] = values[position];
-                    if (!valueIsNull[j]) {
-                        position--;
-                    }
-                }
-            }
-            // Do nothing if there are only nulls
-        }
         return new ByteArrayBlock(0, positionCount, valueIsNull, values);
     }
 }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/Fixed12BlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/Fixed12BlockEncoding.java
index d25a0924fd30..8d37e589f45a 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/Fixed12BlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/Fixed12BlockEncoding.java
@@ -19,6 +19,7 @@
 
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
+import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
 import static java.util.Objects.checkFromIndexSize;
 
 public class Fixed12BlockEncoding
@@ -53,23 +54,7 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
 
         encodeNullsAsBits(sliceOutput, isNull, rawArrayOffset, positionCount);
 
-        if (isNull == null) {
-            sliceOutput.writeInts(rawValues, rawArrayOffset * 3, positionCount * 3);
-        }
-        else {
-            int[] valuesWithoutNull = new int[positionCount * 3];
-            int nonNullPositionCount = 0;
-            for (int i = 0; i < positionCount; i++) {
-                int rawIntOffset = (i + rawArrayOffset) * 3;
-                valuesWithoutNull[nonNullPositionCount] = rawValues[rawIntOffset];
-                valuesWithoutNull[nonNullPositionCount + 1] = rawValues[rawIntOffset + 1];
-                valuesWithoutNull[nonNullPositionCount + 2] = rawValues[rawIntOffset + 2];
-                nonNullPositionCount += isNull[i + rawArrayOffset] ? 0 : 3;
-            }
-
-            sliceOutput.writeInt(nonNullPositionCount / 3);
-            sliceOutput.writeInts(valuesWithoutNull, 0, nonNullPositionCount);
-        }
+        sliceOutput.writeInts(rawValues, rawArrayOffset * 3, positionCount * 3);
     }
 
     @Override
@@ -77,23 +62,16 @@ public Fixed12Block readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput
     {
         int positionCount = sliceInput.readInt();
 
-        boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElse(null);
-
+        byte[] valueIsNullPacked = retrieveNullBits(sliceInput, positionCount);
         int[] values = new int[positionCount * 3];
-        if (valueIsNull == null) {
-            sliceInput.readInts(values);
-        }
-        else {
-            int nonNullPositionCount = sliceInput.readInt();
-            sliceInput.readInts(values, 0, nonNullPositionCount * 3);
-            int position = 3 * (nonNullPositionCount - 1);
-            for (int i = positionCount - 1; i >= 0 && position >= 0; i--) {
-                System.arraycopy(values, position, values, 3 * i, 3);
-                if (!valueIsNull[i]) {
-                    position -= 3;
-                }
-            }
+        sliceInput.readInts(values);
+
+        if (valueIsNullPacked == null) {
+            return new Fixed12Block(0, positionCount, null, values);
         }
+
+        boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
+
         return new Fixed12Block(0, positionCount, valueIsNull, values);
     }
 }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/Int128ArrayBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/Int128ArrayBlockEncoding.java
index 85efe703508b..3b38de012bd4 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/Int128ArrayBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/Int128ArrayBlockEncoding.java
@@ -19,6 +19,7 @@
 
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
+import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
 import static java.util.Objects.checkFromIndexSize;
 
 public class Int128ArrayBlockEncoding
@@ -53,22 +54,7 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
 
         encodeNullsAsBits(sliceOutput, isNull, rawArrayOffset, positionCount);
 
-        if (isNull == null) {
-            sliceOutput.writeLongs(rawValues, rawArrayOffset * 2, positionCount * 2);
-        }
-        else {
-            long[] valuesWithoutNull = new long[positionCount * 2];
-            int nonNullPositionCount = 0;
-            for (int i = 0; i < positionCount; i++) {
-                int rawValuesIndex = (i + rawArrayOffset) * 2;
-                valuesWithoutNull[nonNullPositionCount] = rawValues[rawValuesIndex];
-                valuesWithoutNull[nonNullPositionCount + 1] = rawValues[rawValuesIndex + 1];
-                nonNullPositionCount += isNull[i + rawArrayOffset] ? 0 : 2;
-            }
-
-            sliceOutput.writeInt(nonNullPositionCount / 2);
-            sliceOutput.writeLongs(valuesWithoutNull, 0, nonNullPositionCount);
-        }
+        sliceOutput.writeLongs(rawValues, rawArrayOffset * 2, positionCount * 2);
     }
 
     @Override
@@ -76,23 +62,16 @@ public Int128ArrayBlock readBlock(BlockEncodingSerde blockEncodingSerde, SliceIn
     {
         int positionCount = sliceInput.readInt();
 
-        boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElse(null);
-
+        byte[] valueIsNullPacked = retrieveNullBits(sliceInput, positionCount);
         long[] values = new long[positionCount * 2];
-        if (valueIsNull == null) {
+
+        if (valueIsNullPacked == null) {
             sliceInput.readLongs(values);
+            return new Int128ArrayBlock(0, positionCount, null, values);
         }
-        else {
-            int nonNullPositionCount = sliceInput.readInt();
-            sliceInput.readLongs(values, 0, nonNullPositionCount * 2);
-            int position = 2 * (nonNullPositionCount - 1);
-            for (int i = positionCount - 1; i >= 0 && position >= 0; i--) {
-                System.arraycopy(values, position, values, 2 * i, 2);
-                if (!valueIsNull[i]) {
-                    position -= 2;
-                }
-            }
-        }
+        boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
+
+        sliceInput.readLongs(values);
 
         return new Int128ArrayBlock(0, positionCount, valueIsNull, values);
     }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/IntArrayBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/IntArrayBlockEncoding.java
index 0f76c1d49536..14eed036f63c 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/IntArrayBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/IntArrayBlockEncoding.java
@@ -20,7 +20,6 @@
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
 import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
-import static java.lang.System.arraycopy;
 import static java.util.Objects.checkFromIndexSize;
 
 public class IntArrayBlockEncoding
@@ -55,20 +54,7 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
 
         encodeNullsAsBits(sliceOutput, isNull, rawOffset, positionCount);
 
-        if (isNull == null) {
-            sliceOutput.writeInts(rawValues, rawOffset, positionCount);
-        }
-        else {
-            int[] valuesWithoutNull = new int[positionCount];
-            int nonNullPositionCount = 0;
-            for (int i = 0; i < positionCount; i++) {
-                valuesWithoutNull[nonNullPositionCount] = rawValues[i + rawOffset];
-                nonNullPositionCount += isNull[i + rawOffset] ? 0 : 1;
-            }
-
-            sliceOutput.writeInt(nonNullPositionCount);
-            sliceOutput.writeInts(valuesWithoutNull, 0, nonNullPositionCount);
-        }
+        sliceOutput.writeInts(rawValues, rawOffset, positionCount);
     }
 
     @Override
@@ -78,42 +64,13 @@ public IntArrayBlock readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput
 
         byte[] valueIsNullPacked = retrieveNullBits(sliceInput, positionCount);
         int[] values = new int[positionCount];
+        sliceInput.readInts(values);
 
         if (valueIsNullPacked == null) {
-            sliceInput.readInts(values);
             return new IntArrayBlock(0, positionCount, null, values);
         }
         boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
 
-        int nonNullPositionCount = sliceInput.readInt();
-        sliceInput.readInts(values, 0, nonNullPositionCount);
-        int position = nonNullPositionCount - 1;
-
-        // Handle Last (positionCount % 8) values
-        for (int i = positionCount - 1; i >= (positionCount & ~0b111) && position >= 0; i--) {
-            values[i] = values[position];
-            if (!valueIsNull[i]) {
-                position--;
-            }
-        }
-
-        // Handle the remaining positions.
-        for (int i = (positionCount & ~0b111) - 8; i >= 0 && position >= 0; i -= 8) {
-            byte packed = valueIsNullPacked[i >>> 3];
-            if (packed == 0) { // Only values
-                arraycopy(values, position - 7, values, i, 8);
-                position -= 8;
-            }
-            else if (packed != -1) { // At least one non-null
-                for (int j = i + 7; j >= i && position >= 0; j--) {
-                    values[j] = values[position];
-                    if (!valueIsNull[j]) {
-                        position--;
-                    }
-                }
-            }
-            // Do nothing if there are only nulls
-        }
         return new IntArrayBlock(0, positionCount, valueIsNull, values);
     }
 }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/LongArrayBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/LongArrayBlockEncoding.java
index 38759e13ef21..681f4cf994fc 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/LongArrayBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/LongArrayBlockEncoding.java
@@ -20,7 +20,6 @@
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
 import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
-import static java.lang.System.arraycopy;
 import static java.util.Objects.checkFromIndexSize;
 
 public class LongArrayBlockEncoding
@@ -55,20 +54,7 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
 
         encodeNullsAsBits(sliceOutput, isNull, rawOffset, positionCount);
 
-        if (isNull == null) {
-            sliceOutput.writeLongs(rawValues, rawOffset, positionCount);
-        }
-        else {
-            long[] valuesWithoutNull = new long[positionCount];
-            int nonNullPositionCount = 0;
-            for (int i = 0; i < positionCount; i++) {
-                valuesWithoutNull[nonNullPositionCount] = rawValues[i + rawOffset];
-                nonNullPositionCount += isNull[i + rawOffset] ? 0 : 1;
-            }
-
-            sliceOutput.writeInt(nonNullPositionCount);
-            sliceOutput.writeLongs(valuesWithoutNull, 0, nonNullPositionCount);
-        }
+        sliceOutput.writeLongs(rawValues, rawOffset, positionCount);
     }
 
     @Override
@@ -85,35 +71,8 @@ public LongArrayBlock readBlock(BlockEncodingSerde blockEncodingSerde, SliceInpu
         }
         boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
 
-        int nonNullPositionCount = sliceInput.readInt();
-        sliceInput.readLongs(values, 0, nonNullPositionCount);
-        int position = nonNullPositionCount - 1;
+        sliceInput.readLongs(values);
 
-        // Handle Last (positionCount % 8) values
-        for (int i = positionCount - 1; i >= (positionCount & ~0b111) && position >= 0; i--) {
-            values[i] = values[position];
-            if (!valueIsNull[i]) {
-                position--;
-            }
-        }
-
-        // Handle the remaining positions.
-        for (int i = (positionCount & ~0b111) - 8; i >= 0 && position >= 0; i -= 8) {
-            byte packed = valueIsNullPacked[i >>> 3];
-            if (packed == 0) { // Only values
-                arraycopy(values, position - 7, values, i, 8);
-                position -= 8;
-            }
-            else if (packed != -1) { // At least one non-null
-                for (int j = i + 7; j >= i && position >= 0; j--) {
-                    values[j] = values[position];
-                    if (!valueIsNull[j]) {
-                        position--;
-                    }
-                }
-            }
-            // Do nothing if there are only nulls
-        }
         return new LongArrayBlock(0, positionCount, valueIsNull, values);
     }
 }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/ShortArrayBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/ShortArrayBlockEncoding.java
index 512164c605d5..add11f2b99bd 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/ShortArrayBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/ShortArrayBlockEncoding.java
@@ -20,7 +20,6 @@
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
 import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
-import static java.lang.System.arraycopy;
 import static java.util.Objects.checkFromIndexSize;
 
 public class ShortArrayBlockEncoding
@@ -55,20 +54,7 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
 
         encodeNullsAsBits(sliceOutput, isNull, rawOffset, positionCount);
 
-        if (isNull == null) {
-            sliceOutput.writeShorts(rawValues, rawOffset, positionCount);
-        }
-        else {
-            short[] valuesWithoutNull = new short[positionCount];
-            int nonNullPositionCount = 0;
-            for (int i = 0; i < positionCount; i++) {
-                valuesWithoutNull[nonNullPositionCount] = rawValues[i + rawOffset];
-                nonNullPositionCount += isNull[i + rawOffset] ? 0 : 1;
-            }
-
-            sliceOutput.writeInt(nonNullPositionCount);
-            sliceOutput.writeShorts(valuesWithoutNull, 0, nonNullPositionCount);
-        }
+        sliceOutput.writeShorts(rawValues, rawOffset, positionCount);
     }
 
     @Override
@@ -85,35 +71,8 @@ public ShortArrayBlock readBlock(BlockEncodingSerde blockEncodingSerde, SliceInp
         }
         boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
 
-        int nonNullPositionCount = sliceInput.readInt();
-        sliceInput.readShorts(values, 0, nonNullPositionCount);
-        int position = nonNullPositionCount - 1;
+        sliceInput.readShorts(values);
 
-        // Handle Last (positionCount % 8) values
-        for (int i = positionCount - 1; i >= (positionCount & ~0b111) && position >= 0; i--) {
-            values[i] = values[position];
-            if (!valueIsNull[i]) {
-                position--;
-            }
-        }
-
-        // Handle the remaining positions.
-        for (int i = (positionCount & ~0b111) - 8; i >= 0 && position >= 0; i -= 8) {
-            byte packed = valueIsNullPacked[i >>> 3];
-            if (packed == 0) { // Only values
-                arraycopy(values, position - 7, values, i, 8);
-                position -= 8;
-            }
-            else if (packed != -1) { // At least one non-null
-                for (int j = i + 7; j >= i && position >= 0; j--) {
-                    values[j] = values[position];
-                    if (!valueIsNull[j]) {
-                        position--;
-                    }
-                }
-            }
-            // Do nothing if there are only nulls
-        }
         return new ShortArrayBlock(0, positionCount, valueIsNull, values);
     }
 }
diff --git a/core/trino-spi/src/main/java/io/trino/spi/block/VariableWidthBlockEncoding.java b/core/trino-spi/src/main/java/io/trino/spi/block/VariableWidthBlockEncoding.java
index 8f0e4c1bff3b..2c67213c4bae 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/block/VariableWidthBlockEncoding.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/block/VariableWidthBlockEncoding.java
@@ -19,11 +19,9 @@
 import io.airlift.slice.Slices;
 import jakarta.annotation.Nullable;
 
-import java.util.Arrays;
-
 import static io.trino.spi.block.EncoderUtil.decodeNullBits;
 import static io.trino.spi.block.EncoderUtil.encodeNullsAsBits;
-import static java.lang.String.format;
+import static io.trino.spi.block.EncoderUtil.retrieveNullBits;
 import static java.util.Objects.checkFromIndexSize;
 
 public class VariableWidthBlockEncoding
@@ -57,21 +55,18 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
         int[] rawOffsets = variableWidthBlock.getRawOffsets();
         checkFromIndexSize(arrayBaseOffset, positionCount + 1, rawOffsets.length);
 
-        // lengths
-        int[] lengths = new int[positionCount];
         int totalLength = 0;
-        int nonNullsCount = 0;
+        int firstValueOffset = rawOffsets[arrayBaseOffset];
+
+        int[] adjustedOffsets = new int[positionCount + 1];
 
         for (int position = 0; position < positionCount; position++) {
-            int length = rawOffsets[position + arrayBaseOffset + 1] - rawOffsets[position + arrayBaseOffset];
-            totalLength += length;
-            lengths[nonNullsCount] = length;
-            nonNullsCount += isNull != null && isNull[position + arrayBaseOffset] ? 0 : 1;
+            totalLength += rawOffsets[position + arrayBaseOffset + 1] - rawOffsets[position + arrayBaseOffset];
+            adjustedOffsets[position] = rawOffsets[position + arrayBaseOffset] - firstValueOffset;
         }
+        adjustedOffsets[positionCount] = rawOffsets[positionCount + arrayBaseOffset] - firstValueOffset;
 
-        sliceOutput
-                .appendInt(nonNullsCount)
-                .writeInts(lengths, 0, nonNullsCount);
+        sliceOutput.writeInts(adjustedOffsets);
 
         encodeNullsAsBits(sliceOutput, isNull, arrayBaseOffset, positionCount);
 
@@ -84,61 +79,23 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
     public Block readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput sliceInput)
     {
         int positionCount = sliceInput.readInt();
-        int nonNullsCount = sliceInput.readInt();
-
-        if (nonNullsCount > positionCount) {
-            throw new IllegalArgumentException(format("nonNullsCount must be <= positionCount, found: %s > %s", nonNullsCount, positionCount));
-        }
 
         int[] offsets = new int[positionCount + 1];
         // Read the lengths array into the end of the offsets array, since nonNullsCount <= positionCount
-        int lengthIndex = offsets.length - nonNullsCount;
-        sliceInput.readInts(offsets, lengthIndex, nonNullsCount);
-
-        boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElse(null);
-        // Transform lengths back to offsets
-        if (valueIsNull == null) {
-            if (positionCount != nonNullsCount || lengthIndex != 1) {
-                throw new IllegalArgumentException(format("nonNullsCount must equal positionCount, found: %s <> %s", nonNullsCount, positionCount));
-            }
-            // Simplified loop for no nulls present
-            for (int i = 1; i < offsets.length; i++) {
-                offsets[i] += offsets[i - 1];
-            }
-        }
-        else {
-            computeOffsetsFromLengths(offsets, valueIsNull, lengthIndex);
-        }
+        sliceInput.readInts(offsets);
+
+        byte[] valueIsNullPacked = retrieveNullBits(sliceInput, positionCount);
 
         int blockSize = sliceInput.readInt();
         Slice slice = Slices.allocate(blockSize);
         sliceInput.readBytes(slice);
 
-        return new VariableWidthBlock(0, positionCount, slice, offsets, valueIsNull);
-    }
-
-    private static void computeOffsetsFromLengths(int[] offsets, boolean[] valueIsNull, int lengthIndex)
-    {
-        if (lengthIndex < 0 || lengthIndex > offsets.length) {
-            throw new IllegalArgumentException(format("Invalid lengthIndex %s for offsets %s", lengthIndex, offsets.length));
-        }
-        int currentOffset = 0;
-        for (int i = 1; i < offsets.length; i++) {
-            if (lengthIndex == offsets.length) {
-                // Populate remaining null elements
-                Arrays.fill(offsets, i, offsets.length, currentOffset);
-                break;
-            }
-            boolean isNull = valueIsNull[i - 1];
-            // must be accessed unconditionally, otherwise CMOV optimization isn't applied due to
-            // ArrayIndexOutOfBoundsException checks
-            int length = offsets[lengthIndex];
-            lengthIndex += isNull ? 0 : 1;
-            currentOffset += isNull ? 0 : length;
-            offsets[i] = currentOffset;
-        }
-        if (lengthIndex != offsets.length) {
-            throw new IllegalArgumentException(format("Failed to consume all length entries, found %s <> %s", lengthIndex, offsets.length));
+        if (valueIsNullPacked == null) {
+            return new VariableWidthBlock(0, positionCount, slice, offsets, null);
         }
+
+        boolean[] valueIsNull = decodeNullBits(valueIsNullPacked, positionCount);
+
+        return new VariableWidthBlock(0, positionCount, slice, offsets, valueIsNull);
     }
 }