Merge pull request #22706 from songenjie/fetch_part

[clickhouse][server][ddl]support fetch part

docs/en/sql-reference/statements/alter/partition.md

@@ -16,7 +16,7 @@ The following operations with [partitions](../../../engines/table-engines/merget
 -   [CLEAR COLUMN IN PARTITION](#alter_clear-column-partition) — Resets the value of a specified column in a partition.
 -   [CLEAR INDEX IN PARTITION](#alter_clear-index-partition) — Resets the specified secondary index in a partition.
 -   [FREEZE PARTITION](#alter_freeze-partition) — Creates a backup of a partition.
--   [FETCH PARTITION](#alter_fetch-partition) — Downloads a partition from another server.
+-   [FETCH PARTITION\|PART](#alter_fetch-partition) — Downloads a part or partition from another server.
 -   [MOVE PARTITION\|PART](#alter_move-partition) — Move partition/data part to another disk or volume.
 
 <!-- -->
@@ -198,29 +198,35 @@ ALTER TABLE table_name CLEAR INDEX index_name IN PARTITION partition_expr
 
 The query works similar to `CLEAR COLUMN`, but it resets an index instead of a column data.
 
-## FETCH PARTITION {#alter_fetch-partition}
+## FETCH PARTITION|PART {#alter_fetch-partition}
 
 ``` sql
-ALTER TABLE table_name FETCH PARTITION partition_expr FROM 'path-in-zookeeper'
+ALTER TABLE table_name FETCH PARTITION|PART partition_expr FROM 'path-in-zookeeper'
 ```
 
 Downloads a partition from another server. This query only works for the replicated tables.
 
 The query does the following:
 
-1.  Downloads the partition from the specified shard. In ‘path-in-zookeeper’ you must specify a path to the shard in ZooKeeper.
+1.  Downloads the partition|part from the specified shard. In ‘path-in-zookeeper’ you must specify a path to the shard in ZooKeeper.
 2.  Then the query puts the downloaded data to the `detached` directory of the `table_name` table. Use the [ATTACH PARTITION\|PART](#alter_attach-partition) query to add the data to the table.
 
 For example:
 
+1. FETCH PARTITION
 ``` sql
 ALTER TABLE users FETCH PARTITION 201902 FROM '/clickhouse/tables/01-01/visits';
 ALTER TABLE users ATTACH PARTITION 201902;
 ```
+2. FETCH PART
+``` sql
+ALTER TABLE users FETCH PART 201901_2_2_0 FROM '/clickhouse/tables/01-01/visits';
+ALTER TABLE users ATTACH PART 201901_2_2_0;
+```
 
 Note that:
 
--   The `ALTER ... FETCH PARTITION` query isn’t replicated. It places the partition to the `detached` directory only on the local server.
+-   The `ALTER ... FETCH PARTITION|PART` query isn’t replicated. It places the part or partition to the `detached` directory only on the local server.
 -   The `ALTER TABLE ... ATTACH` query is replicated. It adds the data to all replicas. The data is added to one of the replicas from the `detached` directory, and to the others - from neighboring replicas.
 
 Before downloading, the system checks if the partition exists and the table structure matches. The most appropriate replica is selected automatically from the healthy replicas.

docs/en/sql-reference/statements/grant.md

@@ -279,7 +279,7 @@ Allows executing [ALTER](../../sql-reference/statements/alter/index.md) queries
         -   `ALTER MATERIALIZE TTL`. Level: `TABLE`. Aliases: `MATERIALIZE TTL`
         -   `ALTER SETTINGS`. Level: `TABLE`. Aliases: `ALTER SETTING`, `ALTER MODIFY SETTING`, `MODIFY SETTING`
         -   `ALTER MOVE PARTITION`. Level: `TABLE`. Aliases: `ALTER MOVE PART`, `MOVE PARTITION`, `MOVE PART`
-        -   `ALTER FETCH PARTITION`. Level: `TABLE`. Aliases: `FETCH PARTITION`
+        -   `ALTER FETCH PARTITION`. Level: `TABLE`. Aliases: `ALTER FETCH PART`, `FETCH PARTITION`, `FETCH PART`
         -   `ALTER FREEZE PARTITION`. Level: `TABLE`. Aliases: `FREEZE PARTITION`
     -   `ALTER VIEW` Level: `GROUP`
         -   `ALTER VIEW REFRESH`. Level: `VIEW`. Aliases: `ALTER LIVE VIEW REFRESH`, `REFRESH VIEW`

src/Access/AccessType.h

@@ -62,7 +62,7 @@ enum class AccessType
                                                                        enabled implicitly by the grant ALTER_TABLE */\
     M(ALTER_SETTINGS, "ALTER SETTING, ALTER MODIFY SETTING, MODIFY SETTING", TABLE, ALTER_TABLE) /* allows to execute ALTER MODIFY SETTING */\
     M(ALTER_MOVE_PARTITION, "ALTER MOVE PART, MOVE PARTITION, MOVE PART", TABLE, ALTER_TABLE) \
-    M(ALTER_FETCH_PARTITION, "FETCH PARTITION", TABLE, ALTER_TABLE) \
+    M(ALTER_FETCH_PARTITION, "ALTER FETCH PART, FETCH PARTITION", TABLE, ALTER_TABLE) \
     M(ALTER_FREEZE_PARTITION, "FREEZE PARTITION, UNFREEZE", TABLE, ALTER_TABLE) \
     \
     M(ALTER_TABLE, "", GROUP, ALTER) \

src/Parsers/ASTAlterQuery.cpp

@@ -245,7 +245,7 @@ void ASTAlterCommand::formatImpl(
     else if (type == ASTAlterCommand::FETCH_PARTITION)
     {
         settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "FETCH "
-                      << "PARTITION " << (settings.hilite ? hilite_none : "");
+                      << (part ? "PART " : "PARTITION ") << (settings.hilite ? hilite_none : "");
         partition->formatImpl(settings, state, frame);
         settings.ostr << (settings.hilite ? hilite_keyword : "")
                       << " FROM " << (settings.hilite ? hilite_none : "") << DB::quote << from;

src/Parsers/ParserAlterQuery.cpp

@@ -61,6 +61,7 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
     ParserKeyword s_drop_detached_partition("DROP DETACHED PARTITION");
     ParserKeyword s_drop_detached_part("DROP DETACHED PART");
     ParserKeyword s_fetch_partition("FETCH PARTITION");
+    ParserKeyword s_fetch_part("FETCH PART");
     ParserKeyword s_replace_partition("REPLACE PARTITION");
     ParserKeyword s_freeze("FREEZE");
     ParserKeyword s_unfreeze("UNFREEZE");
@@ -428,6 +429,21 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
             command->from = ast_from->as<ASTLiteral &>().value.get<const String &>();
             command->type = ASTAlterCommand::FETCH_PARTITION;
         }
+        else if (s_fetch_part.ignore(pos, expected))
+        {
+            if (!parser_string_literal.parse(pos, command->partition, expected))
+                return false;
+
+            if (!s_from.ignore(pos, expected))
+                return false;
+
+            ASTPtr ast_from;
+            if (!parser_string_literal.parse(pos, ast_from, expected))
+                return false;
+            command->from = ast_from->as<ASTLiteral &>().value.get<const String &>();
+            command->part = true;
+            command->type = ASTAlterCommand::FETCH_PARTITION;
+        }
         else if (s_freeze.ignore(pos, expected))
         {
             if (s_partition.ignore(pos, expected))

src/Storages/MergeTree/MergeTreeData.cpp

@@ -2909,7 +2909,12 @@ void MergeTreeData::movePartitionToVolume(const ASTPtr & partition, const String
         throw Exception("Cannot move parts because moves are manually disabled", ErrorCodes::ABORTED);
 }
 
-void MergeTreeData::fetchPartition(const ASTPtr & /*partition*/, const StorageMetadataPtr & /*metadata_snapshot*/, const String & /*from*/, ContextPtr /*query_context*/)
+void MergeTreeData::fetchPartition(
+    const ASTPtr & /*partition*/,
+    const StorageMetadataPtr & /*metadata_snapshot*/,
+    const String & /*from*/,
+    bool /*fetch_part*/,
+    ContextPtr /*query_context*/)
 {
     throw Exception(ErrorCodes::NOT_IMPLEMENTED, "FETCH PARTITION is not supported by storage {}", getName());
 }
@@ -2972,7 +2977,7 @@ Pipe MergeTreeData::alterPartition(
             break;
 
             case PartitionCommand::FETCH_PARTITION:
-                fetchPartition(command.partition, metadata_snapshot, command.from_zookeeper_path, query_context);
+                fetchPartition(command.partition, metadata_snapshot, command.from_zookeeper_path, command.part, query_context);
                 break;
 
             case PartitionCommand::FREEZE_PARTITION:

src/Storages/MergeTree/MergeTreeData.h

@@ -970,7 +970,12 @@ protected:
     virtual void movePartitionToTable(const StoragePtr & dest_table, const ASTPtr & partition, ContextPtr context) = 0;
 
     /// Makes sense only for replicated tables
-    virtual void fetchPartition(const ASTPtr & partition, const StorageMetadataPtr & metadata_snapshot, const String & from, ContextPtr query_context);
+    virtual void fetchPartition(
+        const ASTPtr & partition,
+        const StorageMetadataPtr & metadata_snapshot,
+        const String & from,
+        bool fetch_part,
+        ContextPtr query_context);
 
     void writePartLog(
         PartLogElement::Type type,

src/Storages/PartitionCommands.cpp

@@ -82,6 +82,7 @@ std::optional<PartitionCommand> PartitionCommand::parse(const ASTAlterCommand *
         res.type = FETCH_PARTITION;
         res.partition = command_ast->partition;
         res.from_zookeeper_path = command_ast->from;
+        res.part = command_ast->part;
         return res;
     }
     else if (command_ast->type == ASTAlterCommand::FREEZE_PARTITION)
@@ -140,7 +141,10 @@ std::string PartitionCommand::typeToString() const
         else
             return "DROP DETACHED PARTITION";
     case PartitionCommand::Type::FETCH_PARTITION:
-        return "FETCH PARTITION";
+        if (part)
+            return "FETCH PART";
+        else
+            return "FETCH PARTITION";
     case PartitionCommand::Type::FREEZE_ALL_PARTITIONS:
         return "FREEZE ALL";
     case PartitionCommand::Type::FREEZE_PARTITION:

src/Storages/StorageReplicatedMergeTree.cpp

@@ -130,6 +130,7 @@ namespace ErrorCodes
     extern const int UNKNOWN_POLICY;
     extern const int NO_SUCH_DATA_PART;
     extern const int INTERSERVER_SCHEME_DOESNT_MATCH;
+    extern const int DUPLICATE_DATA_PART;
 }
 
 namespace ActionLocks
@@ -5356,11 +5357,11 @@ void StorageReplicatedMergeTree::getReplicaDelays(time_t & out_absolute_delay, t
     }
 }
 
-
 void StorageReplicatedMergeTree::fetchPartition(
     const ASTPtr & partition,
     const StorageMetadataPtr & metadata_snapshot,
     const String & from_,
+    bool fetch_part,
     ContextPtr query_context)
 {
     Macros::MacroExpansionInfo info;
@@ -5373,40 +5374,54 @@ void StorageReplicatedMergeTree::fetchPartition(
     if (from.empty())
         throw Exception("ZooKeeper path should not be empty", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 
-    String partition_id = getPartitionIDFromQuery(partition, query_context);
     zkutil::ZooKeeperPtr zookeeper;
     if (auxiliary_zookeeper_name != default_zookeeper_name)
-    {
         zookeeper = getContext()->getAuxiliaryZooKeeper(auxiliary_zookeeper_name);
-
-        LOG_INFO(log, "Will fetch partition {} from shard {} (auxiliary zookeeper '{}')", partition_id, from_, auxiliary_zookeeper_name);
-    }
     else
-    {
         zookeeper = getZooKeeper();
 
-        LOG_INFO(log, "Will fetch partition {} from shard {}", partition_id, from_);
-    }
-
     if (from.back() == '/')
         from.resize(from.size() - 1);
 
+    if (fetch_part)
+    {
+        String part_name = partition->as<ASTLiteral &>().value.safeGet<String>();
+        auto part_path = findReplicaHavingPart(part_name, from, zookeeper);
+
+        if (part_path.empty())
+            throw Exception(ErrorCodes::NO_REPLICA_HAS_PART, "Part {} does not exist on any replica", part_name);
+        /** Let's check that there is no such part in the `detached` directory (where we will write the downloaded parts).
+          * Unreliable (there is a race condition) - such a part may appear a little later.
+          */
+        if (checkIfDetachedPartExists(part_name))
+            throw Exception(ErrorCodes::DUPLICATE_DATA_PART, "Detached part " + part_name + " already exists.");
+        LOG_INFO(log, "Will fetch part {} from shard {} (zookeeper '{}')", part_name, from_, auxiliary_zookeeper_name);
+
+        try
+        {
+            /// part name , metadata, part_path , true, 0, zookeeper
+            if (!fetchPart(part_name, metadata_snapshot, part_path, true, 0, zookeeper))
+                throw Exception(ErrorCodes::UNFINISHED, "Failed to fetch part {} from {}", part_name, from_);
+        }
+        catch (const DB::Exception & e)
+        {
+            if (e.code() != ErrorCodes::RECEIVED_ERROR_FROM_REMOTE_IO_SERVER && e.code() != ErrorCodes::RECEIVED_ERROR_TOO_MANY_REQUESTS
+                && e.code() != ErrorCodes::CANNOT_READ_ALL_DATA)
+                throw;
+
+            LOG_INFO(log, e.displayText());
+        }
+        return;
+    }
+
+    String partition_id = getPartitionIDFromQuery(partition, query_context);
+    LOG_INFO(log, "Will fetch partition {} from shard {} (zookeeper '{}')", partition_id, from_, auxiliary_zookeeper_name);
 
     /** Let's check that there is no such partition in the `detached` directory (where we will write the downloaded parts).
       * Unreliable (there is a race condition) - such a partition may appear a little later.
       */
-    Poco::DirectoryIterator dir_end;
-    for (const std::string & path : getDataPaths())
-    {
-        for (Poco::DirectoryIterator dir_it{path + "detached/"}; dir_it != dir_end; ++dir_it)
-        {
-            MergeTreePartInfo part_info;
-            if (MergeTreePartInfo::tryParsePartName(dir_it.name(), &part_info, format_version)
-                && part_info.partition_id == partition_id)
-                throw Exception("Detached partition " + partition_id + " already exists.", ErrorCodes::PARTITION_ALREADY_EXISTS);
-        }
-
-    }
+    if (checkIfDetachedPartitionExists(partition_id))
+        throw Exception("Detached partition " + partition_id + " already exists.", ErrorCodes::PARTITION_ALREADY_EXISTS);
 
     zkutil::Strings replicas;
     zkutil::Strings active_replicas;
@@ -6913,4 +6928,46 @@ String StorageReplicatedMergeTree::getSharedDataReplica(
     return best_replica;
 }
 
+String StorageReplicatedMergeTree::findReplicaHavingPart(
+    const String & part_name, const String & zookeeper_path_, zkutil::ZooKeeper::Ptr zookeeper_)
+{
+    Strings replicas = zookeeper_->getChildren(zookeeper_path_ + "/replicas");
+
+    /// Select replicas in uniformly random order.
+    std::shuffle(replicas.begin(), replicas.end(), thread_local_rng);
+
+    for (const String & replica : replicas)
+    {
+        if (zookeeper_->exists(zookeeper_path_ + "/replicas/" + replica + "/parts/" + part_name)
+            && zookeeper_->exists(zookeeper_path_ + "/replicas/" + replica + "/is_active"))
+            return zookeeper_path_ + "/replicas/" + replica;
+    }
+
+    return {};
+}
+
+bool StorageReplicatedMergeTree::checkIfDetachedPartExists(const String & part_name)
+{
+    Poco::DirectoryIterator dir_end;
+    for (const std::string & path : getDataPaths())
+        for (Poco::DirectoryIterator dir_it{path + "detached/"}; dir_it != dir_end; ++dir_it)
+            if (dir_it.name() == part_name)
+                return true;
+    return false;
+}
+
+bool StorageReplicatedMergeTree::checkIfDetachedPartitionExists(const String & partition_name)
+{
+    Poco::DirectoryIterator dir_end;
+    for (const std::string & path : getDataPaths())
+    {
+        for (Poco::DirectoryIterator dir_it{path + "detached/"}; dir_it != dir_end; ++dir_it)
+        {
+            MergeTreePartInfo part_info;
+            if (MergeTreePartInfo::tryParsePartName(dir_it.name(), &part_info, format_version) && part_info.partition_id == partition_name)
+                return true;
+        }
+    }
+    return false;
+}
 }

src/Storages/StorageReplicatedMergeTree.h

@@ -522,8 +522,11 @@ private:
     /** Returns an empty string if no one has a part.
       */
     String findReplicaHavingPart(const String & part_name, bool active);
+    static String findReplicaHavingPart(const String & part_name, const String & zookeeper_path_, zkutil::ZooKeeper::Ptr zookeeper_);
 
     bool checkReplicaHavePart(const String & replica, const String & part_name);
+    bool checkIfDetachedPartExists(const String & part_name);
+    bool checkIfDetachedPartitionExists(const String & partition_name);
 
     /** Find replica having specified part or any part that covers it.
       * If active = true, consider only active replicas.
@@ -626,7 +629,12 @@ private:
     PartitionCommandsResultInfo attachPartition(const ASTPtr & partition, const StorageMetadataPtr & metadata_snapshot, bool part, ContextPtr query_context) override;
     void replacePartitionFrom(const StoragePtr & source_table, const ASTPtr & partition, bool replace, ContextPtr query_context) override;
     void movePartitionToTable(const StoragePtr & dest_table, const ASTPtr & partition, ContextPtr query_context) override;
-    void fetchPartition(const ASTPtr & partition, const StorageMetadataPtr & metadata_snapshot, const String & from, ContextPtr query_context) override;
+    void fetchPartition(
+        const ASTPtr & partition,
+        const StorageMetadataPtr & metadata_snapshot,
+        const String & from,
+        bool fetch_part,
+        ContextPtr query_context) override;
 
     /// Check granularity of already existing replicated table in zookeeper if it exists
     /// return true if it's fixed

tests/fuzz/ast.dict

@@ -156,6 +156,7 @@
 "extractURLParameterNames"
 "extractURLParameters"
 "FETCH PARTITION"
+"FETCH PART"
 "FINAL"
 "FIRST"
 "firstSignificantSubdomain"

tests/integration/test_fetch_partition_from_auxiliary_zookeeper/test.py

@@ -1,5 +1,3 @@
-
-
 import pytest
 from helpers.client import QueryRuntimeException
 from helpers.cluster import ClickHouseCluster
@@ -18,23 +16,33 @@ def start_cluster():
         cluster.shutdown()
 
 
-def test_fetch_part_from_allowed_zookeeper(start_cluster):
+@pytest.mark.parametrize(
+    ('part', 'date', 'part_name'),
+    [
+        ('PARTITION', '2020-08-27', '2020-08-27'),
+        ('PART', '2020-08-28', '20200828_0_0_0'),
+    ]
+)
+def test_fetch_part_from_allowed_zookeeper(start_cluster, part, date, part_name):
     node.query(
-        "CREATE TABLE simple (date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/0/simple', 'node') ORDER BY tuple() PARTITION BY date;"
+        "CREATE TABLE IF NOT EXISTS simple (date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/0/simple', 'node') ORDER BY tuple() PARTITION BY date;"
     )
-    node.query("INSERT INTO simple VALUES ('2020-08-27', 1)")
+
+    node.query("""INSERT INTO simple VALUES ('{date}', 1)""".format(date=date))
 
     node.query(
-        "CREATE TABLE simple2 (date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/1/simple', 'node') ORDER BY tuple() PARTITION BY date;"
+        "CREATE TABLE IF NOT EXISTS simple2 (date Date, id UInt32) ENGINE = ReplicatedMergeTree('/clickhouse/tables/1/simple', 'node') ORDER BY tuple() PARTITION BY date;"
     )
+
     node.query(
-        "ALTER TABLE simple2 FETCH PARTITION '2020-08-27' FROM 'zookeeper2:/clickhouse/tables/0/simple';"
-    )
-    node.query("ALTER TABLE simple2 ATTACH PARTITION '2020-08-27';")
+        """ALTER TABLE simple2 FETCH {part} '{part_name}' FROM 'zookeeper2:/clickhouse/tables/0/simple';""".format(
+            part=part, part_name=part_name))
+
+    node.query("""ALTER TABLE simple2 ATTACH {part} '{part_name}';""".format(part=part, part_name=part_name))
 
     with pytest.raises(QueryRuntimeException):
         node.query(
-            "ALTER TABLE simple2 FETCH PARTITION '2020-08-27' FROM 'zookeeper:/clickhouse/tables/0/simple';"
-        )
+            """ALTER TABLE simple2 FETCH {part} '{part_name}' FROM 'zookeeper:/clickhouse/tables/0/simple';""".format(
+                part=part, part_name=part_name))
 
-    assert node.query("SELECT id FROM simple2").strip() == "1"
+    assert node.query("""SELECT id FROM simple2 where date = '{date}'""".format(date=date)).strip() == "1"

tests/queries/0_stateless/01271_show_privileges.reference

@@ -28,7 +28,7 @@ ALTER TTL	['ALTER MODIFY TTL','MODIFY TTL']	TABLE	ALTER TABLE
 ALTER MATERIALIZE TTL	['MATERIALIZE TTL']	TABLE	ALTER TABLE
 ALTER SETTINGS	['ALTER SETTING','ALTER MODIFY SETTING','MODIFY SETTING']	TABLE	ALTER TABLE
 ALTER MOVE PARTITION	['ALTER MOVE PART','MOVE PARTITION','MOVE PART']	TABLE	ALTER TABLE
-ALTER FETCH PARTITION	['FETCH PARTITION']	TABLE	ALTER TABLE
+ALTER FETCH PARTITION	['ALTER FETCH PART','FETCH PARTITION']	TABLE	ALTER TABLE
 ALTER FREEZE PARTITION	['FREEZE PARTITION','UNFREEZE']	TABLE	ALTER TABLE
 ALTER TABLE	[]	\N	ALTER
 ALTER VIEW REFRESH	['ALTER LIVE VIEW REFRESH','REFRESH VIEW']	VIEW	ALTER VIEW

tests/testflows/rbac/tests/privileges/grant_option.py

@@ -89,7 +89,7 @@ def grant_option_check(grant_option_target, grant_target, user_name, table_type,
 @Examples("privilege", [
     ("ALTER MOVE PARTITION",), ("ALTER MOVE PART",), ("MOVE PARTITION",), ("MOVE PART",),
     ("ALTER DELETE",), ("DELETE",),
-    ("ALTER FETCH PARTITION",), ("FETCH PARTITION",),
+    ("ALTER FETCH PARTITION",), ("ALTER FETCH PART",), ("FETCH PARTITION",),
     ("ALTER FREEZE PARTITION",), ("FREEZE PARTITION",),
     ("ALTER UPDATE",), ("UPDATE",),
     ("ALTER ADD COLUMN",), ("ADD COLUMN",),