[flash-363] support rename column

dbms/src/Debug/DBGInvoker.cpp

@@ -43,6 +43,7 @@ DBGInvoker::DBGInvoker()
     regFunc("add_column_to_tidb_table", MockTiDBTable::dbgFuncAddColumnToTiDBTable);
     regFunc("drop_column_from_tidb_table", MockTiDBTable::dbgFuncDropColumnFromTiDBTable);
     regFunc("modify_column_in_tidb_table", MockTiDBTable::dbgFuncModifyColumnInTiDBTable);
+    regFunc("rename_column_in_tidb_table", MockTiDBTable::dbgFuncRenameColumnInTiDBTable);
     regFunc("rename_tidb_table", MockTiDBTable::dbgFuncRenameTiDBTable);
     regFunc("truncate_tidb_table", MockTiDBTable::dbgFuncTruncateTiDBTable);
 

dbms/src/Debug/MockTiDB.cpp

@@ -400,6 +400,33 @@ void MockTiDB::modifyColumnInTable(const String & database_name, const String &
     version_diff[version] = diff;
 }
 
+void MockTiDB::renameColumnInTable(
+    const String & database_name, const String & table_name, const String & old_column_name, const String & new_column_name)
+{
+    std::lock_guard lock(tables_mutex);
+
+    TablePtr table = getTableByNameInternal(database_name, table_name);
+    String qualified_name = database_name + "." + table_name;
+    auto & columns = table->table_info.columns;
+    auto it = std::find_if(columns.begin(), columns.end(), [&](const ColumnInfo & column_) { return column_.name == old_column_name; });
+    if (it == columns.end())
+        throw Exception("Column " + old_column_name + " does not exist in TiDB table  " + qualified_name, ErrorCodes::LOGICAL_ERROR);
+
+    if (columns.end()
+        != std::find_if(columns.begin(), columns.end(), [&](const ColumnInfo & column_) { return column_.name == new_column_name; }))
+        throw Exception("Column " + new_column_name + " exists in TiDB table  " + qualified_name, ErrorCodes::LOGICAL_ERROR);
+
+    it->name = new_column_name;
+
+    version++;
+    SchemaDiff diff;
+    diff.type = SchemaActionModifyColumn;
+    diff.schema_id = table->table_info.db_id;
+    diff.table_id = table->id();
+    diff.version = version;
+    version_diff[version] = diff;
+}
+
 void MockTiDB::renameTable(const String & database_name, const String & table_name, const String & new_table_name)
 {
     std::lock_guard lock(tables_mutex);

dbms/src/Debug/MockTiDB.h

@@ -74,6 +74,9 @@ class MockTiDB : public ext::singleton<MockTiDB>
 
     void modifyColumnInTable(const String & database_name, const String & table_name, const NameAndTypePair & column);
 
+    void renameColumnInTable(
+        const String & database_name, const String & table_name, const String & old_column_name, const String & new_column_name);
+
     void renameTable(const String & database_name, const String & table_name, const String & new_table_name);
 
     void truncateTable(const String & database_name, const String & table_name);

dbms/src/Debug/dbgFuncMockTiDBTable.cpp

@@ -235,6 +235,23 @@ void MockTiDBTable::dbgFuncModifyColumnInTiDBTable(DB::Context & context, const
     output(ss.str());
 }
 
+void MockTiDBTable::dbgFuncRenameColumnInTiDBTable(DB::Context &, const DB::ASTs & args, DB::DBGInvoker::Printer output)
+{
+    if (args.size() != 4)
+        throw Exception("Args not matched, should be: database-name, table-name, old_col_name, new_col_name", ErrorCodes::BAD_ARGUMENTS);
+
+    const String & database_name = typeid_cast<const ASTIdentifier &>(*args[0]).name;
+    const String & table_name = typeid_cast<const ASTIdentifier &>(*args[1]).name;
+    const String & old_column_name = typeid_cast<const ASTIdentifier &>(*args[2]).name;
+    const String & new_column_name = typeid_cast<const ASTIdentifier &>(*args[3]).name;
+
+    MockTiDB::instance().renameColumnInTable(database_name, table_name, old_column_name, new_column_name);
+
+    std::stringstream ss;
+    ss << "rename column " << old_column_name << " " << new_column_name;
+    output(ss.str());
+}
+
 void MockTiDBTable::dbgFuncRenameTiDBTable(Context & /*context*/, const ASTs & args, DBGInvoker::Printer output)
 {
     if (args.size() != 3)

dbms/src/Debug/dbgFuncMockTiDBTable.h

@@ -57,6 +57,11 @@ struct MockTiDBTable
     //   ./storages-client.sh "DBGInvoke modify_column_in_tidb_table(database_name, table_name, 'col type')"
     static void dbgFuncModifyColumnInTiDBTable(Context & context, const ASTs & args, DBGInvoker::Printer output);
 
+    // Rename a column in a mocked TiDB table.
+    // Usage:
+    //   ./storages-client.sh "DBGInvoke rename_column_in_tidb_table(database_name, table_name, old_col_name, new_col_name)"
+    static void dbgFuncRenameColumnInTiDBTable(Context & context, const ASTs & args, DBGInvoker::Printer output);
+
     // Rename a TiDB table.
     // Usage:
     //   ./storages-client.sh "DBGInvoke rename_tidb_table(database_name, table_name, new_table)"

dbms/src/Storages/AlterCommands.cpp

@@ -160,6 +160,25 @@ void AlterCommand::apply(ColumnsDescription & columns_description) const
         /// This have no relation to changing the list of columns.
         /// TODO Check that all columns exist, that only columns with constant defaults are added.
     }
+    else if (type == RENAME_COLUMN)
+    {
+        auto & columns = columns_description.ordinary;
+        // We only consider ordinary columns.
+        NamesAndTypesList::iterator old_column_it = std::find_if(columns.begin(), columns.end(),
+            [&](const NameAndTypePair & orig_column){ return orig_column.name == column_name; });
+        if(old_column_it == columns.end())
+        {
+            throw Exception("Rename column fails, old column name: " + column_name + " doesn't exist ", ErrorCodes::LOGICAL_ERROR);
+        }
+        // Make sure new column doesn't exist.
+        NamesAndTypesList::iterator new_column_it = std::find_if(columns.begin(), columns.end(),
+            [&](const NameAndTypePair & orig_column){ return orig_column.name == new_column_name; });
+        if(new_column_it != columns.end())
+        {
+            throw Exception("Rename column fails, new column name: " + column_name + " has existed ", ErrorCodes::LOGICAL_ERROR);
+        }
+        old_column_it->name = new_column_name;
+    }
     else
         throw Exception("Wrong parameter type in ALTER query", ErrorCodes::LOGICAL_ERROR);
 }

dbms/src/Storages/AlterCommands.h

@@ -15,12 +15,17 @@ struct AlterCommand
         DROP_COLUMN,
         MODIFY_COLUMN,
         MODIFY_PRIMARY_KEY,
+        // Rename column is only for tmt schema sync.
+        RENAME_COLUMN,
     };
 
     Type type;
 
     String column_name;
 
+    // Only for Rename column.
+    String new_column_name;
+
     /// For DROP COLUMN ... FROM PARTITION
     String partition_name;
 

dbms/src/Storages/MergeTree/MergeTreeData.cpp

@@ -1119,6 +1119,85 @@ void MergeTreeData::createConvertExpression(const DataPartPtr & part, DataPart::
     }
 }
 
+MergeTreeData::AlterDataPartTransactionPtr MergeTreeData::renameColumnPart(
+    const DataPartPtr & part,
+    const NamesAndTypesList & new_columns,
+    const AlterCommand & command) const
+{
+    // check if is compact
+    if (part->isCompactFormat())
+    {
+        throw Exception("L0 compact does not support alter clause.");
+    }
+    AlterDataPartTransactionPtr transaction(new AlterDataPartTransaction(part)); /// Blocks changes to the part.
+    DataPart::Checksums new_checksums = part->checksums;
+
+    DataTypePtr data_type;
+
+    for (const auto & pair : new_columns)
+    {
+        if (pair.name == command.new_column_name)
+        {
+            data_type = pair.type;
+            break;
+        }
+    }
+    if (data_type == nullptr)
+    {
+        throw Exception("Can't find new column " + command.new_column_name + " in column list", ErrorCodes::LOGICAL_ERROR);
+    }
+
+    data_type->enumerateStreams(
+        [&](const IDataType::SubstreamPath & substream_path)
+        {
+            /// Skip array sizes, because they cannot be modified in ALTER.
+            if (!substream_path.empty() && substream_path.back().type == IDataType::Substream::ArraySizes)
+                return;
+
+            String old_file_name = IDataType::getFileNameForStream(command.column_name, substream_path);
+            String new_file_name = IDataType::getFileNameForStream(command.new_column_name, substream_path);
+
+            transaction->rename_map[old_file_name + ".bin"] = new_file_name + ".bin";
+            transaction->rename_map[old_file_name + ".mrk"] = new_file_name + ".mrk";
+
+            auto it = new_checksums.files.find(old_file_name + ".bin");
+            if (it == new_checksums.files.end())
+            {
+                // Can not find old checksums.
+                throw Exception("Cannot find file checksums for " + old_file_name + ".bin", ErrorCodes::LOGICAL_ERROR);
+            }
+            new_checksums.files.erase(it);
+            new_checksums.files.emplace(new_file_name + ".bin", it->second);
+            it = new_checksums.files.find(old_file_name + ".mrk");
+            if (it == new_checksums.files.end())
+            {
+                // Can not find old checksums.
+                throw Exception("Cannot find file checksums for " + old_file_name + ".mrk", ErrorCodes::LOGICAL_ERROR);
+            }
+            new_checksums.files.erase(it);
+            new_checksums.files.emplace(new_file_name + ".mrk", it->second);
+        }, {});
+
+    /// Write the checksums to the temporary file.
+    if (!part->checksums.empty())
+    {
+        transaction->new_checksums = new_checksums;
+        WriteBufferFromFile checksums_file(part->getFullPath() + "checksums.txt.tmp", 4096);
+        new_checksums.write(checksums_file);
+        transaction->rename_map["checksums.txt.tmp"] = "checksums.txt";
+    }
+
+    /// Write the new column list to the temporary file.
+    {
+        transaction->new_columns = new_columns;
+        WriteBufferFromFile columns_file(part->getFullPath() + "columns.txt.tmp", 4096);
+        transaction->new_columns.writeText(columns_file);
+        transaction->rename_map["columns.txt.tmp"] = "columns.txt";
+    }
+
+    return transaction;
+}
+
 MergeTreeData::AlterDataPartTransactionPtr MergeTreeData::alterDataPart(
     const DataPartPtr & part,
     const NamesAndTypesList & new_columns,

dbms/src/Storages/MergeTree/MergeTreeData.h

@@ -444,6 +444,12 @@ class MergeTreeData : public ITableDeclaration
         const ASTPtr & new_primary_key,
         bool skip_sanity_checks);
 
+    /// Performs column rename operation for tmt storage.
+    AlterDataPartTransactionPtr renameColumnPart(
+        const DataPartPtr & part,
+        const NamesAndTypesList & new_columns,
+        const AlterCommand & command) const;
+
     /// Should be called if part data is suspected to be corrupted.
     void reportBrokenPart(const String & name)
     {

dbms/src/Storages/StorageMergeTree.cpp

@@ -354,13 +354,22 @@ void StorageMergeTree::alterInternal(
 
     ASTPtr new_primary_key_ast = data.primary_expr_ast;
 
+    bool rename_column = false;
     for (const AlterCommand & param : params)
     {
         if (param.type == AlterCommand::MODIFY_PRIMARY_KEY)
         {
             primary_key_is_modified = true;
             new_primary_key_ast = param.primary_key;
         }
+        else if (param.type == AlterCommand::RENAME_COLUMN)
+        {
+            rename_column = true;
+            if (params.size() != 1) 
+            {
+                throw Exception("There is an internal error for rename columns", ErrorCodes::LOGICAL_ERROR);
+            }
+        }
     }
 
     if (primary_key_is_modified && supportsSampling())
@@ -370,7 +379,12 @@ void StorageMergeTree::alterInternal(
     auto columns_for_parts = new_columns.getAllPhysical();
     for (const MergeTreeData::DataPartPtr & part : parts)
     {
-        if (auto transaction = data.alterDataPart(part, columns_for_parts, new_primary_key_ast, false))
+        if (rename_column)
+        {
+            auto transaction = data.renameColumnPart(part, columns_for_parts, params[0]);
+            transactions.push_back(std::move(transaction));
+        }
+        else if (auto transaction = data.alterDataPart(part, columns_for_parts, new_primary_key_ast, false))
             transactions.push_back(std::move(transaction));
     }