Change interface

Author: rvu1024

ydb/library/yql/core/services/yql_transform_pipeline.cpp

@@ -39,11 +39,6 @@ TTransformationPipeline& TTransformationPipeline::Add(IGraphTransformer& transfo
     return *this;
 }
 
-TTransformationPipeline& TTransformationPipeline::Add(TTransformStage&& stage) {
-    Transformers_.push_back(std::move(stage));
-    return *this;
-}
-
 TTransformationPipeline& TTransformationPipeline::AddServiceTransformers(EYqlIssueCode issueCode) {
     Transformers_.push_back(TTransformStage(CreateGcNodeTransformer(), "GC", issueCode));
     return *this;

ydb/library/yql/core/services/yql_transform_pipeline.h

@@ -47,7 +47,6 @@ class TTransformationPipeline
         EYqlIssueCode issueCode = TIssuesIds::DEFAULT_ERROR, const TString& issueMessage = {});
     TTransformationPipeline& Add(IGraphTransformer& transformer, const TString& stageName,
         EYqlIssueCode issueCode = TIssuesIds::DEFAULT_ERROR, const TString& issueMessage = {});
-    TTransformationPipeline& Add(TTransformStage&& stage);
 
     TAutoPtr<IGraphTransformer> Build(bool useIssueScopes = true);
     TAutoPtr<IGraphTransformer> BuildWithNoArgChecks(bool useIssueScopes = true);

ydb/library/yql/dq/integration/yql_dq_integration.h

@@ -2,7 +2,6 @@
 
 #include <ydb/library/yql/ast/yql_expr.h>
 #include <ydb/library/yql/core/yql_data_provider.h>
-#include <ydb/library/yql/core/yql_graph_transformer.h>
 #include <ydb/library/yql/core/yql_statistics.h>
 #include <ydb/library/yql/dq/tasks/dq_tasks_graph.h>
 #include <ydb/library/yql/public/issue/yql_issue.h>
@@ -23,6 +22,7 @@ class TJsonValue;
 namespace NYql {
 
 struct TDqSettings;
+class TTransformationPipeline;
 
 namespace NCommon {
     class TMkqlCallableCompilerBase;
@@ -73,8 +73,8 @@ class IDqIntegration {
     // 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;
-    // This transformer will be called before DQ peephole transformations
-    virtual std::vector<TTransformStage> GetPeepholeTransforms(bool beforeDq, const THashMap<TString, TString>& params) = 0;
+    // Called to configure DQ peephole
+    virtual void ConfigurePeepholePipeline(bool beforeDqTransforms, const THashMap<TString, TString>& params, TTransformationPipeline* pipeline) = 0;
 };
 
 } // namespace NYql

ydb/library/yql/providers/common/dq/yql_dq_integration_impl.cpp

@@ -92,8 +92,7 @@ bool TDqIntegrationBase::FillSinkPlanProperties(const NNodes::TExprBase&, TMap<T
     return false;
 }
 
-std::vector<TTransformStage> TDqIntegrationBase::GetPeepholeTransforms(bool, const THashMap<TString, TString>&) {
-    return {};
+void TDqIntegrationBase::ConfigurePeepholePipeline(bool, const THashMap<TString, TString>&, TTransformationPipeline*) {
 }
 
 } // namespace NYql

ydb/library/yql/providers/common/dq/yql_dq_integration_impl.h

@@ -26,7 +26,7 @@ class TDqIntegrationBase: public IDqIntegration {
     void WriteFullResultTableRef(NYson::TYsonWriter& writer, const TVector<TString>& columns, const THashMap<TString, TString>& graphParams) override;
     bool FillSourcePlanProperties(const NNodes::TExprBase& node, TMap<TString, NJson::TJsonValue>& properties) override;
     bool FillSinkPlanProperties(const NNodes::TExprBase& node, TMap<TString, NJson::TJsonValue>& properties) override;
-    std::vector<TTransformStage> GetPeepholeTransforms(bool beforeDq, const THashMap<TString, TString>& params) override;
+    void ConfigurePeepholePipeline(bool beforeDqTransforms, const THashMap<TString, TString>& params, TTransformationPipeline* pipeline) override;
 
 protected:
     bool CanBlockReadTypes(const TStructExprType* node);

ydb/library/yql/providers/dq/provider/exec/yql_dq_exectransformer.cpp

@@ -243,13 +243,14 @@ struct TDqsPipelineConfigurator : public IPipelineConfigurator {
     void AfterCreate(TTransformationPipeline*) const final {}
 
     void AfterTypeAnnotation(TTransformationPipeline* pipeline) const final {
+        // First truncate graph by calculated precomputes
+        pipeline->Add(NDqs::CreateDqsReplacePrecomputesTransformer(*pipeline->GetTypeAnnotationContext(), State_->FunctionRegistry), "ReplacePrecomputes");
+
+        // Then apply provider specific transformers on truncated graph
         std::for_each(UniqIntegrations_.cbegin(), UniqIntegrations_.cend(), [&](const auto dqInt) {
-            for (auto& stage: dqInt->GetPeepholeTransforms(true, ProviderParams_)) {
-                pipeline->Add(std::move(stage));
-            }
+            dqInt->ConfigurePeepholePipeline(true, ProviderParams_, pipeline);
         });
 
-        pipeline->Add(NDqs::CreateDqsReplacePrecomputesTransformer(*pipeline->GetTypeAnnotationContext(), State_->FunctionRegistry), "ReplacePrecomputes");
         if (State_->Settings->UseBlockReader.Get().GetOrElse(false)) {
             pipeline->Add(NDqs::CreateDqsRewritePhyBlockReadOnDqIntegrationTransformer(*pipeline->GetTypeAnnotationContext()), "ReplaceWideReadsWithBlock");
         }
@@ -274,9 +275,7 @@ struct TDqsPipelineConfigurator : public IPipelineConfigurator {
 
     void AfterOptimize(TTransformationPipeline* pipeline) const final {
         std::for_each(UniqIntegrations_.cbegin(), UniqIntegrations_.cend(), [&](const auto dqInt) {
-            for (auto& stage: dqInt->GetPeepholeTransforms(false, ProviderParams_)) {
-                pipeline->Add(std::move(stage));
-            }
+            dqInt->ConfigurePeepholePipeline(false, ProviderParams_, pipeline);
         });
     }