another cleaned attempt

fix schema id cache population

correct splitting of URI into parts

tmp: lazy way to debug cluster

fix storage schema normalization

store per-file object_storage_ptr in object_info

tmp

upd

upd

fix for local

make new storages properly

remove cloneObjectStorage

Author: zvonand

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
@@ -291,14 +295,6 @@ class IObjectStorage
     /// buckets in S3. If object storage doesn't have any namepaces return empty string.
     virtual String getObjectsNamespace() const = 0;
 
-    virtual std::unique_ptr<IObjectStorage> cloneObjectStorage(
-    const std::string &,
-    const Poco::Util::AbstractConfiguration &,
-    const std::string &, ContextPtr)
-    {
-        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method 'cloneObjectStorage' is not implemented");
-    }
-
     /// Generate blob name for passed absolute local path.
     /// Path can be generated either independently or based on `path`.
     virtual ObjectStorageKey generateObjectKeyForPath(const std::string & path, const std::optional<std::string> & key_prefix) const = 0;

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); }

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp

@@ -866,7 +866,12 @@ std::optional<Int32> IcebergMetadata::getSchemaVersionByFileIfOutdated(String da
 {
     auto schema_id_it = schema_id_by_data_file.find(data_path);
     if (schema_id_it == schema_id_by_data_file.end())
-        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot find manifest file for data file: {}", data_path);
+    {
+        std::string error_msg = "";
+        for (const auto & sch : schema_id_by_data_file)
+            error_msg += "Schema id: " + std::to_string(sch.second) + " for file: " + sch.first + "\n";
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot find manifest file for data file: {}.\n Contents:\n{}", data_path, error_msg);
+    }
 
     auto schema_id = schema_id_it->second;
     if (schema_id == relevant_snapshot_schema_id)
@@ -908,7 +913,7 @@ void IcebergMetadata::initializeSchemasFromManifestList(ContextPtr local_context
         for (const auto & manifest_file_entry : manifest_file_ptr->getFiles())
         {
             if (std::holds_alternative<DataFileEntry>(manifest_file_entry.file))
-                schema_id_by_data_file.emplace(std::get<DataFileEntry>(manifest_file_entry.file).file_name, manifest_file_ptr->getSchemaId());
+                schema_id_by_data_file.emplace(Iceberg::makeAbsolutePath(table_location, std::get<DataFileEntry>(manifest_file_entry.file).file_name), manifest_file_ptr->getSchemaId());
         }
     }
 
@@ -921,56 +926,32 @@ ManifestFileCacheKeys IcebergMetadata::getManifestList(ContextPtr local_context,
     if (configuration_ptr == nullptr)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Configuration is expired");
 
+    const String full_filename = Iceberg::makeAbsolutePath(table_location, filename);
+
     auto create_fn = [&]()
     {
-        auto base_table_uri_parsed = Iceberg::parseUri(table_location);
-        auto file_uri_parsed = Iceberg::parseUri(filename);
-
-        ObjectStoragePtr storage_to_use = object_storage;
-        String key_or_path = filename;
-
-        if (!file_uri_parsed.scheme.empty() && !base_table_uri_parsed.scheme.empty() && file_uri_parsed.scheme == base_table_uri_parsed.scheme)
-        {
-            if (file_uri_parsed.authority == base_table_uri_parsed.authority)
-            {
-                // Same namespace as table_location -> use primary storage and strip leading '/'
-                key_or_path = file_uri_parsed.path;
-                if (!key_or_path.empty() && key_or_path.front() == '/')
-                    key_or_path.erase(0, 1);
-            }
-            else if (!file_uri_parsed.scheme.empty() && file_uri_parsed.scheme == base_table_uri_parsed.scheme && !file_uri_parsed.authority.empty())
-            {
-                // Same scheme, different authority (e.g. another S3 bucket) -> clone storage for that authority
-                std::unique_ptr<IObjectStorage> cloned = object_storage->cloneObjectStorage(
-                    file_uri_parsed.authority, local_context->getConfigRef(), configuration_ptr->getTypeName() + ".", local_context);
-                storage_to_use = ObjectStoragePtr(cloned.release());
-
-                key_or_path = file_uri_parsed.path;
-                if (!key_or_path.empty() && key_or_path.front() == '/')
-                    key_or_path.erase(0, 1);
-            }
-        }
-        // TODO: what if storage type is different?
+        auto [storage_to_use, key] = Iceberg::resolveObjectStorageForPath(
+            table_location, filename, object_storage, secondary_storages, local_context);
 
-        StorageObjectStorage::ObjectInfo object_info(key_or_path, std::nullopt, filename);
+        StorageObjectStorage::ObjectInfo object_info(key, std::nullopt, full_filename);
 
         auto read_settings = local_context->getReadSettings();
         /// Do not utilize filesystem cache if more precise cache enabled
         if (manifest_cache)
             read_settings.enable_filesystem_cache = false;
 
         auto manifest_list_buf = StorageObjectStorageSource::createReadBuffer(object_info, storage_to_use, local_context, log, read_settings);
-        AvroForIcebergDeserializer manifest_list_deserializer(std::move(manifest_list_buf), file_uri_parsed.path, getFormatSettings(local_context));
+        AvroForIcebergDeserializer manifest_list_deserializer(std::move(manifest_list_buf), key, getFormatSettings(local_context));
 
         ManifestFileCacheKeys manifest_file_cache_keys;
 
         for (size_t i = 0; i < manifest_list_deserializer.rows(); ++i)
         {
-            const std::string manifest_file_name = manifest_list_deserializer.getValueFromRowByName(i, f_manifest_path, TypeIndex::String).safeGet<std::string>();
+            const std::string manifest_file_path = Iceberg::makeAbsolutePath(table_location, manifest_list_deserializer.getValueFromRowByName(i, f_manifest_path, TypeIndex::String).safeGet<std::string>());
             Int64 added_sequence_number = 0;
             if (format_version > 1)
                 added_sequence_number = manifest_list_deserializer.getValueFromRowByName(i, f_sequence_number, TypeIndex::Int64).safeGet<Int64>();
-            manifest_file_cache_keys.emplace_back(manifest_file_name, added_sequence_number);
+            manifest_file_cache_keys.emplace_back(manifest_file_path, added_sequence_number);
         }
         /// We only return the list of {file name, seq number} for cache.
         /// Because ManifestList holds a list of ManifestFilePtr which consume much memory space.
@@ -980,7 +961,7 @@ ManifestFileCacheKeys IcebergMetadata::getManifestList(ContextPtr local_context,
 
     ManifestFileCacheKeys manifest_file_cache_keys;
     if (manifest_cache)
-        manifest_file_cache_keys = manifest_cache->getOrSetManifestFileCacheKeys(IcebergMetadataFilesCache::getKey(configuration_ptr, filename), create_fn);
+        manifest_file_cache_keys = manifest_cache->getOrSetManifestFileCacheKeys(IcebergMetadataFilesCache::getKey(configuration_ptr, full_filename), create_fn);
     else
         manifest_file_cache_keys = create_fn();
     return manifest_file_cache_keys;
@@ -1080,46 +1061,25 @@ ManifestFilePtr IcebergMetadata::getManifestFile(ContextPtr local_context, const
 {
     auto configuration_ptr = configuration.lock();
 
+    const String full_filename = Iceberg::makeAbsolutePath(table_location, filename);
+
     auto create_fn = [&]()
     {
-        auto base_table_uri_parsed = Iceberg::parseUri(table_location);
-        auto file_uri_parsed = Iceberg::parseUri(filename);
+        // Select proper storage and key for the manifest file
+        auto [storage_to_use, key] = Iceberg::resolveObjectStorageForPath(
+            table_location, full_filename, object_storage, secondary_storages, local_context);
 
-        ObjectStoragePtr storage_to_use = object_storage;
-        String key_or_path = filename;
 
-        if (!file_uri_parsed.scheme.empty() && !base_table_uri_parsed.scheme.empty() && file_uri_parsed.scheme == base_table_uri_parsed.scheme)
-        {
-            if (file_uri_parsed.authority == base_table_uri_parsed.authority)
-            {
-                // Same namespace as table_location -> use primary storage and strip leading '/'
-                key_or_path = file_uri_parsed.path;
-                if (!key_or_path.empty() && key_or_path.front() == '/')
-                    key_or_path.erase(0, 1);
-            }
-            else if (!file_uri_parsed.scheme.empty() && file_uri_parsed.scheme == base_table_uri_parsed.scheme && !file_uri_parsed.authority.empty())
-            {
-                // Same scheme, different authority (e.g. another S3 bucket) -> clone storage for that authority
-                std::unique_ptr<IObjectStorage> cloned = object_storage->cloneObjectStorage(
-                    file_uri_parsed.authority, local_context->getConfigRef(), configuration_ptr->getTypeName() + ".", local_context);
-                storage_to_use = ObjectStoragePtr(cloned.release());
-
-                key_or_path = file_uri_parsed.path;
-                if (!key_or_path.empty() && key_or_path.front() == '/')
-                    key_or_path.erase(0, 1);
-            }
-        }
-
-        ObjectInfo manifest_object_info(key_or_path, std::nullopt, filename);
+        ObjectInfo manifest_object_info(key, std::nullopt, full_filename);
 
         auto read_settings = local_context->getReadSettings();
         /// Do not utilize filesystem cache if more precise cache enabled
         if (manifest_cache)
             read_settings.enable_filesystem_cache = false;
 
         auto buffer = StorageObjectStorageSource::createReadBuffer(manifest_object_info, storage_to_use, local_context, log, read_settings);
-        AvroForIcebergDeserializer manifest_file_deserializer(std::move(buffer), filename, getFormatSettings(local_context));
-        auto [schema_id, schema_object] = parseTableSchemaFromManifestFile(manifest_file_deserializer, filename);
+        AvroForIcebergDeserializer manifest_file_deserializer(std::move(buffer), full_filename, getFormatSettings(local_context));
+        auto [schema_id, schema_object] = parseTableSchemaFromManifestFile(manifest_file_deserializer, full_filename);
         schema_processor.addIcebergTableSchema(schema_object);
         return std::make_shared<ManifestFileContent>(
             manifest_file_deserializer,
@@ -1128,12 +1088,13 @@ ManifestFilePtr IcebergMetadata::getManifestFile(ContextPtr local_context, const
             schema_object,
             schema_processor,
             inherited_sequence_number,
+            table_location,
             local_context);
     };
 
     if (manifest_cache)
     {
-        auto manifest_file = manifest_cache->getOrSetManifestFile(IcebergMetadataFilesCache::getKey(configuration_ptr, filename), create_fn);
+        auto manifest_file = manifest_cache->getOrSetManifestFile(IcebergMetadataFilesCache::getKey(configuration_ptr, full_filename), create_fn);
         schema_processor.addIcebergTableSchema(manifest_file->getSchemaObject());
         return manifest_file;
     }
@@ -1275,12 +1236,14 @@ class IcebergKeysIterator : public IObjectIterator
 public:
     IcebergKeysIterator(
         Strings && data_files_,
+        const std::string & table_location_,
         ObjectStoragePtr object_storage_,
         std::map<String, ObjectStoragePtr> & secondary_storages_,
         IDataLakeMetadata::FileProgressCallback callback_,
         ContextPtr local_context_,
         StorageObjectStorage::ConfigurationPtr configuration_ptr_)
         : data_files(data_files_)
+        , table_location(table_location_)
         , object_storage(object_storage_)
         , secondary_storages(secondary_storages_)
         , callback(callback_)
@@ -1300,33 +1263,23 @@ class IcebergKeysIterator : public IObjectIterator
         if (current_index >= data_files.size())
             return nullptr;
 
-        auto file_uri = data_files[current_index];
-        auto file_uri_parsed = Iceberg::parseUri(file_uri);
-
-        ObjectStoragePtr storage_to_use;
+        const auto & raw_path = data_files[current_index];
 
-        if (file_uri_parsed.authority == object_storage->getObjectsNamespace())
-            storage_to_use = object_storage;
-        else if (secondary_storages.contains(file_uri_parsed.authority))
-            storage_to_use = secondary_storages.at(file_uri_parsed.authority);
-        else
-        {
-            std::unique_ptr<IObjectStorage> cloned = object_storage->cloneObjectStorage(
-                file_uri_parsed.authority, local_context->getConfigRef(), configuration_ptr->getTypeName() + ".", local_context);
-            storage_to_use = ObjectStoragePtr(cloned.release());
-            secondary_storages[file_uri_parsed.authority] = storage_to_use;
-        }
+        // Route to correct storage
+        auto [storage_to_use, key] = Iceberg::resolveObjectStorageForPath(
+            table_location, raw_path, object_storage, secondary_storages, local_context);
 
-        auto object_metadata = storage_to_use->getObjectMetadata(file_uri_parsed.path);
+        auto object_metadata = storage_to_use->getObjectMetadata(key);
 
         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>(key, std::move(object_metadata), raw_path, storage_to_use);
     }
 
 private:
     Strings data_files;
+    const String table_location;
     ObjectStoragePtr object_storage;
     std::map<String, ObjectStoragePtr> & secondary_storages;
     std::atomic<size_t> index = 0;
@@ -1342,7 +1295,7 @@ ObjectIterator IcebergMetadata::createIcebergKeysIterator(
     IDataLakeMetadata::FileProgressCallback callback_,
     ContextPtr local_context)
 {
-    return std::make_shared<IcebergKeysIterator>(std::move(data_files_), object_storage, secondary_storages, callback_, local_context, configuration.lock());
+    return std::make_shared<IcebergKeysIterator>(std::move(data_files_), table_location, object_storage, secondary_storages, callback_, local_context, configuration.lock());
 }
 
 ObjectIterator IcebergMetadata::iterate(

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp

@@ -126,8 +126,10 @@ ManifestFileContent::ManifestFileContent(
     Poco::JSON::Object::Ptr schema_object_,
     const IcebergSchemaProcessor & schema_processor,
     Int64 inherited_sequence_number,
+    std::string table_location_,
     DB::ContextPtr context)
 {
+    this->table_location = std::move(table_location_);
     this->schema_id = schema_id_;
     this->schema_object = schema_object_;
 
@@ -280,7 +282,7 @@ ManifestFileContent::ManifestFileContent(
             columns_infos[column_id].hyperrectangle.emplace(*left, true, *right, true);
         }
 
-        FileEntry file = FileEntry{DataFileEntry{file_path}};
+        FileEntry file = FileEntry{DataFileEntry{Iceberg::makeAbsolutePath(table_location, file_path)}};
 
         Int64 added_sequence_number = 0;
         if (format_version_ > 1)

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h

@@ -92,6 +92,7 @@ class ManifestFileContent
         Poco::JSON::Object::Ptr schema_object_,
         const DB::IcebergSchemaProcessor & schema_processor,
         Int64 inherited_sequence_number,
+         std::string table_location_,
         DB::ContextPtr context);
 
     const std::vector<ManifestFileEntry> & getFiles() const;
@@ -121,6 +122,8 @@ class ManifestFileContent
 
     std::set<Int32> column_ids_which_have_bounds;
 
+    std::string table_location;
+
 };
 
 using ManifestFilePtr = std::shared_ptr<const ManifestFileContent>;