Add check of RW access to paths on start. Add multiple disks tests.

dbms/programs/server/Server.cpp

@@ -83,6 +83,7 @@ namespace ErrorCodes
     extern const int FAILED_TO_GETPWUID;
     extern const int MISMATCHING_USERS_FOR_PROCESS_AND_DATA;
     extern const int NETWORK_ERROR;
+    extern const int PATH_ACCESS_DENIED;
 }
 
 
@@ -261,6 +262,14 @@ int Server::main(const std::vector<std::string> & /*args*/)
     Poco::File(path + "data/" + default_database).createDirectories();
     Poco::File(path + "metadata/" + default_database).createDirectories();
 
+    /// Check that we have write access to all data paths
+    auto disk_selector = global_context->getDiskSelector();
+    for (auto [name, disk] : disk_selector.getDisksMap()) {
+        Poco::File disk_path(disk->getPath());
+        if (!disk_path.canRead() or !disk_path.canWrite())
+            throw Exception("There is no RW access to disk " + name + " (" + disk->getPath() + ")", ErrorCodes::PATH_ACCESS_DENIED);
+    }
+
     StatusFile status{path + "status"};
 
     SCOPE_EXIT({

dbms/src/Common/ErrorCodes.cpp

@@ -441,6 +441,7 @@ namespace ErrorCodes
     extern const int UNKNOWN_POLICY = 464;
     extern const int UNKNOWN_DISK = 465;
     extern const int UNKNOWN_PROTOCOL = 466;
+    extern const int PATH_ACCESS_DENIED = 467;
 
     extern const int KEEPER_EXCEPTION = 999;
     extern const int POCO_EXCEPTION = 1000;

dbms/src/Interpreters/Context.cpp

@@ -1736,7 +1736,7 @@ const DiskSpace::DiskPtr & Context::getDisk(const String & name) const
 }
 
 
-    DiskSpace::DiskSelector & Context::getDiskSelector() const
+DiskSpace::DiskSelector & Context::getDiskSelector() const
 {
     auto lock = getLock();
 

dbms/src/Storages/MergeTree/DiskSpaceMonitor.h

@@ -41,7 +41,13 @@ using ReservationPtr = std::unique_ptr<Reservation>;
 class Space : public std::enable_shared_from_this<Space>
 {
 public:
+    Space() = default;
+
     virtual ~Space() = default;
+    Space(const Space &) = default;
+    Space(Space &&) = default;
+    Space& operator=(const Space &) = default;
+    Space& operator=(Space &&) = default;
 
     virtual ReservationPtr reserve(UInt64 bytes) const = 0;
 

dbms/src/Storages/MergeTree/MergeTreeData.cpp

@@ -2370,7 +2370,9 @@ MergeTreeData::DataPartPtr MergeTreeData::swapActivePart(MergeTreeData::DataPart
             if (it == data_parts_by_info.end())
                 throw Exception("No such active part by info. It is a bug", ErrorCodes::NO_SUCH_DATA_PART);
 
+            (*it)->remove_time.store((*it)->modification_time, std::memory_order_relaxed);
             data_parts_indexes.erase(it);
+
             auto part_it = data_parts_indexes.insert(part).first;
             modifyPartState(part_it, DataPartState::Committed);
             return active_part;
@@ -2547,6 +2549,12 @@ void MergeTreeData::movePartitionToSpace(MergeTreeData::DataPartPtr part, DiskSp
     if (!reservation || !reservation->isValid())
         throw Exception("Move is not possible. Not enough space " + space->getName(), ErrorCodes::NOT_ENOUGH_SPACE);
 
+    auto & reserved_disk = reservation->getDisk();
+    String path_to_clone = getFullPathOnDisk(reserved_disk);
+
+    if (Poco::File(path_to_clone + part->name).exists())
+        throw Exception("Move is not possible: " + path_to_clone + part->name + " already exists", ErrorCodes::DIRECTORY_ALREADY_EXISTS);
+
     LOG_DEBUG(log, "Cloning part " << part->getFullPath() << " to " << getFullPathOnDisk(reservation->getDisk()));
     part->makeCloneOnDiskDetached(reservation);
 
@@ -2557,10 +2565,14 @@ void MergeTreeData::movePartitionToSpace(MergeTreeData::DataPartPtr part, DiskSp
 
     copied_part->loadColumnsChecksumsIndexes(require_part_metadata, true);
 
+    if (Poco::File(path_to_clone + part->name).exists())
+        throw Exception("Move is not possible: " + path_to_clone + part->name + " already exists", ErrorCodes::DIRECTORY_ALREADY_EXISTS);
     copied_part->renameTo(part->name);
 
     auto old_active_part = swapActivePart(copied_part);
 
+    std::cerr << old_active_part.use_count() << std::endl;
+
     old_active_part->deleteOnDestroy();
 }
 
@@ -2594,7 +2606,7 @@ void MergeTreeData::movePartitionToVolume(const ASTPtr & partition, const String
 
     auto volume = storage_policy->getVolumeByName(name);
     if (!volume)
-        throw Exception("Disk " + name + " does not exists on policy " + storage_policy->getName(), ErrorCodes::UNKNOWN_DISK);
+        throw Exception("Volume " + name + " does not exists on policy " + storage_policy->getName(), ErrorCodes::UNKNOWN_DISK);
 
     for (const auto & disk : volume->disks)
         if (part->disk->getName() == disk->getName())

dbms/tests/integration/test_multiple_disks/test.py

@@ -94,20 +94,33 @@ def test_default(test_cluster):
 
 
 def test_move(test_cluster):
-    assert node1.query("create table node1_move_mt ( d UInt64 )\n ENGINE = MergeTree\n ORDER BY d\n SETTINGS storage_policy_name='default_disk_with_external'") == ""
-    assert node1.query("insert into node1_move_mt values (1)") == ""
-    assert node1.query("select disk_name from system.parts where table == 'node1_move_mt'") == "default\n"
+    assert node2.query("create table node1_move_mt ( d UInt64 )\n ENGINE = MergeTree\n ORDER BY d\n SETTINGS storage_policy_name='default_disk_with_external'") == ""
+    assert node2.query("insert into node1_move_mt values (1)") == ""
+    assert node2.query("select disk_name from system.parts where table == 'node1_move_mt'") == "default\n"
+
+    test_cluster.open_bash_shell('node2')
     # move from default to external
-    assert node1.query("alter table node1_move_mt move PART 'all_1_1_0' to disk 'external'") == ""
-    assert node1.query("select disk_name from system.parts where table == 'node1_move_mt'") == "external\n"
-    # move back
-    assert node1.query("alter table node1_move_mt move PART 'all_1_1_0' to disk 'default'") == ""
-    assert node1.query("select disk_name from system.parts where table == 'node1_move_mt'") == "default\n"
+    assert node2.query("alter table node1_move_mt move PART 'all_1_1_0' to disk 'external'") == ""
+    assert node2.query("select disk_name from system.parts where table == 'node1_move_mt'") == "external\n"
+    time.sleep(5)
+    # Check that it really moved
+    assert node2.query("detach table node1_move_mt") == ""
+    assert node2.query("attach table node1_move_mt") == ""
+    assert node2.query("select disk_name from system.parts where table == 'node1_move_mt'") == "external\n"
+
+    # move back by volume small, that contains only 'default' disk
+    assert node2.query("alter table node1_move_mt move PART 'all_1_1_0' to volume 'small'") == ""
+    assert node2.query("select disk_name from system.parts where table == 'node1_move_mt'") == "default\n"
+    time.sleep(5)
+    # Check that it really moved
+    assert node2.query("detach table node1_move_mt") == ""
+    assert node2.query("attach table node1_move_mt") == ""
+    assert node2.query("select disk_name from system.parts where table == 'node1_move_mt'") == "default\n"
 
 
 def test_no_policy(test_cluster):
     try:
-        node2.query("create table node1_move_mt ( d UInt64 )\n ENGINE = MergeTree\n ORDER BY d\n SETTINGS storage_policy_name='name_that_does_not_exists'")
+        node1.query("create table node1_move_mt ( d UInt64 )\n ENGINE = MergeTree\n ORDER BY d\n SETTINGS storage_policy_name='name_that_does_not_exists'")
     except Exception as e:
         assert str(e).strip().split("\n")[1].find("Unknown StoragePolicy name_that_does_not_exists") != -1