diff --git a/core/trino-main/src/test/java/io/trino/execution/buffer/TestPagesSerde.java b/core/trino-main/src/test/java/io/trino/execution/buffer/TestPagesSerde.java
index 77a7085215b2..59de15c34c63 100644
--- a/core/trino-main/src/test/java/io/trino/execution/buffer/TestPagesSerde.java
+++ b/core/trino-main/src/test/java/io/trino/execution/buffer/TestPagesSerde.java
@@ -200,14 +200,14 @@ public void testVarcharSerializedSize()
         // empty page
         Page page = new Page(builder.build());
         int pageSize = serializedSize(ImmutableList.of(VARCHAR), page);
-        assertEquals(pageSize, 44);
+        assertEquals(pageSize, 48);
 
         // page with one value
         VARCHAR.writeString(builder, "alice");
         pageSize = 44; // Now we have moved to the normal block implementation so the page size overhead is 44
         page = new Page(builder.build());
         int firstValueSize = serializedSize(ImmutableList.of(VARCHAR), page) - pageSize;
-        assertEquals(firstValueSize, 4 + 5); // length + "alice"
+        assertEquals(firstValueSize, 8 + 5); // length + nonNullsCount + "alice"
 
         // page with two values
         VARCHAR.writeString(builder, "bob");
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 fea15b7aec70..dcb137eacbf7 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
@@ -42,14 +42,22 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
         int positionCount = variableWidthBlock.getPositionCount();
         sliceOutput.appendInt(positionCount);
 
-        // offsets
+        // lengths
+        int[] lengths = new int[positionCount];
         int totalLength = 0;
+        int nonNullsCount = 0;
+
         for (int position = 0; position < positionCount; position++) {
             int length = variableWidthBlock.getSliceLength(position);
             totalLength += length;
-            sliceOutput.appendInt(totalLength);
+            lengths[nonNullsCount] = length;
+            nonNullsCount += variableWidthBlock.isNull(position) ? 0 : 1;
         }
 
+        sliceOutput
+                .appendInt(nonNullsCount)
+                .writeBytes(Slices.wrappedIntArray(lengths, 0, nonNullsCount));
+
         encodeNullsAsBits(sliceOutput, variableWidthBlock);
 
         sliceOutput
@@ -61,9 +69,11 @@ public void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceO
     public Block readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput sliceInput)
     {
         int positionCount = sliceInput.readInt();
+        int nonNullsCount = sliceInput.readInt();
 
         int[] offsets = new int[positionCount + 1];
-        sliceInput.readBytes(Slices.wrappedIntArray(offsets), SIZE_OF_INT, positionCount * SIZE_OF_INT);
+        int[] lengths = new int[nonNullsCount];
+        sliceInput.readBytes(Slices.wrappedIntArray(lengths), 0, nonNullsCount * SIZE_OF_INT);
 
         boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElse(null);
 
@@ -71,6 +81,16 @@ public Block readBlock(BlockEncodingSerde blockEncodingSerde, SliceInput sliceIn
         Slice slice = Slices.allocate(blockSize);
         sliceInput.readBytes(slice);
 
+        int nonNullPosition = 0;
+        int offset = 0;
+
+        for (int i = 0; i < positionCount; i++) {
+            if (valueIsNull == null || !valueIsNull[i]) {
+                offset += lengths[nonNullPosition];
+                nonNullPosition++;
+            }
+            offsets[i + 1] = offset;
+        }
         return new VariableWidthBlock(0, positionCount, slice, offsets, valueIsNull);
     }
 }
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/optimizer/TestHivePlans.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/optimizer/TestHivePlans.java
index 831317930dcc..ca4bd44b5726 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/optimizer/TestHivePlans.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/optimizer/TestHivePlans.java
@@ -129,7 +129,7 @@ public void testPruneSimplePartitionLikeFilter()
         assertDistributedPlan(
                 "SELECT * FROM table_str_partitioned WHERE str_part LIKE 't%'",
                 output(
-                        filter("\"$like\"(STR_PART, \"$literal$\"(from_base64('DgAAAFZBUklBQkxFX1dJRFRIAQAAAAcAAAAABwAAAAIAAAB0JQA=')))",
+                        filter("\"$like\"(STR_PART, \"$literal$\"(from_base64('DgAAAFZBUklBQkxFX1dJRFRIAQAAAAEAAAAHAAAAAAcAAAACAAAAdCUA')))",
                                 tableScan("table_str_partitioned", Map.of("INT_COL", "int_col", "STR_PART", "str_part")))));
     }
 
@@ -151,12 +151,12 @@ public void testPrunePartitionLikeFilter()
                                 .left(
                                         exchange(REMOTE, REPARTITION,
                                                 project(
-                                                        filter("\"$like\"(L_STR_PART, \"$literal$\"(from_base64('DgAAAFZBUklBQkxFX1dJRFRIAQAAAAcAAAAABwAAAAIAAAB0JQA=')))",
+                                                        filter("\"$like\"(L_STR_PART, \"$literal$\"(from_base64('DgAAAFZBUklBQkxFX1dJRFRIAQAAAAEAAAAHAAAAAAcAAAACAAAAdCUA')))",
                                                                 tableScan("table_str_partitioned", Map.of("L_INT_COL", "int_col", "L_STR_PART", "str_part"))))))
                                 .right(exchange(LOCAL,
                                         exchange(REMOTE, REPARTITION,
                                                 project(
-                                                        filter("R_STR_COL IN ('three', CAST('two' AS varchar(5))) AND \"$like\"(R_STR_COL, \"$literal$\"(from_base64('DgAAAFZBUklBQkxFX1dJRFRIAQAAAAcAAAAABwAAAAIAAAB0JQA=')))",
+                                                        filter("R_STR_COL IN ('three', CAST('two' AS varchar(5))) AND \"$like\"(R_STR_COL, \"$literal$\"(from_base64('DgAAAFZBUklBQkxFX1dJRFRIAQAAAAEAAAAHAAAAAAcAAAACAAAAdCUA')))",
                                                                 tableScan("table_unpartitioned", Map.of("R_STR_COL", "str_col", "R_INT_COL", "int_col"))))))))));
     }
 
diff --git a/testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/BaseFaultTolerantExecutionTest.java b/testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/BaseFaultTolerantExecutionTest.java
index 3c71aa0ec448..f0a3636954ce 100644
--- a/testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/BaseFaultTolerantExecutionTest.java
+++ b/testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/BaseFaultTolerantExecutionTest.java
@@ -66,7 +66,7 @@ public void testExecutePreferredWritePartitioningSkewMitigation()
     {
         @Language("SQL") String createTableSql = """
                 CREATE TABLE test_execute_skew_mitigation WITH (%s = ARRAY['returnflag']) AS
-                SELECT orderkey, partkey, suppkey, linenumber, quantity, extendedprice, discount, tax, linestatus, shipdate, commitdate, receiptdate, shipinstruct, shipmode, comment, returnflag
+                SELECT orderkey, partkey, suppkey, linenumber, quantity, extendedprice, discount, tax, linestatus, shipdate, commitdate, receiptdate, shipinstruct, shipmode, returnflag
                 FROM tpch.sf1.lineitem
                 WHERE returnflag = 'N'
                 LIMIT 1000000""".formatted(partitioningTablePropertyName);