Change ZooKeeper path for zero-copy locks for shared data

src/Disks/DiskDecorator.h

@@ -70,6 +70,18 @@ class DiskDecorator : public IDisk
     void startup() override;
     void applyNewSettings(const Poco::Util::AbstractConfiguration & config, ContextPtr context, const String & config_prefix, const DisksMap & map) override;
 
+    std::unique_ptr<ReadBufferFromFileBase> readMetaFile(
+        const String & path,
+        const ReadSettings & settings,
+        std::optional<size_t> size) const override { return delegate->readMetaFile(path, settings, size); }
+
+    std::unique_ptr<WriteBufferFromFileBase> writeMetaFile(
+        const String & path,
+        size_t buf_size,
+        WriteMode mode) override { return delegate->writeMetaFile(path, buf_size, mode); }
+
+    UInt32 getRefCount(const String & path) const override { return delegate->getRefCount(path); }
+
 protected:
     Executor & getExecutor() override;
 

src/Disks/IDisk.cpp

@@ -86,4 +86,22 @@ SyncGuardPtr IDisk::getDirectorySyncGuard(const String & /* path */) const
     return nullptr;
 }
 
+std::unique_ptr<ReadBufferFromFileBase> IDisk::readMetaFile(
+    const String & path,
+    const ReadSettings & settings,
+    std::optional<size_t> size) const
+{
+    LOG_TRACE(&Poco::Logger::get("IDisk"), "Read local metafile: {}", path);
+    return readFile(path, settings, size);
+}
+
+std::unique_ptr<WriteBufferFromFileBase> IDisk::writeMetaFile(
+    const String & path,
+    size_t buf_size,
+    WriteMode mode)
+{
+    LOG_TRACE(&Poco::Logger::get("IDisk"), "Write local metafile: {}", path);
+    return writeFile(path, buf_size, mode);
+}
+
 }

src/Disks/IDisk.h

@@ -247,6 +247,26 @@ class IDisk : public Space
     /// Applies new settings for disk in runtime.
     virtual void applyNewSettings(const Poco::Util::AbstractConfiguration &, ContextPtr, const String &, const DisksMap &) {}
 
+    /// Open the local file for read and return ReadBufferFromFileBase object.
+    /// Overridden in IDiskRemote.
+    /// Used for work with custom metadata.
+    virtual std::unique_ptr<ReadBufferFromFileBase> readMetaFile(
+        const String & path,
+        const ReadSettings & settings = ReadSettings{},
+        std::optional<size_t> size = {}) const;
+
+    /// Open the local file for write and return WriteBufferFromFileBase object.
+    /// Overridden in IDiskRemote.
+    /// Used for work with custom metadata.
+    virtual std::unique_ptr<WriteBufferFromFileBase> writeMetaFile(
+        const String & path,
+        size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE,
+        WriteMode mode = WriteMode::Rewrite);
+
+    /// Return reference count for remote FS.
+    /// Overridden in IDiskRemote.
+    virtual UInt32 getRefCount(const String &) const { return 0; }
+
 protected:
     friend class DiskDecorator;
 

src/Disks/IDiskRemote.cpp

@@ -484,6 +484,7 @@ bool IDiskRemote::tryReserve(UInt64 bytes)
 
 String IDiskRemote::getUniqueId(const String & path) const
 {
+    LOG_TRACE(log, "Remote path: {}, Path: {}", remote_fs_root_path, path);
     Metadata metadata(remote_fs_root_path, metadata_disk, path);
     String id;
     if (!metadata.remote_fs_objects.empty())
@@ -500,4 +501,28 @@ AsynchronousReaderPtr IDiskRemote::getThreadPoolReader()
     return reader;
 }
 
+std::unique_ptr<ReadBufferFromFileBase> IDiskRemote::readMetaFile(
+    const String & path,
+    const ReadSettings & settings,
+    std::optional<size_t> size) const
+{
+    LOG_TRACE(log, "Read metafile: {}", path);
+    return metadata_disk->readFile(path, settings, size);
+}
+
+std::unique_ptr<WriteBufferFromFileBase> IDiskRemote::writeMetaFile(
+    const String & path,
+    size_t buf_size,
+    WriteMode mode)
+{
+    LOG_TRACE(log, "Write metafile: {}", path);
+    return metadata_disk->writeFile(path, buf_size, mode);
+}
+
+UInt32 IDiskRemote::getRefCount(const String & path) const
+{
+    auto meta = readMeta(path);
+    return meta.ref_count;
+}
+
 }

src/Disks/IDiskRemote.h

@@ -136,6 +136,18 @@ friend class DiskRemoteReservation;
 
     static AsynchronousReaderPtr getThreadPoolReader();
 
+    virtual std::unique_ptr<ReadBufferFromFileBase> readMetaFile(
+        const String & path,
+        const ReadSettings & settings = ReadSettings{},
+        std::optional<size_t> size = {}) const override;
+
+    virtual std::unique_ptr<WriteBufferFromFileBase> writeMetaFile(
+        const String & path,
+        size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE,
+        WriteMode mode = WriteMode::Rewrite) override;
+
+    UInt32 getRefCount(const String & path) const override;
+
 protected:
     Poco::Logger * log;
     const String name;

src/Storages/IStorage.h

@@ -137,6 +137,9 @@ class IStorage : public std::enable_shared_from_this<IStorage>, public TypePromo
     /// Returns true if the storage replicates SELECT, INSERT and ALTER commands among replicas.
     virtual bool supportsReplication() const { return false; }
 
+    /// Returns replica name for replicated storage
+    virtual String getReplicaName() const { return ""; }
+
     /// Returns true if the storage supports parallel insert.
     virtual bool supportsParallelInsert() const { return false; }
 

src/Storages/MergeTree/IMergeTreeDataPart.cpp

@@ -1100,6 +1100,14 @@ void IMergeTreeDataPart::renameTo(const String & new_relative_path, bool remove_
     storage.lockSharedData(*this);
 }
 
+void IMergeTreeDataPart::cleanupOldName(const String & old_name) const
+{
+    if (name == old_name)
+        return;
+
+    storage.unlockSharedData(*this, old_name);
+}
+
 std::optional<bool> IMergeTreeDataPart::keepSharedDataInDecoupledStorage() const
 {
     /// NOTE: It's needed for zero-copy replication
@@ -1561,6 +1569,12 @@ String IMergeTreeDataPart::getUniqueId() const
 }
 
 
+UInt32 IMergeTreeDataPart::getRefCount() const
+{
+    return volume->getDisk()->getRefCount(fs::path(getFullRelativePath()) / "checksums.txt");
+}
+
+
 String IMergeTreeDataPart::getZeroLevelPartBlockID() const
 {
     if (info.level != 0)

src/Storages/MergeTree/IMergeTreeDataPart.h

@@ -334,6 +334,9 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
     /// Changes only relative_dir_name, you need to update other metadata (name, is_temp) explicitly
     virtual void renameTo(const String & new_relative_path, bool remove_new_dir_if_exists) const;
 
+    /// Cleanup after change part
+    virtual void cleanupOldName(const String & old_part_name) const;
+
     /// Makes clone of a part in detached/ directory via hard links
     virtual void makeCloneInDetached(const String & prefix, const StorageMetadataPtr & metadata_snapshot) const;
 
@@ -398,10 +401,14 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
     /// Returns serialization for column according to files in which column is written in part.
     SerializationPtr getSerializationForColumn(const NameAndTypePair & column) const;
 
-    /// Return some uniq string for file
-    /// Required for distinguish different copies of the same part on S3
+    /// Return some uniq string for file.
+    /// Required for distinguish different copies of the same part on remote FS.
     String getUniqueId() const;
 
+    /// Return hardlink count for part.
+    /// Required for keep data on remote FS when part has shadow copies.
+    UInt32 getRefCount() const;
+
 protected:
 
     /// Total size of all columns, calculated once in calcuateColumnSizesOnDisk