tmp

Author: zvonand

src/Storages/ObjectStorage/StorageObjectStorageSource.cpp

@@ -1283,6 +1283,7 @@ StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator(
             continue;
         }
 
+        /// Sometimes, raw is {"retry_after_us":499871}. Then skip?????
         std::cerr << "\npath: " << raw << "\n";
         if (object_storage)
             std::cerr << "\nstorage: " << object_storage->getName() << ", " << object_storage->getObjectsNamespace() << "\n";
@@ -1294,7 +1295,10 @@ StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator(
             getContext()
             );
 
-        buffer.emplace_back(std::make_shared<ObjectInfo>(key, std::nullopt, raw, storage_to_use));
+        if (key.empty()) /// Not a valid key/path, maybe it is "retry_after_us". Store as is.
+            buffer.emplace_back(std::make_shared<ObjectInfo>(raw, std::nullopt));
+        else
+            buffer.emplace_back(std::make_shared<ObjectInfo>(key, std::nullopt, raw, storage_to_use));
     }
 }
 

src/Storages/ObjectStorage/Utils.cpp

@@ -258,30 +258,53 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
     const DB::ContextPtr & context)
 {
     if (isRelativePath(path))
+    {
+        if (base_storage)
+            std::cerr << "\nRelative path: " << base_storage->getName() << ", " << base_storage->getObjectsNamespace() << "\n";
+        else
+            std::cerr << "\nRelative, returning base, which is: NULL\n";
         return {base_storage, path}; // Relative path definitely goes to base storage
+    }
 
     SchemeAuthorityKey base{table_location};
     SchemeAuthorityKey target{path};
 
     if (target.scheme.empty())
+    {
+        if (base_storage)
+            std::cerr << "\nNo scheme, returning base, which is: " << base_storage->getName() << ", " << base_storage->getObjectsNamespace() << "\n";
+        else
+
+            std::cerr << "\nNo scheme, returning base, which is: NULL\n";
         return {base_storage, target.key};
+    }
 
     const std::string base_norm = normalizeSchema(base.scheme);
     const std::string target_norm = normalizeSchema(target.scheme);
 
     // Reuse base storage if scheme and authority (bucket) matches
-    if (!base.scheme.empty()
-        && base_norm == target_norm
-        && base.authority == target.authority)
+    if (base_norm == target_norm && base.authority == target.authority)
     {
+        if (base_storage)
+            std::cerr << "\nSame location, returning base, which is: " << base_storage->getName() << ", " << base_storage->getObjectsNamespace() << "\n";
+        else
+            std::cerr << "\nSame location, returning base, which is: NULL\n";
         return {base_storage, target.key};
     }
 
     const std::string cache_key = endpoint_cache_key(target_norm, target.authority);
     if (auto it = secondary_storages.find(cache_key); it != secondary_storages.end())
+    {
+        if (it->second)
+            std::cerr << "\nfound in cache: " << it->second->getName() << ", " << it->second->getObjectsNamespace() << "\n";
+        else
+            std::cerr << "\nfound in cache: NULL\n";
         return {it->second, target.key};
+    }
+
+    /// TODO: maybe do not invent new configuration. Use old one and clean up later
+    Poco::AutoPtr<Poco::Util::MapConfiguration> cfg(new Poco::Util::MapConfiguration);
 
-    Poco::AutoPtr<Poco::Util::MapConfiguration> cfg;
     const std::string type_for_factory = factoryTypeForScheme(target_norm);
     if (type_for_factory.empty())
         throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unsupported storage scheme '{}' in path '{}'", target_norm, path);
@@ -292,12 +315,12 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
 
     if (target_norm == "s3" || target_norm == "abfs")
     {
-        cfg->setString(config_prefix + ".endpoint", target.authority);
+        cfg->setString(config_prefix + ".endpoint", target_norm + "://" + target.authority);
     }
     else if (target_norm == "hdfs")
     {
         // HDFS endpoint must end with '/'
-        auto endpoint = target.authority;
+        auto endpoint = target_norm + "://" + target.authority;
         if (!endpoint.empty() && endpoint.back() != '/')
             endpoint.push_back('/');
         cfg->setString(config_prefix + ".endpoint", endpoint);
@@ -310,6 +333,10 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
     DB::ObjectStoragePtr storage = factory.create(cache_key, *cfg, config_prefix, context, /*skip_access_check*/ true);
 
     secondary_storages.emplace(cache_key, storage);
+    if (storage)
+        std::cerr << "\ncreated new storage: " << storage->getName() << ", " << storage->getObjectsNamespace() << "\n";
+    else
+        std::cerr << "\ncreated new storage: it is NULL\n";
     return {storage, target.key};
 }