diff --git a/paimon-common/src/main/java/org/apache/paimon/types/InternalRowToSizeVisitor.java b/paimon-common/src/main/java/org/apache/paimon/types/InternalRowToSizeVisitor.java
index dbac55a07dde..3129be9a8769 100644
--- a/paimon-common/src/main/java/org/apache/paimon/types/InternalRowToSizeVisitor.java
+++ b/paimon-common/src/main/java/org/apache/paimon/types/InternalRowToSizeVisitor.java
@@ -18,6 +18,11 @@
 
 package org.apache.paimon.types;
 
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.BlobRef;
+import org.apache.paimon.data.BlobView;
 import org.apache.paimon.data.DataGetters;
 import org.apache.paimon.data.InternalArray;
 import org.apache.paimon.data.InternalMap;
@@ -32,6 +37,7 @@ public class InternalRowToSizeVisitor
         implements DataTypeVisitor<BiFunction<DataGetters, Integer, Integer>> {
 
     public static final int NULL_SIZE = 0;
+    private static final int UNKNOWN_SIZE = 1;
 
     @Override
     public BiFunction<DataGetters, Integer, Integer> visit(CharType charType) {
@@ -229,11 +235,26 @@ public BiFunction<DataGetters, Integer, Integer> visit(BlobType blobType) {
             if (row.isNullAt(index)) {
                 return NULL_SIZE;
             } else {
-                return Math.toIntExact(row.getVariant(index).sizeInBytes());
+                Blob blob = row.getBlob(index);
+                if (blob instanceof BlobData) {
+                    return ((BlobData) blob).toData().length;
+                } else if (blob instanceof BlobRef) {
+                    return descriptorLength(blob.toDescriptor());
+                } else if (blob instanceof BlobView) {
+                    BlobView view = (BlobView) blob;
+                    return view.isResolved()
+                            ? descriptorLength(view.toDescriptor())
+                            : view.viewStruct().serialize().length;
+                }
+                return UNKNOWN_SIZE;
             }
         };
     }
 
+    private static int descriptorLength(BlobDescriptor descriptor) {
+        return descriptor.length() < 0 ? UNKNOWN_SIZE : Math.toIntExact(descriptor.length());
+    }
+
     @Override
     public BiFunction<DataGetters, Integer, Integer> visit(ArrayType arrayType) {
         return (row, index) -> {
diff --git a/paimon-common/src/test/java/org/apache/paimon/types/InternalRowToSizeVisitorTest.java b/paimon-common/src/test/java/org/apache/paimon/types/InternalRowToSizeVisitorTest.java
index cfdae649c190..15c198774c4b 100644
--- a/paimon-common/src/test/java/org/apache/paimon/types/InternalRowToSizeVisitorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/types/InternalRowToSizeVisitorTest.java
@@ -18,13 +18,18 @@
 
 package org.apache.paimon.types;
 
+import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.BlobViewStruct;
 import org.apache.paimon.data.DataGetters;
 import org.apache.paimon.data.Decimal;
 import org.apache.paimon.data.GenericArray;
 import org.apache.paimon.data.GenericMap;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.utils.UriReader;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -192,4 +197,47 @@ void testCalculatorSize() {
 
         Assertions.assertThat(feildSizeCalculator.get(23).apply(row, 23)).isEqualTo(0);
     }
+
+    @Test
+    void testBlobSize() {
+        Assertions.assertThat(blobSize(Blob.fromData(new byte[] {1, 2, 3}))).isEqualTo(3);
+        Assertions.assertThat(blobSize(null)).isEqualTo(0);
+    }
+
+    @Test
+    void testBlobRefSize() {
+        Assertions.assertThat(
+                        blobSize(
+                                Blob.fromDescriptor(
+                                        UriReader.fromHttp(),
+                                        new BlobDescriptor("https://example.com/blob", 0, 3))))
+                .isEqualTo(3);
+        Assertions.assertThat(blobSize(Blob.fromHttp("https://example.com/blob"))).isEqualTo(1);
+    }
+
+    @Test
+    void testUnresolvedBlobViewSize() {
+        BlobViewStruct viewStruct = new BlobViewStruct(Identifier.create("db", "t"), 1, 2L);
+        Assertions.assertThat(blobSize(Blob.fromView(viewStruct)))
+                .isEqualTo(viewStruct.serialize().length);
+    }
+
+    @Test
+    void testBlobStreamSize() {
+        Assertions.assertThat(
+                        blobSize(
+                                Blob.fromInputStream(
+                                        () -> {
+                                            throw new AssertionError();
+                                        })))
+                .isEqualTo(1);
+    }
+
+    private int blobSize(Blob blob) {
+        RowType rowType = RowType.builder().field("b", DataTypes.BLOB()).build();
+        InternalRowToSizeVisitor visitor = new InternalRowToSizeVisitor();
+        BiFunction<DataGetters, Integer, Integer> calculator =
+                rowType.getFieldTypes().get(0).accept(visitor);
+        return calculator.apply(GenericRow.of(blob), 0);
+    }
 }