Tune YDB <-> YQL deps

init
commit_hash:16572ab4e94aea4f7455c2ccb90b70ea99a412db

Author: vitstn

yql/essentials/core/dq_integration/transform/ya.make

@@ -0,0 +1,13 @@
+LIBRARY()
+
+SRCS(
+    yql_dq_task_transform.cpp
+)
+
+PEERDIR(
+    yql/essentials/minikql
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()

yql/essentials/core/dq_integration/transform/yql_dq_task_transform.cpp

@@ -0,0 +1,22 @@
+#include <yql/essentials/core/dq_integration/transform/yql_dq_task_transform.h>
+
+namespace NYql {
+
+TTaskTransformFactory CreateCompositeTaskTransformFactory(TVector<TTaskTransformFactory> factories) {
+    return [factories = std::move(factories)] (const THashMap<TString, TString>& taskParams, const NKikimr::NMiniKQL::IFunctionRegistry* funcRegistry) -> NKikimr::NMiniKQL::TCallableVisitFuncProvider {
+        TVector<NKikimr::NMiniKQL::TCallableVisitFuncProvider> funcProviders;
+        for (auto& factory: factories) {
+            funcProviders.push_back(factory(taskParams, funcRegistry));
+        }
+        return [funcProviders = std::move(funcProviders)] (const NKikimr::NMiniKQL::TInternName& name) -> NKikimr::NMiniKQL::TCallableVisitFunc {
+            for (auto& provider: funcProviders) {
+                if (auto res = provider(name)) {
+                    return res;
+                }
+            }
+            return {};
+        };
+    };
+}
+
+} // NYql

yql/essentials/core/dq_integration/transform/yql_dq_task_transform.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <yql/essentials/minikql/mkql_node_visitor.h>
+#include <yql/essentials/minikql/mkql_function_registry.h>
+
+#include <util/generic/hash.h>
+#include <util/generic/string.h>
+
+#include <functional>
+
+namespace NYql {
+
+using TTaskTransformFactory = std::function<NKikimr::NMiniKQL::TCallableVisitFuncProvider(const THashMap<TString, TString>&, const NKikimr::NMiniKQL::IFunctionRegistry*)>;
+
+TTaskTransformFactory CreateCompositeTaskTransformFactory(TVector<TTaskTransformFactory> factories);
+
+} // namespace NYql

yql/essentials/core/dq_integration/ya.make

@@ -0,0 +1,22 @@
+LIBRARY()
+
+SRCS(
+    yql_dq_integration.cpp
+    yql_dq_optimization.cpp
+)
+
+PEERDIR(
+    contrib/libs/protobuf
+    library/cpp/yson
+    yql/essentials/ast
+    yql/essentials/core
+    yql/essentials/core/expr_nodes
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()
+
+RECURSE(
+    transform
+)

yql/essentials/core/dq_integration/yql_dq_integration.cpp

@@ -0,0 +1,24 @@
+#include "yql_dq_integration.h"
+
+#include <yql/essentials/core/yql_type_annotation.h>
+
+namespace NYql {
+
+std::unordered_set<IDqIntegration*> GetUniqueIntegrations(const TTypeAnnotationContext& typesCtx) {
+    std::unordered_set<IDqIntegration*> uniqueIntegrations(typesCtx.DataSources.size() + typesCtx.DataSinks.size());
+    for (const auto& provider : typesCtx.DataSources) {
+        if (auto* dqIntegration = provider->GetDqIntegration()) {
+            uniqueIntegrations.emplace(dqIntegration);
+        }
+    }
+
+    for (const auto& provider : typesCtx.DataSinks) {
+        if (auto* dqIntegration = provider->GetDqIntegration()) {
+            uniqueIntegrations.emplace(dqIntegration);
+        }
+    }
+
+    return uniqueIntegrations;
+}
+
+} // namespace NYql

yql/essentials/core/dq_integration/yql_dq_integration.h

@@ -0,0 +1,84 @@
+#pragma once
+
+#include <yql/essentials/ast/yql_expr.h>
+#include <yql/essentials/core/yql_data_provider.h>
+#include <yql/essentials/core/yql_statistics.h>
+#include <yql/essentials/core/expr_nodes/yql_expr_nodes.h>
+#include <yql/essentials/public/issue/yql_issue.h>
+
+#include <library/cpp/yson/writer.h>
+
+#include <util/generic/string.h>
+#include <util/generic/vector.h>
+#include <util/generic/map.h>
+#include <util/generic/maybe.h>
+
+#include <google/protobuf/any.pb.h>
+
+namespace NJson {
+class TJsonValue;
+} // namespace NJson
+
+namespace NYql {
+
+struct TDqSettings;
+class TTransformationPipeline;
+
+namespace NCommon {
+    class TMkqlCallableCompilerBase;
+}
+
+class TFallbackError: public yexception {
+public:
+    TFallbackError(TIssuePtr issue = {})
+        : Issue_(std::move(issue))
+    {}
+
+    TIssuePtr GetIssue() const {
+        return Issue_;
+    }
+private:
+    TIssuePtr Issue_;
+};
+
+class IDqIntegration {
+public:
+    virtual ~IDqIntegration() {}
+
+    virtual ui64 Partition(const TDqSettings& config, size_t maxPartitions, const TExprNode& node,
+        TVector<TString>& partitions, TString* clusterName, TExprContext& ctx, bool canFallback) = 0;
+    virtual bool CheckPragmas(const TExprNode& node, TExprContext& ctx, bool skipIssues = false) = 0;
+    virtual bool CanRead(const TExprNode& read, TExprContext& ctx, bool skipIssues = true) = 0;
+    virtual TMaybe<ui64> EstimateReadSize(ui64 dataSizePerJob, ui32 maxTasksPerStage, const TVector<const TExprNode*>& nodes, TExprContext& ctx) = 0;
+    virtual TExprNode::TPtr WrapRead(const TDqSettings& config, const TExprNode::TPtr& read, TExprContext& ctx) = 0;
+    virtual TMaybe<TOptimizerStatistics> ReadStatistics(const TExprNode::TPtr& readWrap, TExprContext& ctx) = 0;
+    virtual TExprNode::TPtr RecaptureWrite(const TExprNode::TPtr& write, TExprContext& ctx) = 0;
+
+    // Nothing if callable is not for writing,
+    // false if callable is for writing and there are some errors (they are added to ctx),
+    // true if callable is for writing and no issues occured.
+    virtual TMaybe<bool> CanWrite(const TExprNode& write, TExprContext& ctx) = 0;
+
+    virtual TExprNode::TPtr WrapWrite(const TExprNode::TPtr& write, TExprContext& ctx) = 0;
+    virtual bool CanBlockRead(const NNodes::TExprBase& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx) = 0;
+    virtual void RegisterMkqlCompiler(NCommon::TMkqlCallableCompilerBase& compiler) = 0;
+    virtual bool CanFallback() = 0;
+    virtual void FillSourceSettings(const TExprNode& node, ::google::protobuf::Any& settings, TString& sourceType, size_t maxPartitions, TExprContext& ctx) = 0;
+    virtual void FillLookupSourceSettings(const TExprNode& node, ::google::protobuf::Any& settings, TString& sourceType) = 0;
+    virtual void FillSinkSettings(const TExprNode& node, ::google::protobuf::Any& settings, TString& sinkType) = 0;
+    virtual void FillTransformSettings(const TExprNode& node, ::google::protobuf::Any& settings) = 0;
+    virtual void Annotate(const TExprNode& node, THashMap<TString, TString>& params) = 0;
+    virtual bool PrepareFullResultTableParams(const TExprNode& root, TExprContext& ctx, THashMap<TString, TString>& params, THashMap<TString, TString>& secureParams) = 0;
+    virtual void WriteFullResultTableRef(NYson::TYsonWriter& writer, const TVector<TString>& columns, const THashMap<TString, TString>& graphParams) = 0;
+
+    // Fill plan operator properties for sources/sinks
+    // Return true if node was handled
+    virtual bool FillSourcePlanProperties(const NNodes::TExprBase& node, TMap<TString, NJson::TJsonValue>& properties) = 0;
+    virtual bool FillSinkPlanProperties(const NNodes::TExprBase& node, TMap<TString, NJson::TJsonValue>& properties) = 0;
+    // Called to configure DQ peephole
+    virtual void ConfigurePeepholePipeline(bool beforeDqTransforms, const THashMap<TString, TString>& params, TTransformationPipeline* pipeline) = 0;
+};
+
+std::unordered_set<IDqIntegration*> GetUniqueIntegrations(const TTypeAnnotationContext& typesCtx);
+
+} // namespace NYql

yql/essentials/core/dq_integration/yql_dq_optimization.cpp

@@ -0,0 +1 @@
+#include "yql_dq_optimization.h"

yql/essentials/core/dq_integration/yql_dq_optimization.h

@@ -0,0 +1,86 @@
+#pragma once
+
+#include <yql/essentials/ast/yql_expr.h>
+
+namespace NYql {
+
+class IDqOptimization {
+public:
+    virtual ~IDqOptimization() {}
+
+    /**
+        Rewrite DqReadWrap's underlying provider specific read callable
+        Args:
+            * read - provider specific read callable
+            * ctx - expr context
+        Returns one of:
+            * empty TPtr on error
+            * original `read`, if no changes
+            * new read, if any optimizations
+    */
+    virtual TExprNode::TPtr RewriteRead(const TExprNode::TPtr& read, TExprContext& ctx) = 0;
+
+    /**
+        Rewrite DqReadWrap's underlying provider specific read callable for lookup
+        Args:
+            * read - provider specific read callable
+            * ctx - expr context
+        Returns of of:
+            * empty TPtr, if lookup read is not supported
+            * DqLookupSourceWrap callable
+    */
+    virtual TExprNode::TPtr RewriteLookupRead(const TExprNode::TPtr& read, TExprContext& ctx) = 0;
+
+    /**
+        Apply new members subset for DqReadWrap's underlying provider specific read callable
+        Args:
+            * read - provider specific read callable
+            * members - expr list of atoms with new members
+            * ctx - expr context
+        Returns one of:
+            * empty TPtr on error
+            * original `read`, if no changes
+            * new read with applyed new members
+    */
+    virtual TExprNode::TPtr ApplyExtractMembers(const TExprNode::TPtr& read, const TExprNode::TPtr& members, TExprContext& ctx) = 0;
+
+    /**
+        Apply `take` or `skip` setting for DqReadWrap's underlying provider specific read callable
+        Args:
+            * read - provider specific read callable
+            * countBase - `Take`, `Skip` or `Limit` callable
+            * ctx - expr context
+        Returns one of:
+            * empty TPtr on error
+            * original `read`, if no changes
+            * new read with applyed setting
+    */
+    virtual TExprNode::TPtr ApplyTakeOrSkip(const TExprNode::TPtr& read, const TExprNode::TPtr& countBase, TExprContext& ctx) = 0;
+
+    /**
+        Apply `unordered` setting for DqReadWrap's underlying provider specific read callable
+        Args:
+            * read - provider specific read callable
+            * ctx - expr context
+        Returns one of:
+            * empty TPtr on error
+            * original `read`, if no changes
+            * new read with applyed setting
+    */
+    virtual TExprNode::TPtr ApplyUnordered(const TExprNode::TPtr& read, TExprContext& ctx) = 0;
+
+    /**
+        Optimize list/stream extend for set of DqReadWrap's underlying provider specific read callable
+        Args:
+            * listOfRead - expr list of provider specific read callables
+            * ordered - `true` for ordered extend (must keep original order of reads); `false`, if reads may be reordred due the optimization
+            * ctx - expr context
+        Returns one of:
+            * empty list on error
+            * original `listOfRead`, if no changes
+            * new optimized list of reads. Returned list length must not be greater than original `listOfRead` length
+    */
+    virtual TExprNode::TListType ApplyExtend(const TExprNode::TListType& listOfRead, bool ordered, TExprContext& ctx) = 0;
+};
+
+} // namespace NYql

yql/essentials/core/ya.make

@@ -98,6 +98,7 @@ END()
 RECURSE(
     cbo
     credentials
+    dq_integration
     file_storage
     issue
     minsketch