Fix potential data inconsistency when widen pk column type if pk is h…

…andle (#3568) (#3572)

dbms/src/Storages/Transaction/RegionBlockReader.cpp

@@ -352,7 +352,25 @@ bool setColumnValues(ColumnUInt8 & delmark_col,
             }
         }
         else
-            column_map.getMutableColumnPtr(pk_column_ids[0])->insert(Field(static_cast<Int64>(pk)));
+        {
+            // The pk_type must be Int32/Uint32 or more narrow type
+            // so cannot tell its' exact type here, just use `insert(Field)`
+            HandleID handle_value(static_cast<Int64>(pk));
+            auto & pk_column = column_map.getMutableColumnPtr(pk_column_ids[0]);
+            pk_column->insert(Field(handle_value));
+            if (unlikely(pk_column->getInt(index) != handle_value))
+            {
+                if (!force_decode)
+                {
+                    return false;
+                }
+                else
+                {
+                    throw Exception("Detected overflow value when decoding pk column of type " + pk_column->getName(),
+                                    ErrorCodes::LOGICAL_ERROR);
+                }
+            }
+        }
         index++;
     }
     decoded_data.checkValid();

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 |
++---------------------+