Backport #46402 to 22.3: Apply ALTER TABLE ON CLUSTER MOVE PARTITION TO DISK|VOLUME to all replicas.

src/Interpreters/DDLWorker.cpp

@@ -655,10 +655,9 @@ bool DDLWorker::taskShouldBeExecutedOnLeader(const ASTPtr & ast_ddl, const Stora
     if (auto * alter = ast_ddl->as<ASTAlterQuery>())
     {
         // Setting alters should be executed on all replicas
-        if (alter->isSettingsAlter())
-            return false;
-
-        if (alter->isFreezeAlter())
+        if (alter->isSettingsAlter() ||
+            alter->isFreezeAlter() ||
+            alter->isMovePartitionToDiskOrVolumeAlter())
             return false;
     }
 

src/Parsers/ASTAlterQuery.cpp

@@ -533,6 +533,24 @@ bool ASTAlterQuery::isDropPartitionAlter() const
     return isOneCommandTypeOnly(ASTAlterCommand::DROP_PARTITION) || isOneCommandTypeOnly(ASTAlterCommand::DROP_DETACHED_PARTITION);
 }
 
+bool ASTAlterQuery::isMovePartitionToDiskOrVolumeAlter() const
+{
+    if (command_list)
+    {
+        if (command_list->children.empty())
+            return false;
+        for (const auto & child : command_list->children)
+        {
+            const auto & command = child->as<const ASTAlterCommand &>();
+            if (command.type != ASTAlterCommand::MOVE_PARTITION ||
+                (command.move_destination_type != DataDestinationType::DISK && command.move_destination_type != DataDestinationType::VOLUME))
+                return false;
+        }
+        return true;
+    }
+    return false;
+}
+
 
 /** Get the text that identifies this element. */
 String ASTAlterQuery::getID(char delim) const

src/Parsers/ASTAlterQuery.h

@@ -239,6 +239,8 @@ class ASTAlterQuery : public ASTQueryWithTableAndOutput, public ASTQueryWithOnCl
 
     bool isDropPartitionAlter() const;
 
+    bool isMovePartitionToDiskOrVolumeAlter() const;
+
     String getID(char) const override;
 
     ASTPtr clone() const override;

tests/integration/test_move_partition_to_disk_on_cluster/configs/config.d/cluster.xml

@@ -0,0 +1,17 @@
+<clickhouse>
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <internal_replication>true</internal_replication>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_cluster>
+    </remote_servers>
+</clickhouse>

tests/integration/test_move_partition_to_disk_on_cluster/configs/config.d/storage_configuration.xml

@@ -0,0 +1,28 @@
+<clickhouse>
+
+<storage_configuration>
+    <disks>
+        <jbod1>
+            <path>/jbod1/</path>
+        </jbod1>
+        <external>
+            <path>/external/</path>
+        </external>
+    </disks>
+
+    <policies>
+        <jbod_with_external>
+            <volumes>
+                <main>
+                    <disk>jbod1</disk>
+                </main>
+                <external>
+                    <disk>external</disk>
+                </external>
+            </volumes>
+        </jbod_with_external>
+    </policies>
+
+</storage_configuration>
+
+</clickhouse>

tests/integration/test_move_partition_to_disk_on_cluster/test.py

@@ -0,0 +1,92 @@
+import pytest
+from helpers.client import QueryRuntimeException
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+
+node1 = cluster.add_instance(
+    "node1",
+    main_configs=[
+        "configs/config.d/storage_configuration.xml",
+        "configs/config.d/cluster.xml",
+    ],
+    with_zookeeper=True,
+    stay_alive=True,
+    tmpfs=["/jbod1:size=10M", "/external:size=10M"],
+    macros={"shard": 0, "replica": 1},
+)
+
+node2 = cluster.add_instance(
+    "node2",
+    main_configs=[
+        "configs/config.d/storage_configuration.xml",
+        "configs/config.d/cluster.xml",
+    ],
+    with_zookeeper=True,
+    stay_alive=True,
+    tmpfs=["/jbod1:size=10M", "/external:size=10M"],
+    macros={"shard": 0, "replica": 2},
+)
+
+
+@pytest.fixture(scope="module")
+def start_cluster():
+    try:
+        cluster.start()
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def test_move_partition_to_disk_on_cluster(start_cluster):
+    for node in [node1, node2]:
+        node.query(
+            sql="CREATE TABLE test_local_table"
+            "(x UInt64) "
+            "ENGINE=ReplicatedMergeTree('/clickhouse/tables/test_local_table', '{replica}') "
+            "ORDER BY tuple()"
+            "SETTINGS storage_policy = 'jbod_with_external';",
+        )
+
+    node1.query("INSERT INTO test_local_table VALUES (0)")
+    node1.query("SYSTEM SYNC REPLICA test_local_table", timeout=30)
+
+    try:
+        node1.query(
+            sql="ALTER TABLE test_local_table ON CLUSTER 'test_cluster' MOVE PARTITION tuple() TO DISK 'jbod1';",
+        )
+    except QueryRuntimeException:
+        pass
+
+    for node in [node1, node2]:
+        assert (
+            node.query(
+                "SELECT partition_id, disk_name FROM system.parts WHERE table = 'test_local_table' FORMAT Values"
+            )
+            == "('all','jbod1')"
+        )
+
+    node1.query(
+        sql="ALTER TABLE test_local_table ON CLUSTER 'test_cluster' MOVE PARTITION tuple() TO DISK 'external';",
+    )
+
+    for node in [node1, node2]:
+        assert (
+            node.query(
+                "SELECT partition_id, disk_name FROM system.parts WHERE table = 'test_local_table' FORMAT Values"
+            )
+            == "('all','external')"
+        )
+
+    node1.query(
+        sql="ALTER TABLE test_local_table ON CLUSTER 'test_cluster' MOVE PARTITION tuple() TO VOLUME 'main';",
+    )
+
+    for node in [node1, node2]:
+        assert (
+            node.query(
+                "SELECT partition_id, disk_name FROM system.parts WHERE table = 'test_local_table' FORMAT Values"
+            )
+            == "('all','jbod1')"
+        )