store per-file object_storage_ptr in object_info

zvonand · zvonand · commit 93296ec9f76b · 2025-09-24T21:47:01.000+02:00
tmp

upd

upd
diff --git a/src/Disks/ObjectStorages/IObjectStorage.h b/src/Disks/ObjectStorages/IObjectStorage.h
@@ -125,14 +125,16 @@ struct PathWithMetadata
     std::optional<ObjectMetadata> metadata;
     CommandInTaskResponse command;
     String absolute_path;
+    std::optional<ObjectStoragePtr> object_storage_to_use = std::nullopt;
 
     PathWithMetadata() = default;
 
-    explicit PathWithMetadata(const String & task_string, std::optional<ObjectMetadata> metadata_ = std::nullopt, String absolute_path_ = "")
+    explicit PathWithMetadata(const String & task_string, std::optional<ObjectMetadata> metadata_ = std::nullopt, String absolute_path_ = "", std::optional<ObjectStoragePtr> object_storage_to_use_ = std::nullopt)
         : metadata(std::move(metadata_))
         , command(task_string)
+        , absolute_path(absolute_path_)
+        , object_storage_to_use(object_storage_to_use_)
     {
-        absolute_path = absolute_path_;
         if (!command.is_parsed())
         {
             relative_path = task_string;
@@ -152,6 +154,8 @@ struct PathWithMetadata
 
     void loadMetadata(ObjectStoragePtr object_storage, bool ignore_non_existent_file);
     const CommandInTaskResponse & getCommand() const { return command; }
+
+    std::optional<ObjectStoragePtr> getObjectStorage() const { return object_storage_to_use; }
 };
 
 struct ObjectKeyWithMetadata
diff --git a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h
@@ -356,7 +356,7 @@ class StorageIcebergConfiguration : public StorageObjectStorage::Configuration,
     IDataLakeMetadata * getExternalMetadata() override { return getImpl().getExternalMetadata(); }
 
     std::shared_ptr<NamesAndTypesList> getInitialSchemaByPath(ContextPtr context, const String & path) const override
-        { return getImpl().getInitialSchemaByPath(context, path); }
+    { return getImpl().getInitialSchemaByPath(context, path); }
 
     std::shared_ptr<const ActionsDAG> getSchemaTransformer(ContextPtr context, const String & path) const override
         { return getImpl().getSchemaTransformer(context, path); }
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp
@@ -842,7 +842,7 @@ std::shared_ptr<NamesAndTypesList> IcebergMetadata::getInitialSchemaByPath(Conte
     }
 
     SharedLockGuard lock(mutex);
-    auto version_if_outdated = getSchemaVersionByFileIfOutdated(Iceberg::makeAbsolutePath(table_location, data_path));
+    auto version_if_outdated = getSchemaVersionByFileIfOutdated(data_path);
     return version_if_outdated.has_value() ? schema_processor.getClickhouseTableSchemaById(version_if_outdated.value()) : nullptr;
 }
 
@@ -856,7 +856,7 @@ std::shared_ptr<const ActionsDAG> IcebergMetadata::getSchemaTransformer(ContextP
     }
 
     SharedLockGuard lock(mutex);
-    auto version_if_outdated = getSchemaVersionByFileIfOutdated(Iceberg::makeAbsolutePath(table_location, data_path));
+    auto version_if_outdated = getSchemaVersionByFileIfOutdated(data_path);
     return version_if_outdated.has_value()
         ? schema_processor.getSchemaTransformationDagByIds(version_if_outdated.value(), relevant_snapshot_schema_id)
         : nullptr;
@@ -1327,7 +1327,7 @@ class IcebergKeysIterator : public IObjectIterator
         if (callback)
             callback(FileProgress(0, object_metadata.size_bytes));
 
-        return std::make_shared<ObjectInfo>(file_uri_parsed.path, std::move(object_metadata), file_uri);
+        return std::make_shared<ObjectInfo>(file_uri_parsed.path, std::move(object_metadata), file_uri, storage_to_use);
     }
 
 private:
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp
@@ -92,16 +92,15 @@ bool isRelativePath(const std::string & path)
     if (!extractStorageType(path).empty())
         return false;
 
-    // // Non-relative if it starts with '/' (absolute POSIX path)
-    // if (!path.empty() && path.front() == '/')
-    //     return false;
-
     return true;
 }
 
 
 UriParts parseUri(const std::string & uri)
 {
+    if (uri.empty())
+        return {};
+
     UriParts parts;
 
     // scheme://authority/path
@@ -123,34 +122,20 @@ UriParts parseUri(const std::string & uri)
         return parts;
     }
 
-    // scheme:/path or file:///path
-    if (!uri.empty() && uri.front() == '/')
+    // if part has no scheme and starts with '/' -- it is an absolute uri for local file: file:///path
+    if (uri.front() == '/')
     {
+        parts.scheme = "file";
         parts.path = std::string(uri);
         return parts;
     }
 
-    // Best-effort for "scheme:..." without //
-    auto colon = uri.find(':');
-    auto slash = uri.find('/');
-    if (colon != std::string_view::npos && (slash == std::string_view::npos || colon < slash))
-    {
-        parts.scheme = Poco::toLower(uri.substr(0, colon));
-        auto after = uri.substr(colon + 1);
-        // file:/path or file:///path -> treat as absolute path
-        if (!after.empty() && after.front() == '/')
-            parts.path = std::string(after);
-        else
-            parts.path = std::string(after);
-        return parts;
-    }
-
-    // Relative or plain path
+    // Relative path, return as is
     parts.path = std::string(uri);
     return parts;
 }
 
-std::string makeAbsolutePath(const std::string & table_location, const std::string & path)
+std::string makeAbsolutePath(const std::string & table_location, const std::string & path, [[maybe_unused]] bool is_s3_key)
 {
     if (!isRelativePath(path))
         return path;
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h
@@ -34,7 +34,7 @@ struct UriParts
 
 UriParts parseUri(const std::string & uri);
 
-std::string makeAbsolutePath(const std::string & table_location, const std::string & path);
+std::string makeAbsolutePath(const std::string & table_location, const std::string & path, bool is_s3_key = false);
 
 }
 
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
@@ -472,6 +472,8 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
     ObjectInfoPtr object_info;
     auto query_settings = configuration->getQuerySettings(context_);
 
+    ObjectStoragePtr storage_to_use = object_storage;
+
     bool not_a_path = false;
 
     do
@@ -499,7 +501,10 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
         if (object_info->getPath().empty())
             return {};
 
-        object_info->loadMetadata(object_storage, query_settings.ignore_non_existent_file);
+        if (auto storage_for_file = object_info->getObjectStorage(); storage_for_file.has_value())
+            storage_to_use = storage_for_file.value();
+
+        object_info->loadMetadata(storage_to_use, query_settings.ignore_non_existent_file);
     }
     while (not_a_path || (query_settings.skip_empty_files && object_info->metadata->size_bytes == 0));
 
@@ -552,7 +557,7 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
         else
         {
             compression_method = chooseCompressionMethod(object_info->getFileName(), configuration->getCompressionMethod());
-            read_buf = createReadBuffer(*object_info, object_storage, context_, log);
+            read_buf = createReadBuffer(*object_info, storage_to_use, context_, log);
         }
 
         Block initial_header = read_from_format_info.format_header;
@@ -644,14 +649,16 @@ std::unique_ptr<ReadBufferFromFileBase> StorageObjectStorageSource::createReadBu
     const auto & settings = context_->getSettingsRef();
     const auto & effective_read_settings = read_settings.has_value() ? read_settings.value() : context_->getReadSettings();
 
+    ObjectStoragePtr storage_to_use = object_info.getObjectStorage().value_or(object_storage);
+
     bool use_distributed_cache = false;
 #if ENABLE_DISTRIBUTED_CACHE
     ObjectStorageConnectionInfoPtr connection_info;
     if (settings[Setting::table_engine_read_through_distributed_cache]
         && DistributedCache::Registry::instance().isReady(
             effective_read_settings.distributed_cache_settings.read_only_from_current_az))
     {
-        connection_info = object_storage->getConnectionInfo();
+        connection_info = storage_to_use->getConnectionInfo();
         if (connection_info)
             use_distributed_cache = true;
     }
@@ -664,15 +671,15 @@ std::unique_ptr<ReadBufferFromFileBase> StorageObjectStorageSource::createReadBu
         filesystem_cache_name = settings[Setting::filesystem_cache_name].value;
         use_filesystem_cache = effective_read_settings.enable_filesystem_cache
             && !filesystem_cache_name.empty()
-            && (object_storage->getType() == ObjectStorageType::Azure
-                || object_storage->getType() == ObjectStorageType::S3);
+            && (storage_to_use->getType() == ObjectStorageType::Azure
+                || storage_to_use->getType() == ObjectStorageType::S3);
     }
 
     /// We need object metadata for two cases:
     /// 1. object size suggests whether we need to use prefetch
     /// 2. object etag suggests a cache key in case we use filesystem cache
     if (!object_info.metadata)
-        object_info.metadata = object_storage->getObjectMetadata(object_info.getPath());
+        object_info.metadata = storage_to_use->getObjectMetadata(object_info.getPath());
 
     const auto & object_size = object_info.metadata->size_bytes;
 
@@ -710,9 +717,9 @@ std::unique_ptr<ReadBufferFromFileBase> StorageObjectStorageSource::createReadBu
     {
         const std::string path = object_info.getPath();
         StoredObject object(path, "", object_size);
-        auto read_buffer_creator = [object, nested_buffer_read_settings, object_storage]()
+        auto read_buffer_creator = [object, nested_buffer_read_settings, storage_to_use]()
         {
-            return object_storage->readObject(object, nested_buffer_read_settings);
+            return storage_to_use->readObject(object, nested_buffer_read_settings);
         };
 
         impl = std::make_unique<ReadBufferFromDistributedCache>(
@@ -745,9 +752,9 @@ std::unique_ptr<ReadBufferFromFileBase> StorageObjectStorageSource::createReadBu
             const auto cache_key = FileCacheKey::fromKey(hash.get128());
             auto cache = FileCacheFactory::instance().get(filesystem_cache_name);
 
-            auto read_buffer_creator = [path = object_info.getPath(), object_size, modified_read_settings, object_storage]()
+            auto read_buffer_creator = [path = object_info.getPath(), object_size, modified_read_settings, storage_to_use]()
             {
-                return object_storage->readObject(StoredObject(path, "", object_size), modified_read_settings);
+                return storage_to_use->readObject(StoredObject(path, "", object_size), modified_read_settings);
             };
 
             modified_read_settings.filesystem_cache_boundary_alignment = settings[Setting::filesystem_cache_boundary_alignment];
@@ -777,7 +784,7 @@ std::unique_ptr<ReadBufferFromFileBase> StorageObjectStorageSource::createReadBu
     }
 
     if (!impl)
-        impl = object_storage->readObject(StoredObject(object_info.getPath(), "", object_size), modified_read_settings);
+        impl = storage_to_use->readObject(StoredObject(object_info.getPath(), "", object_size), modified_read_settings);
 
     if (!use_async_buffer)
         return impl;
@@ -1157,7 +1164,7 @@ StorageObjectStorageSource::ArchiveIterator::createArchiveReader(ObjectInfoPtr o
         /* path_to_archive */object_info->getPath(),
         /* archive_read_function */[=, this]()
         {
-            return StorageObjectStorageSource::createReadBuffer(*object_info, object_storage, getContext(), log);
+            return StorageObjectStorageSource::createReadBuffer(*object_info, object_info->getObjectStorage().value_or(object_storage), getContext(), log);
         },
         /* archive_size */size);
 }
@@ -1177,7 +1184,10 @@ ObjectInfoPtr StorageObjectStorageSource::ArchiveIterator::next(size_t processor
                     return {};
 
                 if (!archive_object->metadata)
-                    archive_object->metadata = object_storage->getObjectMetadata(archive_object->getPath());
+                {
+                    ObjectStoragePtr storage_to_use = archive_object->getObjectStorage().value_or(object_storage);
+                    archive_object->metadata = storage_to_use->getObjectMetadata(archive_object->getPath());
+                }
 
                 archive_reader = createArchiveReader(archive_object);
                 file_enumerator = archive_reader->firstFile();
@@ -1202,7 +1212,10 @@ ObjectInfoPtr StorageObjectStorageSource::ArchiveIterator::next(size_t processor
                 return {};
 
             if (!archive_object->metadata)
-                archive_object->metadata = object_storage->getObjectMetadata(archive_object->getPath());
+            {
+                ObjectStoragePtr storage_to_use = archive_object->getObjectStorage().value_or(object_storage);
+                archive_object->metadata = storage_to_use->getObjectMetadata(archive_object->getPath());
+            }
 
             archive_reader = createArchiveReader(archive_object);
             if (!archive_reader->fileExists(path_in_archive))

Original file line number	Diff line number	Diff line change
`@@ -842,7 +842,7 @@ std::shared_ptr<NamesAndTypesList> IcebergMetadata::getInitialSchemaByPath(Conte`
`842`	`842`	`}`
`843`	`843`
`844`	`844`	`SharedLockGuard lock(mutex);`
`845`		`- auto version_if_outdated = getSchemaVersionByFileIfOutdated(Iceberg::makeAbsolutePath(table_location, data_path));`
	`845`	`+ auto version_if_outdated = getSchemaVersionByFileIfOutdated(data_path);`
`846`	`846`	`return version_if_outdated.has_value() ? schema_processor.getClickhouseTableSchemaById(version_if_outdated.value()) : nullptr;`
`847`	`847`	`}`
`848`	`848`
`@@ -856,7 +856,7 @@ std::shared_ptr<const ActionsDAG> IcebergMetadata::getSchemaTransformer(ContextP`
`856`	`856`	`}`
`857`	`857`
`858`	`858`	`SharedLockGuard lock(mutex);`
`859`		`- auto version_if_outdated = getSchemaVersionByFileIfOutdated(Iceberg::makeAbsolutePath(table_location, data_path));`
	`859`	`+ auto version_if_outdated = getSchemaVersionByFileIfOutdated(data_path);`
`860`	`860`	`return version_if_outdated.has_value()`
`861`	`861`	`? schema_processor.getSchemaTransformationDagByIds(version_if_outdated.value(), relevant_snapshot_schema_id)`
`862`	`862`	`: nullptr;`
`@@ -1327,7 +1327,7 @@ class IcebergKeysIterator : public IObjectIterator`
`1327`	`1327`	`if (callback)`
`1328`	`1328`	`callback(FileProgress(0, object_metadata.size_bytes));`
`1329`	`1329`
`1330`		`- return std::make_shared<ObjectInfo>(file_uri_parsed.path, std::move(object_metadata), file_uri);`
	`1330`	`+ return std::make_shared<ObjectInfo>(file_uri_parsed.path, std::move(object_metadata), file_uri, storage_to_use);`
`1331`	`1331`	`}`
`1332`	`1332`
`1333`	`1333`	`private:`
Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,7 @@ struct UriParts`
`34`	`34`
`35`	`35`	`UriParts parseUri(const std::string & uri);`
`36`	`36`
`37`		`-std::string makeAbsolutePath(const std::string & table_location, const std::string & path);`
	`37`	`+std::string makeAbsolutePath(const std::string & table_location, const std::string & path, bool is_s3_key = false);`
`38`	`38`
`39`	`39`	`}`
`40`	`40`