Fix an invalid type of a column after attach and alter.

src/Storages/MergeTree/MutateTask.cpp

@@ -220,8 +220,11 @@ getColumnsForNewDataPart(
     if (!isWidePart(source_part))
         return {updated_header.getNamesAndTypesList(), new_serialization_infos};
 
-    Names source_column_names = source_part->getColumns().getNames();
-    NameSet source_columns_name_set(source_column_names.begin(), source_column_names.end());
+    const auto & source_columns = source_part->getColumns();
+    std::unordered_map<String, DataTypePtr> source_columns_name_to_type;
+    for (const auto & it : source_columns)
+        source_columns_name_to_type[it.name] = it.type;
+
     for (auto it = storage_columns.begin(); it != storage_columns.end();)
     {
         if (updated_header.has(it->name))
@@ -233,14 +236,25 @@ getColumnsForNewDataPart(
         }
         else
         {
-            if (!source_columns_name_set.contains(it->name))
+            auto source_col = source_columns_name_to_type.find(it->name);
+            if (source_col == source_columns_name_to_type.end())
             {
                 /// Source part doesn't have column but some other column
                 /// was renamed to it's name.
                 auto renamed_it = renamed_columns_to_from.find(it->name);
-                if (renamed_it != renamed_columns_to_from.end()
-                    && source_columns_name_set.contains(renamed_it->second))
-                    ++it;
+                if (renamed_it != renamed_columns_to_from.end())
+                {
+                    source_col = source_columns_name_to_type.find(renamed_it->second);
+                    if (source_col == source_columns_name_to_type.end())
+                        it = storage_columns.erase(it);
+                    else
+                    {
+                        /// Take a type from source part column.
+                        /// It may differ from column type in storage.
+                        it->type = source_col->second;
+                        ++it;
+                    }
+                }
                 else
                     it = storage_columns.erase(it);
             }
@@ -262,7 +276,12 @@ getColumnsForNewDataPart(
                 if (!renamed_columns_to_from.contains(it->name) && (was_renamed || was_removed))
                     it = storage_columns.erase(it);
                 else
+                {
+                    /// Take a type from source part column.
+                    /// It may differ from column type in storage.
+                    it->type = source_col->second;
                     ++it;
+                }
             }
         }
     }

src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp

@@ -315,7 +315,12 @@ void ReplicatedMergeTreeSink::commitPart(
     DataPartStorageBuilderPtr builder,
     size_t replicas_num)
 {
-    metadata_snapshot->check(part->getColumns());
+    /// It is possible that we alter a part with different types of source columns.
+    /// In this case, if column was not altered, the result type will be different with what we have in metadata.
+    /// For now, consider it is ok. See 02461_alter_update_respect_part_column_type_bug for an example.
+    ///
+    /// metadata_snapshot->check(part->getColumns());
+
     assertSessionIsNotExpired(zookeeper);
 
     String temporary_part_relative_path = part->data_part_storage->getPartDirectory();

tests/queries/0_stateless/02012_zookeeper_changed_enum_type_incompatible.reference

@@ -0,0 +1,4 @@
+one	1
+two	1
+one	1
+two	1

tests/queries/0_stateless/02012_zookeeper_changed_enum_type_incompatible.sql

@@ -11,5 +11,6 @@ alter table enum_alter_issue detach partition id 'all';
 alter table enum_alter_issue modify column a Enum8('one' = 1, 'two' = 2, 'three' = 3);
 insert into enum_alter_issue values ('one', 1), ('two', 1);
 
-alter table enum_alter_issue attach partition id 'all'; -- {serverError TYPE_MISMATCH}
+alter table enum_alter_issue attach partition id 'all';
+select * from enum_alter_issue;
 drop table enum_alter_issue;

tests/queries/0_stateless/02461_alter_update_respect_part_column_type_bug.reference

@@ -0,0 +1,9 @@
+1	one	test1
+one	one	test1
+one	one	test
+one	one	test
+-----
+1	one	test1
+one	one	test1
+one	one	test
+one	one	test

tests/queries/0_stateless/02461_alter_update_respect_part_column_type_bug.sql

@@ -0,0 +1,94 @@
+drop table if exists src;
+create table src( A Int64, B String, C String) Engine=MergeTree order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values(1, 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column B Nullable(String);
+alter table src attach partition tuple();
+
+alter table src update C = 'test1' where 1 settings mutations_sync=2;
+select * from src;
+
+
+drop table if exists src;
+create table src( A String, B String, C String) Engine=MergeTree order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values('one', 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column A LowCardinality(String);
+alter table src attach partition tuple();
+
+alter table src update C = 'test1' where 1 settings mutations_sync=2;
+select * from src;
+
+
+drop table if exists src;
+create table src( A String, B String, C String) Engine=MergeTree order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values('one', 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column A LowCardinality(String);
+alter table src attach partition tuple();
+
+alter table src modify column C LowCardinality(String);
+select * from src;
+
+drop table if exists src;
+create table src( A String, B String, C String) Engine=MergeTree order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values('one', 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column B Nullable(String);
+alter table src attach partition tuple();
+
+alter table src rename column B to D;
+select * from src;
+
+select '-----';
+
+drop table if exists src;
+create table src( A Int64, B String, C String) Engine=ReplicatedMergeTree('/clickhouse/{database}/test/src1', '1') order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values(1, 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column B Nullable(String);
+alter table src attach partition tuple();
+
+alter table src update C = 'test1' where 1 settings mutations_sync=2;
+select * from src;
+
+
+drop table if exists src;
+create table src( A String, B String, C String) Engine=ReplicatedMergeTree('/clickhouse/{database}/test/src2', '1') order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values('one', 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column A LowCardinality(String);
+alter table src attach partition tuple();
+
+alter table src update C = 'test1' where 1 settings mutations_sync=2;
+select * from src;
+
+
+drop table if exists src;
+create table src( A String, B String, C String) Engine=ReplicatedMergeTree('/clickhouse/{database}/test/src3', '1') order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values('one', 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column A LowCardinality(String);
+alter table src attach partition tuple();
+
+alter table src modify column C LowCardinality(String);
+select * from src;
+
+drop table if exists src;
+create table src( A String, B String, C String) Engine=ReplicatedMergeTree('/clickhouse/{database}/test/src4', '1') order by A SETTINGS min_bytes_for_wide_part=0;
+insert into src values('one', 'one', 'test');
+
+alter table src detach partition tuple();
+alter table src modify column B Nullable(String);
+alter table src attach partition tuple();
+
+alter table src rename column B to D;
+select * from src;
+