diff --git a/fluss-common/src/test/java/org/apache/fluss/row/encode/paimon/PaimonKeyEncoderTest.java b/fluss-common/src/test/java/org/apache/fluss/row/encode/paimon/PaimonKeyEncoderTest.java
index e770d96de5..7925b5657e 100644
--- a/fluss-common/src/test/java/org/apache/fluss/row/encode/paimon/PaimonKeyEncoderTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/row/encode/paimon/PaimonKeyEncoderTest.java
@@ -109,8 +109,8 @@ private BinaryRow genPaimonRowForAllTypes(int arity) {
                 (int)
                         TypeUtils.castFromString(
                                 "09:30:00.0", org.apache.fluss.types.DataTypes.TIME()));
-        binaryRowWriter.writeBinary(9, "1234567890".getBytes());
-        binaryRowWriter.writeBinary(10, "20".getBytes());
+        binaryRowWriter.writeBinary(9, "1234567890".getBytes(), 0, "1234567890".getBytes().length);
+        binaryRowWriter.writeBinary(10, "20".getBytes(), 0, "20".getBytes().length);
         binaryRowWriter.writeString(11, BinaryString.fromString("1"));
         binaryRowWriter.writeString(12, BinaryString.fromString("hello"));
         binaryRowWriter.writeDecimal(13, Decimal.fromUnscaledLong(9, 5, 2), 5);
diff --git a/fluss-lake/fluss-lake-paimon/pom.xml b/fluss-lake/fluss-lake-paimon/pom.xml
index a1e6d1abca..f56e35aef5 100644
--- a/fluss-lake/fluss-lake-paimon/pom.xml
+++ b/fluss-lake/fluss-lake-paimon/pom.xml
@@ -32,10 +32,6 @@
 
     <packaging>jar</packaging>
 
-    <properties>
-        <paimon.version>1.3.1</paimon.version>
-    </properties>
-
     <dependencies>
         <dependency>
             <groupId>org.apache.fluss</groupId>
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussArrayAsPaimonArray.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussArrayAsPaimonArray.java
index 0552bb4dea..c815f03c84 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussArrayAsPaimonArray.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussArrayAsPaimonArray.java
@@ -21,10 +21,12 @@
 import org.apache.fluss.row.TimestampNtz;
 
 import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
 import org.apache.paimon.data.Decimal;
 import org.apache.paimon.data.InternalArray;
 import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalVector;
 import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.data.variant.Variant;
 import org.apache.paimon.types.ArrayType;
@@ -149,6 +151,12 @@ public Variant getVariant(int pos) {
                 "getVariant is not supported for Fluss array currently.");
     }
 
+    @Override
+    public Blob getBlob(int pos) {
+        throw new UnsupportedOperationException(
+                "getBlob is not supported for Fluss array currently.");
+    }
+
     @Override
     public InternalArray getArray(int pos) {
         org.apache.fluss.row.InternalArray innerArray = flussArray.getArray(pos);
@@ -158,6 +166,12 @@ public InternalArray getArray(int pos) {
                         innerArray, ((ArrayType) elementType).getElementType());
     }
 
+    @Override
+    public InternalVector getVector(int pos) {
+        throw new UnsupportedOperationException(
+                "getVector is not supported for Fluss array currently.");
+    }
+
     @Override
     public InternalMap getMap(int pos) {
         org.apache.fluss.row.InternalMap flussMap = flussArray.getMap(pos);
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRow.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRow.java
index 0b0b9537c4..f20d6437ec 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRow.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/FlussRowAsPaimonRow.java
@@ -21,10 +21,12 @@
 import org.apache.fluss.row.TimestampNtz;
 
 import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
 import org.apache.paimon.data.Decimal;
 import org.apache.paimon.data.InternalArray;
 import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalVector;
 import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.data.variant.Variant;
 import org.apache.paimon.types.ArrayType;
@@ -158,6 +160,12 @@ public Variant getVariant(int i) {
                 "getVariant is not support for Fluss record currently.");
     }
 
+    @Override
+    public Blob getBlob(int pos) {
+        throw new UnsupportedOperationException(
+                "getBlob is not support for Fluss record currently.");
+    }
+
     @Override
     public InternalArray getArray(int pos) {
         org.apache.fluss.row.InternalArray flussArray = internalRow.getArray(pos);
@@ -167,6 +175,12 @@ public InternalArray getArray(int pos) {
                         flussArray, ((ArrayType) tableRowType.getTypeAt(pos)).getElementType());
     }
 
+    @Override
+    public InternalVector getVector(int pos) {
+        throw new UnsupportedOperationException(
+                "getVector is not support for Fluss record currently.");
+    }
+
     @Override
     public InternalMap getMap(int pos) {
         org.apache.fluss.row.InternalMap flussMap = internalRow.getMap(pos);
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java
index 339ec2b2b5..371da5af99 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java
@@ -31,10 +31,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestCommittable;
-import org.apache.paimon.manifest.ManifestEntry;
-import org.apache.paimon.manifest.SimpleFileEntry;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.TableSnapshot;
 import org.apache.paimon.table.sink.CommitCallback;
@@ -287,12 +284,8 @@ private FileStoreTable getTable(TablePath tablePath, boolean isAutoSnapshotExpir
     public static class PaimonCommitCallback implements CommitCallback {
 
         @Override
-        public void call(
-                List<SimpleFileEntry> baseFiles,
-                List<ManifestEntry> deltaFiles,
-                List<IndexManifestEntry> indexFiles,
-                Snapshot snapshot) {
-            currentCommitSnapshotId.set(snapshot.id());
+        public void call(Context context) {
+            currentCommitSnapshotId.set(context.snapshot.id());
         }
 
         @Override
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
index a8e395725a..d6593311f3 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/paimon/arrow/converter/Arrow2PaimonVectorConverter.java
@@ -64,6 +64,7 @@
 import org.apache.paimon.types.ArrayType;
 import org.apache.paimon.types.BigIntType;
 import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.BlobType;
 import org.apache.paimon.types.BooleanType;
 import org.apache.paimon.types.CharType;
 import org.apache.paimon.types.DataField;
@@ -85,6 +86,7 @@
 import org.apache.paimon.types.VarBinaryType;
 import org.apache.paimon.types.VarCharType;
 import org.apache.paimon.types.VariantType;
+import org.apache.paimon.types.VectorType;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -478,6 +480,11 @@ public Arrow2PaimonVectorConverter visit(VariantType variantType) {
             throw new UnsupportedOperationException();
         }
 
+        @Override
+        public Arrow2PaimonVectorConverter visit(BlobType blobType) {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         public Arrow2PaimonVectorConverter visit(ArrayType arrayType) {
             final Arrow2PaimonVectorConverter arrowVectorConvertor =
@@ -519,6 +526,11 @@ public ColumnVector getColumnVector() {
                     };
         }
 
+        @Override
+        public Arrow2PaimonVectorConverter visit(VectorType vectorType) {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         public Arrow2PaimonVectorConverter visit(MultisetType multisetType) {
             throw new UnsupportedOperationException("Doesn't support MultisetType.");
diff --git a/pom.xml b/pom.xml
index d39a91d9b3..271f769a21 100644
--- a/pom.xml
+++ b/pom.xml
@@ -102,7 +102,7 @@
         <curator.version>5.4.0</curator.version>
         <netty.version>4.1.104.Final</netty.version>
         <arrow.version>15.0.0</arrow.version>
-        <paimon.version>1.3.1</paimon.version>
+        <paimon.version>1.4.2</paimon.version>
         <iceberg.version>1.10.1</iceberg.version>
         <hudi.version>1.1.0</hudi.version>
         <roaringbitmap.version>1.3.0</roaringbitmap.version>