YQL-20072: Support mrrun + yqlworker

MrLolthe1st · MrLolthe1st · commit 2e188b853dc1 · 2025-10-29T12:12:49.000+03:00
commit_hash:a70465d9a63f8a86c6990ae04e9c615827498fe0
diff --git a/yql/essentials/cfg/tests/fs_http.conf b/yql/essentials/cfg/tests/fs_http.conf
@@ -1,5 +1,5 @@
 MaxRedirects: 10
-SocketTimeoutMs: 300000
+TimeoutMs: 300000
 MaxRetries: 3
 
 MinDelayMs: 1000
diff --git a/yql/essentials/core/facade/yql_facade.cpp b/yql/essentials/core/facade/yql_facade.cpp
@@ -203,6 +203,10 @@ void TProgramFactory::SetCredentials(TCredentials::TPtr credentials) {
     Credentials_ = std::move(credentials);
 }
 
+void TProgramFactory::AddRemoteLayersProvider(const TString& alias, NLayers::IRemoteLayerProviderPtr provider) {
+    RemoteLayersProviders_.emplace(alias, std::move(provider));
+}
+
 void TProgramFactory::SetGatewaysConfig(const TGatewaysConfig* gatewaysConfig) {
     GatewaysConfig_ = gatewaysConfig;
 }
@@ -281,7 +285,7 @@ TProgramPtr TProgramFactory::Create(
                         LangVer_, MaxLangVer_, VolatileResults_, UserDataTable_, Credentials_, moduleResolver, urlListerManager,
                         udfResolver, udfIndex, udfIndexPackageSet, FileStorage_, UrlPreprocessing_,
                         GatewaysConfig_, filename, sourceCode, sessionId, Runner_, EnableRangeComputeFor_, ArrowResolver_, hiddenMode,
-                        qContext, gatewaysForMerge);
+                        qContext, gatewaysForMerge, RemoteLayersProviders_);
 }
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -314,7 +318,8 @@ TProgram::TProgram(
     const IArrowResolver::TPtr& arrowResolver,
     EHiddenMode hiddenMode,
     const TQContext& qContext,
-    TMaybe<TString> gatewaysForMerge)
+    TMaybe<TString> gatewaysForMerge,
+    THashMap<TString, NLayers::IRemoteLayerProviderPtr> remoteLayersProviders)
     : FunctionRegistry_(functionRegistry)
     , RandomProvider_(randomProvider)
     , TimeProvider_(timeProvider)
@@ -347,6 +352,7 @@ TProgram::TProgram(
     , HiddenMode_(hiddenMode)
     , QContext_(qContext)
     , GatewaysForMerge_(gatewaysForMerge)
+    , RemoteLayersProviders_(std::move(remoteLayersProviders))
 {
     if (SessionId_.empty()) {
         SessionId_ = CreateGuidAsString();
@@ -2073,6 +2079,9 @@ TTypeAnnotationContextPtr TProgram::BuildTypeAnnotationContext(const TString& us
     typeAnnotationContext->FileStorage = FileStorage_;
     typeAnnotationContext->QContext = QContext_;
     typeAnnotationContext->HiddenMode = HiddenMode_;
+    for (auto& [alias, provider] : RemoteLayersProviders_) {
+        typeAnnotationContext->AddRemoteLayersProvider(alias, provider);
+    }
 
     if (UdfIndex_ && UdfIndexPackageSet_) {
         // setup default versions at the beginning
diff --git a/yql/essentials/core/facade/yql_facade.h b/yql/essentials/core/facade/yql_facade.h
@@ -8,6 +8,7 @@
 #include <yql/essentials/core/url_preprocessing/interface/url_preprocessing.h>
 #include <yql/essentials/core/yql_type_annotation.h>
 #include <yql/essentials/core/yql_user_data.h>
+#include <yql/essentials/core/layers/remote_layer_provider.h>
 #include <yql/essentials/core/qplayer/storage/interface/yql_qstorage.h>
 #include <yql/essentials/providers/config/yql_config_provider.h>
 #include <yql/essentials/providers/result/provider/yql_result_provider.h>
@@ -65,6 +66,7 @@ class TProgramFactory: public TThrRefBase, private TMoveOnly {
     void EnableRangeComputeFor();
     void SetArrowResolver(IArrowResolver::TPtr arrowResolver);
     void SetUdfResolverLogfile(const TString& path);
+    void AddRemoteLayersProvider(const TString& alias, NLayers::IRemoteLayerProviderPtr provider);
 
     TProgramPtr Create(
         const TFile& file,
@@ -105,6 +107,7 @@ class TProgramFactory: public TThrRefBase, private TMoveOnly {
     bool EnableRangeComputeFor_ = false;
     IArrowResolver::TPtr ArrowResolver_;
     TMaybe<TString> UdfResolverLogfile_;
+    THashMap<TString, NLayers::IRemoteLayerProviderPtr> RemoteLayersProviders_;
 };
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -387,7 +390,8 @@ class TProgram: public TThrRefBase, private TNonCopyable {
         const IArrowResolver::TPtr& arrowResolver,
         EHiddenMode hiddenMode,
         const TQContext& qContext,
-        TMaybe<TString> gatewaysForMerge);
+        TMaybe<TString> gatewaysForMerge,
+        THashMap<TString, NLayers::IRemoteLayerProviderPtr> remoteLayersProviders);
 
     TTypeAnnotationContextPtr BuildTypeAnnotationContext(const TString& username);
     TTypeAnnotationContextPtr GetAnnotationContext() const;
@@ -490,6 +494,7 @@ class TProgram: public TThrRefBase, private TNonCopyable {
     TIssues FinalIssues_;
     TMaybe<TIssue> ParametersIssue_;
     bool EnableLineage_ = false;
+    THashMap<TString, NLayers::IRemoteLayerProviderPtr> RemoteLayersProviders_;
 };
 
 void UpdateSqlFlagsFromQContext(const TQContext& qContext, THashSet<TString>& flags, TMaybe<TString> gatewaysPatch = {});
diff --git a/yql/essentials/core/file_storage/file_storage_ut.cpp b/yql/essentials/core/file_storage/file_storage_ut.cpp
@@ -3,7 +3,7 @@
 #include <yql/essentials/utils/test_http_server/test_http_server.h>
 #include <yql/essentials/core/file_storage/proto/file_storage.pb.h>
 #include <yql/essentials/core/file_storage/http_download/http_download.h>
-#include <yql/essentials/core/file_storage/http_download/proto/http_download.pb.h>
+#include <yql/essentials/utils/fetch/proto/fetch_config.pb.h>
 
 #include <library/cpp/threading/future/future.h>
 #include <library/cpp/threading/future/async.h>
@@ -26,7 +26,7 @@ static TString ReadFileContent(const TString& path) {
     return TIFStream(path).ReadAll();
 }
 
-static TFileStoragePtr CreateTestFS(TFileStorageConfig params = {}, const THttpDownloaderConfig* httpCfg = nullptr) {
+static TFileStoragePtr CreateTestFS(TFileStorageConfig params = {}, const TFetchConfig* httpCfg = nullptr) {
     if (httpCfg) {
         TStringStream strCfg;
         SerializeToTextFormat(*httpCfg, strCfg);
@@ -550,8 +550,8 @@ Y_UNIT_TEST(SocketTimeout) {
         return TTestHttpServer::TReply::Ok("ABC");
     });
 
-    THttpDownloaderConfig httpCfg;
-    httpCfg.SetSocketTimeoutMs(1000);
+    TFetchConfig httpCfg;
+    httpCfg.SetTimeoutMs(1000);
     TFileStorageConfig params;
     TFileStoragePtr fs = CreateTestFS(params, &httpCfg);
 
diff --git a/yql/essentials/core/file_storage/http_download/http_download.cpp b/yql/essentials/core/file_storage/http_download/http_download.cpp
@@ -1,10 +1,10 @@
 #include "http_download.h"
 
 #include <yql/essentials/core/file_storage/proto/file_storage.pb.h>
-#include <yql/essentials/core/file_storage/http_download/proto/http_download.pb.h>
 #include <yql/essentials/core/file_storage/download/download_stream.h>
 #include <yql/essentials/core/file_storage/download/download_config.h>
 #include <yql/essentials/core/file_storage/defs/downloader.h>
+#include <yql/essentials/utils/fetch/proto/fetch_config.pb.h>
 #include <yql/essentials/utils/fetch/fetch.h>
 #include <yql/essentials/utils/log/context.h>
 #include <yql/essentials/utils/md5_stream.h>
@@ -22,7 +22,7 @@
 
 namespace NYql {
 
-class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderConfig>, public NYql::NFS::IDownloader {
+class THttpDownloader: public TDownloadConfig<THttpDownloader, TFetchConfig>, public NYql::NFS::IDownloader {
 public:
     THttpDownloader(const TFileStorageConfig& config)
         : UseFakeChecksums(GetEnv("YQL_LOCAL") == "1")
@@ -31,7 +31,7 @@ class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderCo
     }
     ~THttpDownloader() = default;
 
-    void DoConfigure(const THttpDownloaderConfig& cfg) {
+    void DoConfigure(const TFetchConfig& cfg) {
         Policy_ = IRetryPolicy<unsigned>::GetExponentialBackoffPolicy(
             DefaultClassifyHttpCode,
             TDuration::MilliSeconds(cfg.GetMinDelayMs()),
@@ -41,7 +41,7 @@ class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderCo
             TDuration::MilliSeconds(cfg.GetMaxTotalDelayTimeMs()),
             cfg.GetScale());
         Redirects_ = cfg.GetMaxRedirects();
-        SocketTimeoutMs = cfg.GetSocketTimeoutMs();
+        TimeoutMs = cfg.GetTimeoutMs();
     }
 
     bool Accept(const THttpURL& url) final {
@@ -56,7 +56,7 @@ class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderCo
     }
 
     std::tuple<NYql::NFS::TDataProvider, TString, TString> Download(const THttpURL& url, const TString& token, const TString& oldEtag, const TString& oldLastModified) final {
-        TFetchResultPtr fr1 = FetchWithETagAndLastModified(url, token, oldEtag, oldLastModified, SocketTimeoutMs, Redirects_, Policy_);
+        TFetchResultPtr fr1 = FetchWithETagAndLastModified(url, token, oldEtag, oldLastModified, TimeoutMs, Redirects_, Policy_);
         switch (fr1->GetRetCode()) {
             case HTTP_NOT_MODIFIED:
                 return std::make_tuple(NYql::NFS::TDataProvider{}, TString{}, TString{});
@@ -76,7 +76,7 @@ class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderCo
     }
 
 private:
-    static TFetchResultPtr FetchWithETagAndLastModified(const THttpURL& url, const TString& token, const TString& oldEtag, const TString& oldLastModified, ui32 socketTimeoutMs, size_t redirects, const IRetryPolicy<unsigned>::TPtr& policy) {
+    static TFetchResultPtr FetchWithETagAndLastModified(const THttpURL& url, const TString& token, const TString& oldEtag, const TString& oldLastModified, ui32 timeoutMs, size_t redirects, const IRetryPolicy<unsigned>::TPtr& policy) {
         // more details about ETag and ModifiedSince: https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.26
         THttpHeaders headers;
         if (!token.empty()) {
@@ -91,7 +91,7 @@ class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderCo
         }
 
         try {
-            return Fetch(url, headers, TDuration::MilliSeconds(socketTimeoutMs), redirects, policy);
+            return Fetch(url, headers, TDuration::MilliSeconds(timeoutMs), redirects, policy);
         } catch (const std::exception& e) {
             // remap exception type to leverage retry logic
             throw TDownloadError() << e.what();
@@ -159,7 +159,7 @@ class THttpDownloader: public TDownloadConfig<THttpDownloader, THttpDownloaderCo
 private:
     const bool UseFakeChecksums = false;
     IRetryPolicy<unsigned>::TPtr Policy_;
-    ui32 SocketTimeoutMs = 300000;
+    ui32 TimeoutMs = 300000;
     size_t Redirects_ = 10;
 };
 
diff --git a/yql/essentials/core/file_storage/http_download/ya.make b/yql/essentials/core/file_storage/http_download/ya.make
@@ -8,7 +8,6 @@ PEERDIR(
     yql/essentials/core/file_storage/defs
     yql/essentials/core/file_storage/download
     yql/essentials/core/file_storage/proto
-    yql/essentials/core/file_storage/http_download/proto
     yql/essentials/utils/fetch
     yql/essentials/utils/log
     yql/essentials/utils
diff --git a/yql/essentials/core/file_storage/ut/ya.make b/yql/essentials/core/file_storage/ut/ya.make
@@ -11,6 +11,7 @@ PEERDIR(
     library/cpp/threading/future
     library/cpp/deprecated/atomic
     yql/essentials/utils/test_http_server
+    yql/essentials/utils/fetch/proto
 )
 
 END()
diff --git a/yql/essentials/core/layers/layers.cpp b/yql/essentials/core/layers/layers.cpp
@@ -264,6 +264,21 @@ class TLayersRegistry: public ILayersRegistry {
     THashMap<TString, TLayerInfo> InfoByUrl_;
     THashMap<TString, TLogicalInfo> LogicalInfoByName_;
 };
+
+class TDummyRemoteLayerProvider: public IRemoteLayerProvider {
+public:
+    TDummyRemoteLayerProvider(TString errorMessage)
+        : ErrorMessage_(std::move(errorMessage))
+    {
+    }
+
+    NThreading::TFuture<TLayerInfo> GetLayerInfo(const TMaybe<TString>&, const TString&) const override {
+        return NThreading::MakeErrorFuture<TLayerInfo>(std::make_exception_ptr(yexception() << ErrorMessage_));
+    }
+
+private:
+    TString ErrorMessage_;
+};
 } // namespace
 
 namespace NYql::NLayers {
@@ -343,4 +358,9 @@ TMaybe<TVector<TLocations>> RemoveDuplicates(const TVector<std::pair<TKey, const
 ILayersRegistryPtr MakeLayersRegistry(const THashMap<TString, IRemoteLayerProviderPtr>& remoteProviders, const THashMap<TString, ILayersIntegrationPtr>& integrations) {
     return MakeIntrusive<TLayersRegistry>(remoteProviders, integrations);
 }
+
+IRemoteLayerProviderPtr MakeDummyRemoteLayerProvider(TString errorMessage) {
+    return MakeIntrusive<TDummyRemoteLayerProvider>(std::move(errorMessage));
+}
+
 } // namespace NYql::NLayers
diff --git a/yql/essentials/core/layers/remote_layer_provider.h b/yql/essentials/core/layers/remote_layer_provider.h
@@ -11,4 +11,7 @@ class IRemoteLayerProvider: public TThrRefBase {
 };
 
 using IRemoteLayerProviderPtr = TIntrusivePtr<IRemoteLayerProvider>;
+
+IRemoteLayerProviderPtr MakeDummyRemoteLayerProvider(TString errorMessage);
+
 } // namespace NYql::NLayers
diff --git a/yql/essentials/providers/common/proto/gateways_config.proto b/yql/essentials/providers/common/proto/gateways_config.proto
@@ -2,6 +2,7 @@ package NYql;
 option java_package = "ru.yandex.yql.proto";
 
 import "yql/essentials/protos/clickhouse.proto";
+import "yql/essentials/utils/fetch/proto/fetch_config.proto";
 
 
 /////////////////////////////// common ///////////////////////////////
@@ -933,6 +934,11 @@ message TFmrGatewayConfig {
     repeated TFmrConfig FmrConfigurations = 1;
 }
 
+message TLayersConfig {
+    optional string FetchUrl = 1;
+    optional TFetchConfig FetchConfig = 2;
+}
+
 /////////////////////////////// Root ///////////////////////////////
 
 message TGatewaysConfig {
@@ -961,4 +967,5 @@ message TGatewaysConfig {
     optional TDbResolverConfig DbResolver = 23;
     optional TYtflowGatewayConfig Ytflow = 24;
     optional TFmrGatewayConfig Fmr = 25;
+    optional TLayersConfig Layers = 26; 
 }
diff --git a/yql/essentials/providers/common/proto/ya.make b/yql/essentials/providers/common/proto/ya.make
@@ -7,6 +7,7 @@ SRCS(
 
 PEERDIR(
     yql/essentials/protos
+    yql/essentials/utils/fetch/proto
 )
 
 EXCLUDE_TAGS(GO_PROTO)
diff --git a/yql/essentials/tools/yql_facade_run/ya.make b/yql/essentials/tools/yql_facade_run/ya.make
@@ -48,6 +48,7 @@ PEERDIR(
     yql/essentials/sql
     yql/essentials/public/langver
     yql/essentials/core/langver
+    yql/essentials/core/layers
 
     library/cpp/resource
     library/cpp/getopt
diff --git a/yql/essentials/tools/yql_facade_run/yql_facade_run.cpp b/yql/essentials/tools/yql_facade_run/yql_facade_run.cpp
@@ -766,10 +766,16 @@ int TFacadeRunner::DoMain(int argc, const char* argv[]) {
     factory.SetGatewaysConfig(RunOptions_.GatewaysConfig.Get());
     factory.SetCredentials(RunOptions_.Credentials);
     factory.EnableRangeComputeFor();
+
     if (!urlListers.empty()) {
         factory.SetUrlListerManager(MakeUrlListerManager(urlListers));
     }
 
+    for (auto& factoryFn : RemoteLayersFactories_) {
+        auto result = factoryFn();
+        factory.AddRemoteLayersProvider(result.first, result.second);
+    }
+
     int result = DoRun(factory);
     if (result == 0 && EQPlayerMode::Capture == RunOptions_.QPlayerMode) {
         RunOptions_.QPlayerContext.GetWriter()->Commit().GetValueSync();
diff --git a/yql/essentials/tools/yql_facade_run/yql_facade_run.h b/yql/essentials/tools/yql_facade_run/yql_facade_run.h
@@ -9,6 +9,7 @@
 #include <yql/essentials/core/yql_user_data.h>
 #include <yql/essentials/core/facade/yql_facade.h>
 #include <yql/essentials/core/qplayer/storage/interface/yql_qstorage.h>
+#include <yql/essentials/core/layers/remote_layer_provider.h>
 #include <yql/essentials/public/langver/yql_langver.h>
 #include <yql/essentials/utils/log/log.h>
 
@@ -224,6 +225,10 @@ class TFacadeRunner {
         return RunOptions_;
     }
 
+    void AddRemoteLayersFactory(std::function<std::pair<TString, NLayers::IRemoteLayerProviderPtr>()>&& factory) {
+        RemoteLayersFactories_.emplace_back(std::move(factory));
+    }
+
 protected:
     virtual int DoMain(int argc, const char* argv[]);
     virtual int DoRun(TProgramFactory& factory);
@@ -234,6 +239,7 @@ class TFacadeRunner {
     std::vector<std::function<NFS::IDownloaderPtr()>> FsDownloadFactories_;
     std::vector<std::function<TDataProviderInitializer()>> ProviderFactories_;
     std::vector<std::function<IUrlListerPtr()>> UrlListerFactories_;
+    std::vector<std::function<std::pair<TString, NLayers::IRemoteLayerProviderPtr>()>> RemoteLayersFactories_;
     THashMap<TString, TString> ClusterMapping_;
     THolder<TFileStorageConfig> FileStorageConfig_;
     TFileStoragePtr FileStorage_;
diff --git a/yql/essentials/utils/fetch/fetch.cpp b/yql/essentials/utils/fetch/fetch.cpp
@@ -214,4 +214,20 @@ TFetchResultPtr Fetch(const THttpURL& url, const THttpHeaders& additionalHeaders
     ythrow yexception() << "Failed to fetch url '" << currentUrl.PrintS() << "': too many redirects";
 }
 
+THttpURL AppendUrlPath(const THttpURL& url, const TString& part) {
+    THttpURL resultUrl;
+    if (resultUrl.Parse(part, THttpURL::FeaturesDefault | THttpURL::FeatureSchemeKnown) != THttpURL::ParsedOK) {
+        throw yexception() << "url.Parse finished with not ParsedOK. Tried to parse `" << part << "`";
+    }
+    auto path = TString(url.Get(NUri::TField::FieldPath));
+    if (!path.EndsWith('/')) {
+        auto copy = url;
+        copy.Set(NUri::TField::FieldPath, path + "/");
+        resultUrl.Merge(copy);
+        return resultUrl;
+    }
+    resultUrl.Merge(url);
+    return resultUrl;
+}
+
 } // namespace NYql
diff --git a/yql/essentials/utils/fetch/fetch.h b/yql/essentials/utils/fetch/fetch.h
@@ -22,6 +22,7 @@ using TFetchResultPtr = TIntrusivePtr<IFetchResult>;
 THttpURL ParseURL(const TStringBuf addr);
 IRetryPolicy<unsigned>::TPtr GetDefaultPolicy();
 ERetryErrorClass DefaultClassifyHttpCode(unsigned code);
+THttpURL AppendUrlPath(const THttpURL& url, const TString& part);
 TFetchResultPtr Fetch(const THttpURL& url, const THttpHeaders& additionalHeaders = {}, const TDuration& timeout = TDuration::Max(), size_t redirects = 10, const IRetryPolicy<unsigned>::TPtr& policy = nullptr);
 
 } // namespace NYql
diff --git a/yql/essentials/utils/fetch/proto/fetch_config.proto b/yql/essentials/utils/fetch/proto/fetch_config.proto
@@ -1,8 +1,8 @@
 package NYql;
 
-message THttpDownloaderConfig {
+message TFetchConfig {
     optional uint32 MaxRedirects = 2 [default = 10];
-    optional uint32 SocketTimeoutMs = 3 [default = 300000];
+    optional uint32 TimeoutMs = 3 [default = 300000];
     optional uint32 MaxRetries = 4 [default = 3];
 
     optional uint32 MinDelayMs = 5 [default = 1000];
diff --git a/yql/essentials/utils/fetch/proto/ya.make b/yql/essentials/utils/fetch/proto/ya.make
@@ -1,7 +1,7 @@
 PROTO_LIBRARY()
 
 SRCS(
-    http_download.proto
+    fetch_config.proto
 )
 
 EXCLUDE_TAGS(GO_PROTO)
diff --git a/yql/essentials/utils/fetch/ya.make b/yql/essentials/utils/fetch/ya.make
@@ -12,6 +12,7 @@ PEERDIR(
     library/cpp/uri
     library/cpp/retry
     yql/essentials/utils/log
+    yql/essentials/utils/fetch/proto
 )
 
 END()

Original file line number	Diff line number	Diff line change
`@@ -11,6 +11,7 @@ PEERDIR(`
`11`	`11`	`library/cpp/threading/future`
`12`	`12`	`library/cpp/deprecated/atomic`
`13`	`13`	`yql/essentials/utils/test_http_server`
	`14`	`+ yql/essentials/utils/fetch/proto`
`14`	`15`	`)`
`15`	`16`
`16`	`17`	`END()`