ddl: Fix NULL value in non-nullable column (#8722)

dbms/src/Storages/KVStore/Decode/RegionBlockReader.cpp

@@ -244,11 +244,10 @@ bool RegionBlockReader::readImpl(Block & block, const RegionDataReadInfoList & d
                         else
                         {
                             throw Exception(
-                                fmt::format(
-                                    "Detected overflow value when decoding pk column, type={} handle={}",
-                                    raw_pk_column->getName(),
-                                    handle_value),
-                                ErrorCodes::LOGICAL_ERROR);
+                                ErrorCodes::LOGICAL_ERROR,
+                                "Detected overflow value when decoding pk column, type={} handle={}",
+                                raw_pk_column->getName(),
+                                handle_value);
                         }
                     }
                 }

dbms/src/Storages/KVStore/tests/gtest_region_block_reader.cpp

@@ -51,6 +51,7 @@ class RegionBlockReaderTest : public ::testing::Test
 
     RegionDataReadInfoList data_list_read;
     std::unordered_map<ColumnID, Field> fields_map;
+    std::unordered_set<ColumnID> invalid_null_column_ids;
 
     LoggerPtr logger;
 
@@ -175,9 +176,18 @@ class RegionBlockReaderTest : public ::testing::Test
                 }
                 else
                 {
-                    if (fields_map.count(column_element.column_id) > 0)
-                        ASSERT_FIELD_EQ((*column_element.column)[row], fields_map.at(column_element.column_id))
-                            << gen_error_log();
+                    if (fields_map.contains(column_element.column_id))
+                    {
+                        if (!invalid_null_column_ids.contains(column_element.column_id))
+                        {
+                            ASSERT_FIELD_EQ((*column_element.column)[row], fields_map.at(column_element.column_id))
+                                << gen_error_log();
+                        }
+                        else
+                        {
+                            ASSERT_FIELD_EQ((*column_element.column)[row], UInt64(0));
+                        }
+                    }
                     else
                         LOG_INFO(
                             logger,
@@ -411,22 +421,27 @@ try
     encodeColumns(table_info, fields, RowEncodeVersion::RowV2);
 
     auto new_table_info = getTableInfoFieldsForInvalidNULLTest({EXTRA_HANDLE_COLUMN_ID}, false);
+    invalid_null_column_ids.emplace(11);
     ASSERT_TRUE(new_table_info.getColumnInfo(11).hasNotNullFlag()); // col 11 is not null
 
     auto new_decoding_schema = getDecodingStorageSchemaSnapshot(new_table_info);
     ASSERT_FALSE(decodeAndCheckColumns(new_decoding_schema, false));
-    ASSERT_ANY_THROW(decodeAndCheckColumns(new_decoding_schema, true));
+    ASSERT_TRUE(decodeAndCheckColumns(new_decoding_schema, true));
 }
 CATCH
 
 TEST_F(RegionBlockReaderTest, InvalidNULLRowV1)
 {
     auto [table_info, fields] = getNormalTableInfoFields({EXTRA_HANDLE_COLUMN_ID}, false);
     encodeColumns(table_info, fields, RowEncodeVersion::RowV1);
+
     auto new_table_info = getTableInfoFieldsForInvalidNULLTest({EXTRA_HANDLE_COLUMN_ID}, false);
+    invalid_null_column_ids.emplace(11);
+    ASSERT_TRUE(new_table_info.getColumnInfo(11).hasNotNullFlag()); // col 11 is not null
+
     auto new_decoding_schema = getDecodingStorageSchemaSnapshot(new_table_info);
     ASSERT_FALSE(decodeAndCheckColumns(new_decoding_schema, false));
-    ASSERT_ANY_THROW(decodeAndCheckColumns(new_decoding_schema, true));
+    ASSERT_TRUE(decodeAndCheckColumns(new_decoding_schema, true));
 }
 
 

dbms/src/TiDB/Decode/RowCodec.cpp

@@ -227,7 +227,7 @@ struct RowEncoderV2
         is_big = is_big || value_length > std::numeric_limits<RowV2::Types<false>::ValueOffsetType>::max();
 
         /// Encode header.
-        encodeUInt(UInt8(RowCodecVer::ROW_V2), ss);
+        encodeUInt(static_cast<UInt8>(RowCodecVer::ROW_V2), ss);
         UInt8 row_flag = 0;
         row_flag |= is_big ? RowV2::BigRowMask : 0;
         encodeUInt(row_flag, ss);
@@ -533,19 +533,26 @@ bool appendRowV2ToBlockImpl(
                 {
                     if (!force_decode)
                     {
+                        // Detect `NULL` column value in a non-nullable column in the schema, let upper level try to sync the schema.
                         return false;
                     }
                     else
                     {
-                        throw Exception(
-                            "Detected invalid null when decoding data of column " + column_info.name
-                                + " with column type " + raw_column->getName(),
-                            ErrorCodes::LOGICAL_ERROR);
+                        // If user turn a nullable column (with default value) to a non-nullable column. There could exists some rows
+                        // with `NULL` values inside the SST files. Or some key-values encoded with old schema come after the schema
+                        // change is applied in TiFlash because of network issue.
+                        // If the column_id exists in null_column_ids, it means the column has default value but filled with NULL.
+                        // Just filled with default value for these old schema rows.
+                        raw_column->insert(column_info.defaultValueToField());
+                        idx_null++;
                     }
                 }
-                // ColumnNullable::insertDefault just insert a null value
-                raw_column->insertDefault();
-                idx_null++;
+                else
+                {
+                    // ColumnNullable::insertDefault just insert a null value
+                    raw_column->insertDefault();
+                    idx_null++;
+                }
             }
             else
             {
@@ -668,20 +675,25 @@ bool appendRowV1ToBlock(
             }
             if (datum.invalidNull(column_info))
             {
-                // Null value with non-null type detected, fatal if force_decode is true,
-                // as schema being newer and with invalid null shouldn't happen.
-                // Otherwise return false to outer, outer should sync schema and try again.
-                if (force_decode)
+                if (!force_decode)
                 {
-                    throw Exception(
-                        "Detected invalid null when decoding data " + std::to_string(unflattened.get<UInt64>())
-                            + " of column " + column_info.name + " with type " + raw_column->getName(),
-                        ErrorCodes::LOGICAL_ERROR);
+                    // Detect `NULL` column value in a non-nullable column in the schema, let upper level try to sync the schema.
+                    return false;
                 }
-
-                return false;
+                else
+                {
+                    // If user turn a nullable column (with default value) to a non-nullable column. There could exists some rows
+                    // with `NULL` values inside the SST files. Or some key-values encoded with old schema come after the schema
+                    // change is applied in TiFlash because of network issue.
+                    // If the column_id exists in null_column_ids, it means the column has default value but filled with NULL.
+                    // Just filled with default value for these old schema rows.
+                    raw_column->insert(column_info.defaultValueToField());
+                }
+            }
+            else
+            {
+                raw_column->insert(unflattened);
             }
-            raw_column->insert(unflattened);
             decoded_field_iter++;
             column_ids_iter++;
             block_column_pos++;

tests/fullstack-test2/ddl/alter_column_nullable.test

@@ -91,3 +91,56 @@ mysql> select * from test.a1
 +----+-----+------+
 
 mysql> drop table if exists test.a1
+
+##########
+# From nullable to not null, then add tiflash replica
+mysql> drop table if exists test.alt2;
+mysql> create table test.alt2 (a int, b DECIMAL(20) NULL default 1.0, c int NOT NULL);
+mysql> insert into test.alt2(a,b,c) values (1,NULL,1),(2,NULL,2),(3,NULL,3),(4,NULL,4);
+mysql> insert into test.alt2(a,b,c) values (11,0.1,11),(12,0.2,12),(13,0.3,13),(14,0.4,14);
+mysql> update test.alt2 set b = 0 where b is null;
+mysql> select * from test.alt2 order by a;
++------+------+----+
+| a    | b    | c  |
++------+------+----+
+|    1 |    0 |  1 |
+|    2 |    0 |  2 |
+|    3 |    0 |  3 |
+|    4 |    0 |  4 |
+|   11 |    0 | 11 |
+|   12 |    0 | 12 |
+|   13 |    0 | 13 |
+|   14 |    0 | 14 |
++------+------+----+
+mysql> alter table test.alt2 modify column b DECIMAL(20) NOT NULL;
+mysql> select * from test.alt2 order by a;
++------+---+----+
+| a    | b | c  |
++------+---+----+
+|    1 | 0 |  1 |
+|    2 | 0 |  2 |
+|    3 | 0 |  3 |
+|    4 | 0 |  4 |
+|   11 | 0 | 11 |
+|   12 | 0 | 12 |
+|   13 | 0 | 13 |
+|   14 | 0 | 14 |
++------+---+----+
+# Send snapshot to TiFlash
+mysql> alter table test.alt2 set tiflash replica 1;
+func> wait_table test alt2
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.alt2 order by a;
++------+---+----+
+| a    | b | c  |
++------+---+----+
+|    1 | 0 |  1 |
+|    2 | 0 |  2 |
+|    3 | 0 |  3 |
+|    4 | 0 |  4 |
+|   11 | 0 | 11 |
+|   12 | 0 | 12 |
+|   13 | 0 | 13 |
+|   14 | 0 | 14 |
++------+---+----+
+
+mysql> drop table if exists test.alt2