Fix bugs with abs path and create mode

src/Disks/IDisk.cpp

@@ -100,6 +100,6 @@ SyncGuardPtr IDisk::getDirectorySyncGuard(const String & /* path */) const
     return nullptr;
 }
 
-MetadataStoragePtr IDisk::getMetadataStorage() { return std::make_shared<MetadataStorageFromDisk>(std::static_pointer_cast<IDisk>(shared_from_this())); }
+MetadataStoragePtr IDisk::getMetadataStorage() { return std::make_shared<MetadataStorageFromDisk>(std::static_pointer_cast<IDisk>(shared_from_this()), ""); }
 
 }

src/Disks/ObjectStorages/AzureBlobStorage/registerDiskAzureBlobStorage.cpp

@@ -83,7 +83,7 @@ void registerDiskAzureBlobStorage(DiskFactory & factory)
         uint64_t copy_thread_pool_size = config.getUInt(config_prefix + ".thread_pool_size", 16);
         bool send_metadata = config.getBool(config_prefix + ".send_metadata", false);
 
-        auto metadata_storage = std::make_shared<MetadataStorageFromDisk>(metadata_disk);
+        auto metadata_storage = std::make_shared<MetadataStorageFromDisk>(metadata_disk, "");
 
         std::shared_ptr<IDisk> azure_blob_storage_disk = std::make_shared<DiskObjectStorage>(
             name,

src/Disks/ObjectStorages/DiskObjectStorage.cpp

@@ -175,7 +175,7 @@ bool DiskObjectStorage::isFile(const String & path) const
 void DiskObjectStorage::createFile(const String & path)
 {
     auto tx = metadata_storage->createTransaction();
-    metadata_storage->createMetadataFile(path, tx);
+    metadata_storage->createEmptyMetadataFile(path, tx);
     tx->commit();
 }
 
@@ -537,7 +537,7 @@ std::unique_ptr<ReadBufferFromFileBase> DiskObjectStorage::readFile(
 std::unique_ptr<WriteBufferFromFileBase> DiskObjectStorage::writeFile(
     const String & path,
     size_t buf_size,
-    WriteMode,
+    WriteMode mode,
     const WriteSettings & settings)
 {
     auto blob_name = getRandomASCIIString();
@@ -553,10 +553,13 @@ std::unique_ptr<WriteBufferFromFileBase> DiskObjectStorage::writeFile(
         blob_name = "r" + revisionToString(revision) + "-file-" + blob_name;
     }
 
-    auto create_metadata_callback = [this, path, blob_name] (size_t count)
+    auto create_metadata_callback = [this, mode, path, blob_name] (size_t count)
     {
         auto tx = metadata_storage->createTransaction();
-        metadata_storage->addBlobToMetadata(path, blob_name, count, tx);
+        if (mode == WriteMode::Rewrite)
+            metadata_storage->createMetadataFile(path, blob_name, count, tx);
+        else
+            metadata_storage->addBlobToMetadata(path, blob_name, count, tx);
         tx->commit();
     };
 

src/Disks/ObjectStorages/HDFS/registerDiskHDFS.cpp

@@ -38,7 +38,7 @@ void registerDiskHDFS(DiskFactory & factory)
         ObjectStoragePtr hdfs_storage = std::make_unique<HDFSObjectStorage>(nullptr, uri, std::move(settings), config);
 
         auto metadata_disk = prepareForLocalMetadata(name, config, config_prefix, context_).second;
-        auto metadata_storage = std::make_shared<MetadataStorageFromDisk>(metadata_disk);
+        auto metadata_storage = std::make_shared<MetadataStorageFromDisk>(metadata_disk, uri);
         uint64_t copy_thread_pool_size = config.getUInt(config_prefix + ".thread_pool_size", 16);
 
         return std::make_shared<DiskObjectStorage>(

src/Disks/ObjectStorages/IMetadataStorage.h

@@ -53,7 +53,7 @@ public:
 
     virtual DirectoryIteratorPtr iterateDirectory(const String & path) = 0;
 
-    virtual void createMetadataFile(const std::string & path, MetadataTransactionPtr transaction) = 0;
+    virtual void createEmptyMetadataFile(const std::string & path, MetadataTransactionPtr transaction) = 0;
 
     virtual std::string readMetadataFileToString(const std::string & path) const = 0;
 
@@ -94,6 +94,8 @@ public:
 
     virtual BlobsPathToSize getBlobs(const std::string & path) const = 0;
 
+    virtual void createMetadataFile(const std::string & path, const std::string & blob_name, uint64_t size_in_bytes, MetadataTransactionPtr transaction) = 0;
+
     virtual void addBlobToMetadata(const std::string & path, const std::string & blob_name, uint64_t size_in_bytes, MetadataTransactionPtr transaction) = 0;
 
     virtual uint32_t unlinkAndGetHardlinkCount(const std::string & path, MetadataTransactionPtr transaction) = 0;

src/Disks/ObjectStorages/MetadataStorageFromDisk.cpp

@@ -538,7 +538,7 @@ std::string MetadataStorageFromDisk::readMetadataFileToString(const std::string
     return result;
 }
 
-void MetadataStorageFromDisk::createMetadataFile(const std::string & path, MetadataTransactionPtr transaction)
+void MetadataStorageFromDisk::createEmptyMetadataFile(const std::string & path, MetadataTransactionPtr transaction)
 {
     auto metadata = std::make_unique<DiskObjectStorageMetadata>(disk->getPath(), root_path_for_remote_metadata, path);
     writeMetadataToFile(path, transaction, metadata->serializeToString());
@@ -609,6 +609,14 @@ void MetadataStorageFromDisk::setReadOnly(const std::string & path, MetadataTran
 }
 
 
+
+void MetadataStorageFromDisk::createMetadataFile(const std::string & path, const std::string & blob_name, uint64_t size_in_bytes, MetadataTransactionPtr transaction)
+{
+    DiskObjectStorageMetadataPtr metadata = std::make_unique<DiskObjectStorageMetadata>(disk->getPath(), root_path_for_remote_metadata, path);
+    metadata->addObject(blob_name, size_in_bytes);
+    writeMetadataToFile(path, transaction, metadata->serializeToString());
+}
+
 void MetadataStorageFromDisk::addBlobToMetadata(const std::string & path, const std::string & blob_name, uint64_t size_in_bytes, MetadataTransactionPtr transaction)
 {
     DiskObjectStorageMetadataPtr metadata;

src/Disks/ObjectStorages/MetadataStorageFromDisk.h

@@ -47,7 +47,7 @@ private:
     mutable std::shared_mutex metadata_mutex;
 
 public:
-    explicit MetadataStorageFromDisk(DiskPtr disk_, const std::string & root_path_from_remote_metadata_ = "")
+    MetadataStorageFromDisk(DiskPtr disk_, const std::string & root_path_from_remote_metadata_)
         : disk(disk_)
         , root_path_for_remote_metadata(root_path_from_remote_metadata_)
     {
@@ -89,7 +89,9 @@ public:
          MetadataTransactionPtr transaction,
          const std::string & data) override;
 
-    void createMetadataFile(const std::string & path, MetadataTransactionPtr transaction) override;
+    void createEmptyMetadataFile(const std::string & path, MetadataTransactionPtr transaction) override;
+
+    void createMetadataFile(const std::string & path, const std::string & blob_name, uint64_t size_in_bytes, MetadataTransactionPtr transaction) override;
 
     void addBlobToMetadata(const std::string & path, const std::string & blob_name, uint64_t size_in_bytes, MetadataTransactionPtr transaction) override;
 

src/Disks/ObjectStorages/S3/registerDiskS3.cpp

@@ -86,7 +86,7 @@ void registerDiskS3(DiskFactory & factory)
 
         auto [metadata_path, metadata_disk] = prepareForLocalMetadata(name, config, config_prefix, context);
 
-        auto metadata_storage = std::make_shared<MetadataStorageFromDisk>(metadata_disk);
+        auto metadata_storage = std::make_shared<MetadataStorageFromDisk>(metadata_disk, uri.key);
 
         FileCachePtr cache = getCachePtrForDisk(name, config, config_prefix, context);