Move DqReplicate check from type_ann to optimizers (#14426)

Co-authored-by: Roman Udovichenko <udovichenko-r@localhost.localdomain>

Author: rvu1024

ydb/library/yql/providers/dq/opt/physical_optimize.cpp

@@ -20,12 +20,16 @@ using namespace NYql::NNodes;
 class TDqsPhysicalOptProposalTransformer : public TOptimizeTransformerBase {
 public:
     TDqsPhysicalOptProposalTransformer(TTypeAnnotationContext* typeCtx, const TDqConfiguration::TPtr& config)
-        : TOptimizeTransformerBase(typeCtx, NLog::EComponent::ProviderDq, {})
+        : TOptimizeTransformerBase(/* TODO: typeCtx*/nullptr, NLog::EComponent::ProviderDq, {})
         , Config(config)
     {
         const bool enablePrecompute = Config->_EnablePrecompute.Get().GetOrElse(false);
+        const bool enableDqReplicate = Config->IsDqReplicateEnabled(*typeCtx);
 
 #define HNDL(name) "DqsPhy-"#name, Hndl(&TDqsPhysicalOptProposalTransformer::name)
+        if (!enableDqReplicate) {
+            AddHandler(0, &TDqReplicate::Match, HNDL(FailOnDqReplicate));
+        }
         AddHandler(0, &TDqSourceWrap::Match, HNDL(BuildStageWithSourceWrap));
         AddHandler(0, &TDqReadWrap::Match, HNDL(BuildStageWithReadWrap));
         AddHandler(0, &TCoSkipNullMembers::Match, HNDL(PushSkipNullMembersToStage<false>));
@@ -104,6 +108,11 @@ class TDqsPhysicalOptProposalTransformer : public TOptimizeTransformerBase {
     }
 
 protected:
+    TMaybeNode<TExprBase> FailOnDqReplicate(TExprBase node, TExprContext& ctx) {
+        ctx.AddError(YqlIssue(ctx.GetPosition(node.Pos()), TIssuesIds::DQ_OPTIMIZE_ERROR, "Reading multiple times from the same source is not supported"));
+        return {};
+    }
+
     TMaybeNode<TExprBase> BuildStageWithSourceWrap(TExprBase node, TExprContext& ctx) {
         return DqBuildStageWithSourceWrap(node, ctx);
     }

ydb/library/yql/providers/dq/provider/yql_dq_datasink.cpp

@@ -32,12 +32,9 @@ class TDqDataProviderSink: public TDataProviderBase {
     TDqDataProviderSink(const TDqState::TPtr& state)
         : State(state)
         , LogOptTransformer([state] () { return CreateDqsLogOptTransformer(state->TypeCtx, state->Settings); })
-        , PhyOptTransformer([state] () { return CreateDqsPhyOptTransformer(/*TODO*/nullptr, state->Settings); })
+        , PhyOptTransformer([state] () { return CreateDqsPhyOptTransformer(state->TypeCtx, state->Settings); })
         , PhysicalFinalizingTransformer([] () { return CreateDqsFinalizingOptTransformer(); })
-        , TypeAnnotationTransformer([state] () {
-            return CreateDqsDataSinkTypeAnnotationTransformer(
-                state->TypeCtx, state->Settings->IsDqReplicateEnabled(*state->TypeCtx));
-        })
+        , TypeAnnotationTransformer([state] () { return CreateDqsDataSinkTypeAnnotationTransformer(state->TypeCtx); })
         , ConstraintsTransformer([] () { return CreateDqDataSinkConstraintTransformer(); })
         , RecaptureTransformer([state] () { return CreateDqsRecaptureTransformer(state); })
     { }

ydb/library/yql/providers/dq/provider/yql_dq_datasink_type_ann.cpp

@@ -14,7 +14,7 @@ namespace {
 
 class TDqsDataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
 public:
-    TDqsDataSinkTypeAnnotationTransformer(TTypeAnnotationContext* typeCtx, bool enableDqReplicate)
+    TDqsDataSinkTypeAnnotationTransformer(TTypeAnnotationContext* typeCtx)
         : TVisitorTransformerBase(true), TypeCtx(typeCtx)
     {
         AddHandler({TDqStage::CallableName()}, Hndl(&NDq::AnnotateDqStage));
@@ -28,11 +28,7 @@ class TDqsDataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
         AddHandler({TDqCnBroadcast::CallableName()}, Hndl(&NDq::AnnotateDqConnection));
         AddHandler({TDqCnValue::CallableName()}, Hndl(&NDq::AnnotateDqCnValue));
         AddHandler({TDqCnMerge::CallableName()}, Hndl(&NDq::AnnotateDqCnMerge));
-        if (enableDqReplicate) {
-            AddHandler({TDqReplicate::CallableName()}, Hndl(&NDq::AnnotateDqReplicate));
-        } else {
-            AddHandler({TDqReplicate::CallableName()}, Hndl(&TDqsDataSinkTypeAnnotationTransformer::AnnotateDqReplicateAlwaysError));
-        }
+        AddHandler({TDqReplicate::CallableName()}, Hndl(&NDq::AnnotateDqReplicate));
         AddHandler({TDqJoin::CallableName()}, Hndl(&NDq::AnnotateDqJoin));
         AddHandler({TDqPhyGraceJoin::CallableName()}, Hndl(&NDq::AnnotateDqMapOrDictJoin));
         AddHandler({TDqPhyMapJoin::CallableName()}, Hndl(&NDq::AnnotateDqMapOrDictJoin));
@@ -47,11 +43,6 @@ class TDqsDataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
     }
 
 private:
-    TStatus AnnotateDqReplicateAlwaysError(const TExprNode::TPtr& input, TExprContext& ctx) {
-        ctx.AddError(YqlIssue(ctx.GetPosition(input->Pos()), TIssuesIds::DQ_OPTIMIZE_ERROR, "Reading multiple times from the same source is not supported"));
-        return TStatus::Error;
-    }
-
     TStatus AnnotateDqWrite(const TExprNode::TPtr& input, TExprNode::TPtr& output, TExprContext& ctx) {
         if (!EnsureMinArgsCount(*input, 2, ctx)) {
             return TStatus::Error;
@@ -86,8 +77,8 @@ class TDqsDataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
 
 } // unnamed
 
-THolder<TVisitorTransformerBase> CreateDqsDataSinkTypeAnnotationTransformer(TTypeAnnotationContext* typeCtx, bool enableDqReplicate) {
-    return THolder(new TDqsDataSinkTypeAnnotationTransformer(typeCtx, enableDqReplicate));
+THolder<TVisitorTransformerBase> CreateDqsDataSinkTypeAnnotationTransformer(TTypeAnnotationContext* typeCtx) {
+    return THolder(new TDqsDataSinkTypeAnnotationTransformer(typeCtx));
 }
 
 } // NYql

ydb/library/yql/providers/dq/provider/yql_dq_datasink_type_ann.h

@@ -7,6 +7,6 @@
 
 namespace NYql {
 
-THolder<TVisitorTransformerBase> CreateDqsDataSinkTypeAnnotationTransformer(TTypeAnnotationContext* typeCtx, bool enableDqReplicate);
+THolder<TVisitorTransformerBase> CreateDqsDataSinkTypeAnnotationTransformer(TTypeAnnotationContext* typeCtx);
 
 } // NYql