diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 04fac9cc873..a8af7e69ca5 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -101,12 +101,13 @@ void ReorderRegionDataReadList(RegionDataReadInfoList & data_list)
 }
 
 template <TMTPKType pk_type>
-void setPKVersionDel(ColumnUInt8 & delmark_col,
+bool setPKVersionDel(ColumnUInt8 & delmark_col,
     ColumnUInt64 & version_col,
     MutableColumnPtr & pk_column,
     const RegionDataReadInfoList & data_list,
     const Timestamp tso,
-    RegionScanFilterPtr scan_filter)
+    RegionScanFilterPtr scan_filter,
+    bool force_decode[[maybe_unused]])
 {
     ColumnUInt8::Container & delmark_data = delmark_col.getData();
     ColumnUInt64::Container & version_data = version_col.getData();
@@ -114,6 +115,7 @@ void setPKVersionDel(ColumnUInt8 & delmark_col,
     delmark_data.reserve(data_list.size());
     version_data.reserve(data_list.size());
 
+    size_t index = 0;
     for (const auto & [handle, write_type, commit_ts, value] : data_list)
     {
         std::ignore = value;
@@ -142,8 +144,24 @@ void setPKVersionDel(ColumnUInt8 & delmark_col,
         else if constexpr (pk_type == TMTPKType::UINT64)
             typeid_cast<ColumnVector<UInt64> &>(*pk_column).insert(static_cast<UInt64>(handle));
         else
-            pk_column->insert(Field(static_cast<Int64>(handle)));
+        {
+            pk_column->insert(Field(handle));
+            if (unlikely(pk_column->getInt(index) != handle))
+            {
+                if (!force_decode)
+                {
+                    return false;
+                }
+                else
+                {
+                    throw Exception("Detected overflow value when decoding pk column of type " + pk_column->getName(),
+                                    ErrorCodes::LOGICAL_ERROR);
+                }
+            }
+        }
+        index++;
     }
+    return true;
 }
 
 RegionBlockReader::RegionBlockReader(const ManageableStoragePtr & storage)
@@ -253,7 +271,8 @@ std::tuple<Block, bool> RegionBlockReader::read(const Names & column_names_to_re
                 break;
         }
 
-        func(*delmark_col, *version_col, column_map.getMutableColumnPtr(handle_col_id), data_list, start_ts, scan_filter);
+        if (!func(*delmark_col, *version_col, column_map.getMutableColumnPtr(handle_col_id), data_list, start_ts, scan_filter, force_decode))
+            return std::make_tuple(Block(), false);
     }
 
     // optimize for only need handle, tso, delmark.
diff --git a/tests/fullstack-test2/ddl/widen_pk.test b/tests/fullstack-test2/ddl/widen_pk.test
new file mode 100644
index 00000000000..2fdb5ec98ba
--- /dev/null
+++ b/tests/fullstack-test2/ddl/widen_pk.test
@@ -0,0 +1,28 @@
+mysql> drop table if exists test.t
+mysql> create table test.t(a int primary key)
+mysql> alter table test.t set tiflash replica 1
+
+func> wait_table test t
+
+mysql> insert into test.t values(1);
+
+mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t;
++---+
+| a |
++---+
+| 1 |
++---+
+
+>> DBGInvoke __enable_schema_sync_service('false')
+
+mysql> alter table test.t modify column a bigint;
+
+mysql> insert into test.t values(9223372036854775807);
+
+mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t;
++---------------------+
+| a                   |
++---------------------+
+|                   1 |
+| 9223372036854775807 |
++---------------------+