Fix create table error (#4630)

dbms/src/Databases/test/gtest_database.cpp

@@ -70,7 +70,7 @@ class DatabaseTiFlash_test : public ::testing::Test
         }
     }
 
-    void recreateMetadataPath() const
+    static void recreateMetadataPath()
     {
         String path = TiFlashTestEnv::getContext().getPath();
 
@@ -628,6 +628,118 @@ try
 }
 CATCH
 
+TEST_F(DatabaseTiFlash_test, ISSUE4596)
+try
+{
+    const String db_name = "db_1";
+    auto ctx = TiFlashTestEnv::getContext();
+
+    {
+        // Create database
+        const String statement = "CREATE DATABASE " + db_name + " ENGINE=TiFlash";
+        ASTPtr ast = parseCreateStatement(statement);
+        InterpreterCreateQuery interpreter(ast, ctx);
+        interpreter.setInternal(true);
+        interpreter.setForceRestoreData(false);
+        interpreter.execute();
+    }
+
+    auto db = ctx.getDatabase(db_name);
+
+    const String tbl_name = "t_111";
+    {
+        /// Create table
+        ParserCreateQuery parser;
+        const String stmt = fmt::format("CREATE TABLE `{}`.`{}` ", db_name, tbl_name) +
+            R"stmt( 
+                (`id` Int32,`b` String) Engine = DeltaMerge((`id`),
+                    '{
+                        "cols":[{
+                            "comment":"",
+                            "default":null,
+                            "default_bit":null,
+                            "id":1,
+                            "name":{
+                                "L":"id",
+                                "O":"id"
+                            },
+                            "offset":0,
+                            "origin_default":null,
+                            "state":5,
+                            "type":{
+                                "Charset":"binary",
+                                "Collate":"binary",
+                                "Decimal":0,
+                                "Elems":null,
+                                "Flag":515,
+                                "Flen":16,
+                                "Tp":3
+                            }
+                        },
+                        {
+                            "comment":"",
+                            "default":"",
+                            "default_bit":null,
+                            "id":15,
+                            "name":{
+                                "L":"b",
+                                "O":"b"
+                            },
+                            "offset":12,
+                            "origin_default":"",
+                            "state":5,
+                            "type":{
+                                "Charset":"binary",
+                                "Collate":"binary",
+                                "Decimal":0,
+                                "Elems":null,
+                                "Flag":4225,
+                                "Flen":-1,
+                                "Tp":251
+                            }
+                        }],
+                        "comment":"",
+                        "id":330,
+                        "index_info":[],
+                        "is_common_handle":false,
+                        "name":{
+                            "L":"test",
+                            "O":"test"
+                        },
+                        "partition":null,
+                        "pk_is_handle":true,
+                        "schema_version":465,
+                        "state":5,
+                        "update_timestamp":99999
+                    }'
+                )
+            )stmt";
+        ASTPtr ast = parseQuery(parser, stmt, 0);
+
+        InterpreterCreateQuery interpreter(ast, ctx);
+        interpreter.setInternal(true);
+        interpreter.setForceRestoreData(false);
+        interpreter.execute();
+    }
+
+    EXPECT_FALSE(db->empty(ctx));
+    EXPECT_TRUE(db->isTableExist(ctx, tbl_name));
+
+    {
+        // Get storage from database
+        auto storage = db->tryGetTable(ctx, tbl_name);
+        ASSERT_NE(storage, nullptr);
+
+        EXPECT_EQ(storage->getName(), MutableSupport::delta_tree_storage_name);
+        EXPECT_EQ(storage->getTableName(), tbl_name);
+
+        auto managed_storage = std::dynamic_pointer_cast<IManageableStorage>(storage);
+        EXPECT_EQ(managed_storage->getDatabaseName(), db_name);
+        EXPECT_EQ(managed_storage->getTableInfo().name, "test");
+    }
+}
+CATCH
+
 TEST_F(DatabaseTiFlash_test, ISSUE_1055)
 try
 {
@@ -664,7 +776,7 @@ try
     DatabaseLoading::loadTable(ctx, *db, meta_path, db_name, db_data_path, "TiFlash", "t_45.sql", false);
 
     // Get storage from database
-    const auto tbl_name = "t_45";
+    const auto * tbl_name = "t_45";
     auto storage = db->tryGetTable(ctx, tbl_name);
     ASSERT_NE(storage, nullptr);
     EXPECT_EQ(storage->getName(), MutableSupport::delta_tree_storage_name);
@@ -752,7 +864,7 @@ try
         auto db = ctx.getDatabase(name_mapper.mapDatabaseName(*db_info));
         ASSERT_NE(db, nullptr);
         EXPECT_EQ(db->getEngineName(), "TiFlash");
-        auto flash_db = typeid_cast<DatabaseTiFlash *>(db.get());
+        auto * flash_db = typeid_cast<DatabaseTiFlash *>(db.get());
         auto & db_info_get = flash_db->getDatabaseInfo();
         ASSERT_EQ(db_info_get.name, expect_name);
     }
@@ -817,7 +929,7 @@ try
 )",
     };
 
-    for (auto & statement : statements)
+    for (const auto & statement : statements)
     {
         {
             // Cleanup: Drop database if exists

dbms/src/Storages/Transaction/TiDB.cpp

@@ -90,10 +90,13 @@ Field ColumnInfo::defaultValueToField() const
             auto v = value.convert<String>();
             if (hasBinaryFlag())
             {
-                // For binary column, we have to pad trailing zeros according to the specified type length.
+                // For some binary column(like varchar(20)), we have to pad trailing zeros according to the specified type length.
                 // User may define default value `0x1234` for a `BINARY(4)` column, TiDB stores it in a string "\u12\u34" (sized 2).
                 // But it actually means `0x12340000`.
-                v.append(flen - v.length(), '\0');
+                // And for some binary column(like longblob), we do not need to pad trailing zeros.
+                // And the `Flen` is set to -1, therefore we need to check `Flen >= 0` here.
+                if (Int32 vlen = v.length(); flen >= 0 && vlen < flen)
+                    v.append(flen - vlen, '\0');
             }
             return v;
         }

dbms/src/Storages/Transaction/TiDB.h

@@ -79,7 +79,7 @@ enum TP
 #ifdef M
 #error "Please undefine macro M first."
 #endif
-#define M(tt, v, cf, ct, w) Type##tt = v,
+#define M(tt, v, cf, ct, w) Type##tt = (v),
     COLUMN_TYPES(M)
 #undef M
 };
@@ -111,7 +111,7 @@ enum ColumnFlag
 #ifdef M
 #error "Please undefine macro M first."
 #endif
-#define M(cf, v) ColumnFlag##cf = v,
+#define M(cf, v) ColumnFlag##cf = (v),
     COLUMN_FLAGS(M)
 #undef M
 };
@@ -140,7 +140,7 @@ enum CodecFlag
 #ifdef M
 #error "Please undefine macro M first."
 #endif
-#define M(cf, v) CodecFlag##cf = v,
+#define M(cf, v) CodecFlag##cf = (v),
     CODEC_FLAGS(M)
 #undef M
 };
@@ -185,10 +185,10 @@ struct ColumnInfo
 #ifdef M
 #error "Please undefine macro M first."
 #endif
-#define M(f, v)                                                  \
-    inline bool has##f##Flag() const { return (flag & v) != 0; } \
-    inline void set##f##Flag() { flag |= v; }                    \
-    inline void clear##f##Flag() { flag &= (~v); }
+#define M(f, v)                                                    \
+    inline bool has##f##Flag() const { return (flag & (v)) != 0; } \
+    inline void set##f##Flag() { flag |= (v); }                    \
+    inline void clear##f##Flag() { flag &= (~(v)); }
     COLUMN_FLAGS(M)
 #undef M
 
@@ -213,7 +213,7 @@ struct PartitionDefinition
 {
     PartitionDefinition() = default;
 
-    PartitionDefinition(Poco::JSON::Object::Ptr json);
+    explicit PartitionDefinition(Poco::JSON::Object::Ptr json);
 
     Poco::JSON::Object::Ptr getJSONObject() const;
 
@@ -229,7 +229,7 @@ struct PartitionInfo
 {
     PartitionInfo() = default;
 
-    PartitionInfo(Poco::JSON::Object::Ptr json);
+    explicit PartitionInfo(Poco::JSON::Object::Ptr json);
 
     Poco::JSON::Object::Ptr getJSONObject() const;
 
@@ -252,7 +252,7 @@ struct DBInfo
     SchemaState state;
 
     DBInfo() = default;
-    DBInfo(const String & json) { deserialize(json); }
+    explicit DBInfo(const String & json) { deserialize(json); }
 
     String serialize() const;
 
@@ -357,9 +357,9 @@ struct TableInfo
     ::TiDB::StorageEngine engine_type = ::TiDB::StorageEngine::UNSPECIFIED;
 
     ColumnID getColumnID(const String & name) const;
-    String getColumnName(const ColumnID id) const;
+    String getColumnName(ColumnID id) const;
 
-    const ColumnInfo & getColumnInfo(const ColumnID id) const;
+    const ColumnInfo & getColumnInfo(ColumnID id) const;
 
     std::optional<std::reference_wrapper<const ColumnInfo>> getPKHandleColumn() const;