[YQ-3761] Fix RewriteAsHoppingWindow optimization (#10455)

24 files changed, 110 insertions, 82 deletions

diff --git a/ydb/library/yql/dq/opt/dq_opt_log.cpp b/ydb/library/yql/dq/opt/dq_opt_log.cpp
index cf544dfa22..8f6e152e39 100644
--- a/ydb/library/yql/dq/opt/dq_opt_log.cpp
+++ b/ydb/library/yql/dq/opt/dq_opt_log.cpp
@@ -184,32 +184,18 @@ static void CollectSinkStages(const NNodes::TDqQuery& dqQuery, THashSet<TExprNod
 }
 
 NNodes::TExprBase DqMergeQueriesWithSinks(NNodes::TExprBase dqQueryNode, TExprContext& ctx) {
-    NNodes::TDqQuery dqQuery = dqQueryNode.Cast<NNodes::TDqQuery>();
-
-    THashSet<TExprNode::TPtr, TExprNode::TPtrHash> sinkStages;
-    CollectSinkStages(dqQuery, sinkStages);
-    TOptimizeExprSettings settings{nullptr};
-    settings.VisitLambdas = false;
-    bool deletedDqQueryChild = false;
-    TExprNode::TPtr newDqQueryNode;
-    auto status = OptimizeExpr(dqQueryNode.Ptr(), newDqQueryNode, [&sinkStages, &deletedDqQueryChild](const TExprNode::TPtr& node, TExprContext& ctx) -> TExprNode::TPtr {
-        for (ui32 childIndex = 0; childIndex < node->ChildrenSize(); ++childIndex) {
-            TExprNode* child = node->Child(childIndex);
-            if (child->IsCallable(NNodes::TDqQuery::CallableName())) {
-                NNodes::TDqQuery dqQueryChild(child);
-                CollectSinkStages(dqQueryChild, sinkStages);
-                deletedDqQueryChild = true;
-                return ctx.ChangeChild(*node, childIndex, dqQueryChild.World().Ptr());
-            }
-        }
-        return node;
-    }, ctx, settings);
-    YQL_ENSURE(status != IGraphTransformer::TStatus::Error, "Failed to merge DqQuery nodes: " << status);
-
-    if (deletedDqQueryChild) {
-        auto dqQueryBuilder = Build<TDqQuery>(ctx, dqQuery.Pos());
-        dqQueryBuilder.World(newDqQueryNode->ChildPtr(TDqQuery::idx_World));
-
+    auto maybeDqQuery = dqQueryNode.Maybe<NNodes::TDqQuery>();
+    YQL_ENSURE(maybeDqQuery, "Expected DqQuery!");
+    auto dqQuery = maybeDqQuery.Cast();
+
+    if (auto maybeDqQueryChild = dqQuery.World().Maybe<NNodes::TDqQuery>()) {
+        auto dqQueryChild = maybeDqQueryChild.Cast();
+        auto dqQueryBuilder = Build<TDqQuery>(ctx, dqQuery.Pos())
+            .World(dqQueryChild.World());
+
+        THashSet<TExprNode::TPtr, TExprNode::TPtrHash> sinkStages;
+        CollectSinkStages(dqQuery, sinkStages);
+        CollectSinkStages(maybeDqQueryChild.Cast(), sinkStages);
         auto sinkStagesBuilder = dqQueryBuilder.SinkStages();
         for (const TExprNode::TPtr& stage : sinkStages) {
             sinkStagesBuilder.Add(stage);
diff --git a/ydb/library/yql/providers/clickhouse/expr_nodes/yql_clickhouse_expr_nodes.json b/ydb/library/yql/providers/clickhouse/expr_nodes/yql_clickhouse_expr_nodes.json
index a3378a3981..358d2e112b 100644
--- a/ydb/library/yql/providers/clickhouse/expr_nodes/yql_clickhouse_expr_nodes.json
+++ b/ydb/library/yql/providers/clickhouse/expr_nodes/yql_clickhouse_expr_nodes.json
@@ -41,9 +41,10 @@
             "Base": "TCallable",
             "Match": {"Type": "Callable", "Name": "ClSourceSettings"},
             "Children": [
-                {"Index": 0, "Name": "Table", "Type": "TCoAtom"},
-                {"Index": 1, "Name": "Token", "Type": "TCoSecureParam"},
-                {"Index": 2, "Name": "Columns", "Type": "TCoAtomList"}
+                {"Index": 0, "Name": "World", "Type": "TExprBase"},
+                {"Index": 1, "Name": "Table", "Type": "TCoAtom"},
+                {"Index": 2, "Name": "Token", "Type": "TCoSecureParam"},
+                {"Index": 3, "Name": "Columns", "Type": "TCoAtomList"}
             ]
         }
     ]
diff --git a/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_datasource_type_ann.cpp b/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_datasource_type_ann.cpp
index 23e0e3d1b1..1c52e0764d 100644
--- a/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_datasource_type_ann.cpp
@@ -25,7 +25,11 @@ public:
     }
 
     TStatus HandleSourceSettings(const TExprNode::TPtr& input, TExprContext& ctx) {
-        if (!EnsureArgsCount(*input, 3U, ctx)) {
+        if (!EnsureArgsCount(*input, 4, ctx)) {
+            return TStatus::Error;
+        }
+
+        if (!EnsureWorldType(*input->Child(TClSourceSettings::idx_World), ctx)) {
             return TStatus::Error;
         }
 
diff --git a/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_dq_integration.cpp b/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_dq_integration.cpp
index 0f997f1e44..f9d370bdac 100644
--- a/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_dq_integration.cpp
+++ b/ydb/library/yql/providers/clickhouse/provider/yql_clickhouse_dq_integration.cpp
@@ -52,6 +52,7 @@ public:
 
             return Build<TDqSourceWrap>(ctx, read->Pos())
                 .Input<TClSourceSettings>()
+                    .World(clReadTable.World())
                     .Table(clReadTable.Table())
                     .Token<TCoSecureParam>()
                         .Name().Build(token)
diff --git a/ydb/library/yql/providers/generic/expr_nodes/yql_generic_expr_nodes.json b/ydb/library/yql/providers/generic/expr_nodes/yql_generic_expr_nodes.json
index 88e63d1583..90652bc303 100644
--- a/ydb/library/yql/providers/generic/expr_nodes/yql_generic_expr_nodes.json
+++ b/ydb/library/yql/providers/generic/expr_nodes/yql_generic_expr_nodes.json
@@ -41,11 +41,12 @@
             "Base": "TCallable",
             "Match": {"Type": "Callable", "Name": "GenSourceSettings"},
             "Children": [
-                {"Index": 0, "Name": "Cluster", "Type": "TCoAtom"},
-                {"Index": 1, "Name": "Table", "Type": "TCoAtom"},
-                {"Index": 2, "Name": "Token", "Type": "TCoSecureParam"},
-                {"Index": 3, "Name": "Columns", "Type": "TCoAtomList"},
-                {"Index": 4, "Name": "FilterPredicate", "Type": "TCoLambda"}
+                {"Index": 0, "Name": "World", "Type": "TExprBase"},
+                {"Index": 1, "Name": "Cluster", "Type": "TCoAtom"},
+                {"Index": 2, "Name": "Table", "Type": "TCoAtom"},
+                {"Index": 3, "Name": "Token", "Type": "TCoSecureParam"},
+                {"Index": 4, "Name": "Columns", "Type": "TCoAtomList"},
+                {"Index": 5, "Name": "FilterPredicate", "Type": "TCoLambda"}
             ]
         }
     ]
diff --git a/ydb/library/yql/providers/generic/provider/yql_generic_datasource_type_ann.cpp b/ydb/library/yql/providers/generic/provider/yql_generic_datasource_type_ann.cpp
index 791c9ecb7f..78d89b89d8 100644
--- a/ydb/library/yql/providers/generic/provider/yql_generic_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/generic/provider/yql_generic_datasource_type_ann.cpp
@@ -49,7 +49,11 @@ namespace NYql {
         }
 
         TStatus HandleSourceSettings(const TExprNode::TPtr& input, TExprContext& ctx) {
-            if (!EnsureArgsCount(*input, 5, ctx)) {
+            if (!EnsureArgsCount(*input, 6, ctx)) {
+                return TStatus::Error;
+            }
+
+            if (!EnsureWorldType(*input->Child(TGenSourceSettings::idx_World), ctx)) {
                 return TStatus::Error;
             }
 
diff --git a/ydb/library/yql/providers/generic/provider/yql_generic_dq_integration.cpp b/ydb/library/yql/providers/generic/provider/yql_generic_dq_integration.cpp
index 777cf1b6c7..88001bf7a9 100644
--- a/ydb/library/yql/providers/generic/provider/yql_generic_dq_integration.cpp
+++ b/ydb/library/yql/providers/generic/provider/yql_generic_dq_integration.cpp
@@ -87,6 +87,7 @@ namespace NYql {
                     // clang-format off
                     return Build<TDqSourceWrap>(ctx, read->Pos())
                         .Input<TGenSourceSettings>()
+                            .World(genReadTable.World())
                             .Cluster(genReadTable.DataSource().Cluster())
                             .Table(genReadTable.Table())
                             .Token<TCoSecureParam>()
diff --git a/ydb/library/yql/providers/pq/expr_nodes/yql_pq_expr_nodes.json b/ydb/library/yql/providers/pq/expr_nodes/yql_pq_expr_nodes.json
index 0b178695aa..d70cc37688 100644
--- a/ydb/library/yql/providers/pq/expr_nodes/yql_pq_expr_nodes.json
+++ b/ydb/library/yql/providers/pq/expr_nodes/yql_pq_expr_nodes.json
@@ -67,11 +67,12 @@
             "Base": "TCallable",
             "Match": {"Type": "Callable", "Name": "DqPqTopicSource"},
             "Children": [
-                {"Index": 0, "Name": "Topic", "Type": "TPqTopic"},
-                {"Index": 1, "Name": "Columns", "Type": "TExprBase"},
-                {"Index": 2, "Name": "Settings", "Type": "TCoNameValueTupleList"},
-                {"Index": 3, "Name": "Token", "Type": "TCoSecureParam"},
-                {"Index": 4, "Name": "FilterPredicate", "Type": "TCoLambda"}
+                {"Index": 0, "Name": "World", "Type": "TExprBase"},
+                {"Index": 1, "Name": "Topic", "Type": "TPqTopic"},
+                {"Index": 2, "Name": "Columns", "Type": "TExprBase"},
+                {"Index": 3, "Name": "Settings", "Type": "TCoNameValueTupleList"},
+                {"Index": 4, "Name": "Token", "Type": "TCoSecureParam"},
+                {"Index": 5, "Name": "FilterPredicate", "Type": "TCoLambda"}
             ]
         },
         {
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
index ad404949ed..2e2a458d72 100644
--- a/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
@@ -132,11 +132,16 @@ public:
     }
 
     TStatus HandleDqTopicSource(TExprBase input, TExprContext& ctx) {
-        if (!EnsureArgsCount(input.Ref(), 5, ctx)) {
+        if (!EnsureArgsCount(input.Ref(), 6, ctx)) {
             return TStatus::Error;
         }
 
         TDqPqTopicSource topicSource = input.Cast<TDqPqTopicSource>();
+
+        if (!EnsureWorldType(topicSource.World().Ref(), ctx)) {
+            return TStatus::Error;
+        }
+
         TPqTopic topic = topicSource.Topic();
 
         if (!EnsureCallable(topic.Ref(), ctx)) {
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp
index 3a305edaf5..530bda256d 100644
--- a/ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp
+++ b/ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp
@@ -147,6 +147,7 @@ public:
 
             return Build<TDqSourceWrap>(ctx, read->Pos())
                 .Input<TDqPqTopicSource>()
+                    .World(pqReadTopic.World())
                     .Topic(pqReadTopic.Topic())
                     .Columns(std::move(columnNames))
                     .Settings(BuildTopicReadSettings(clusterName, dqSettings, read->Pos(), format, ctx))
diff --git a/ydb/library/yql/providers/s3/expr_nodes/yql_s3_expr_nodes.json b/ydb/library/yql/providers/s3/expr_nodes/yql_s3_expr_nodes.json
index 631e07dd39..f7121ceaf6 100644
--- a/ydb/library/yql/providers/s3/expr_nodes/yql_s3_expr_nodes.json
+++ b/ydb/library/yql/providers/s3/expr_nodes/yql_s3_expr_nodes.json
@@ -45,10 +45,11 @@
             "Base": "TCallable",
             "Match": {"Type": "CallableBase"},
             "Children": [
-                {"Index": 0, "Name": "Paths", "Type": "TS3Paths"},
-                {"Index": 1, "Name": "Token", "Type": "TCoSecureParam"},
-                {"Index": 2, "Name": "RowsLimitHint", "Type": "TCoAtom"},
-                {"Index": 3, "Name": "Path", "Type": "TCoAtom"}
+                {"Index": 0, "Name": "World", "Type": "TExprBase"},
+                {"Index": 1, "Name": "Paths", "Type": "TS3Paths"},
+                {"Index": 2, "Name": "Token", "Type": "TCoSecureParam"},
+                {"Index": 3, "Name": "RowsLimitHint", "Type": "TCoAtom"},
+                {"Index": 4, "Name": "Path", "Type": "TCoAtom"}
             ]
         },
         {
@@ -56,9 +57,9 @@
             "Base": "TS3SourceSettingsBase",
             "Match": {"Type": "Callable", "Name": "S3SourceSettings"},
             "Children": [
-                {"Index": 4, "Name": "SizeLimit", "Type": "TExprBase", "Optional": true},
-                {"Index": 5, "Name": "PathPattern", "Type": "TExprBase", "Optional": true},
-                {"Index": 6, "Name": "PathPatternVariant", "Type": "TExprBase", "Optional": true}
+                {"Index": 5, "Name": "SizeLimit", "Type": "TExprBase", "Optional": true},
+                {"Index": 6, "Name": "PathPattern", "Type": "TExprBase", "Optional": true},
+                {"Index": 7, "Name": "PathPatternVariant", "Type": "TExprBase", "Optional": true}
             ]
         },
         {
@@ -66,10 +67,10 @@
             "Base": "TS3SourceSettingsBase",
             "Match": {"Type": "Callable", "Name": "S3ParseSettings"},
             "Children": [
-                {"Index": 4, "Name": "Format", "Type": "TCoAtom"},
-                {"Index": 5, "Name": "RowType", "Type": "TExprBase"},
-                {"Index": 6, "Name": "FilterPredicate", "Type": "TCoLambda"},
-                {"Index": 7, "Name": "Settings", "Type": "TExprBase", "Optional": true}
+                {"Index": 5, "Name": "Format", "Type": "TCoAtom"},
+                {"Index": 6, "Name": "RowType", "Type": "TExprBase"},
+                {"Index": 7, "Name": "FilterPredicate", "Type": "TCoLambda"},
+                {"Index": 8, "Name": "Settings", "Type": "TExprBase", "Optional": true}
             ]
         },
         {
diff --git a/ydb/library/yql/providers/s3/provider/yql_s3_datasource_type_ann.cpp b/ydb/library/yql/providers/s3/provider/yql_s3_datasource_type_ann.cpp
index 838d522b0b..246c2292c5 100644
--- a/ydb/library/yql/providers/s3/provider/yql_s3_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/s3/provider/yql_s3_datasource_type_ann.cpp
@@ -303,7 +303,11 @@ public:
     }
 
     TStatus HandleS3SourceSettings(const TExprNode::TPtr& input, TExprContext& ctx) {
-        if (!EnsureMinArgsCount(*input, 4U, ctx)) {
+        if (!EnsureMinMaxArgsCount(*input, 5, 8, ctx)) {
+            return TStatus::Error;
+        }
+
+        if (!EnsureWorldType(*input->Child(TS3SourceSettings::idx_World), ctx)) {
             return TStatus::Error;
         }
 
@@ -335,7 +339,11 @@ public:
     }
 
     TStatus HandleS3ParseSettings(const TExprNode::TPtr& input, TExprContext& ctx) {
-        if (!EnsureMinMaxArgsCount(*input, 7U, 8U, ctx)) {
+        if (!EnsureMinMaxArgsCount(*input, 8, 9, ctx)) {
+            return TStatus::Error;
+        }
+
+        if (!EnsureWorldType(*input->Child(TS3ParseSettings::idx_World), ctx)) {
             return TStatus::Error;
         }
 
diff --git a/ydb/library/yql/providers/s3/provider/yql_s3_dq_integration.cpp b/ydb/library/yql/providers/s3/provider/yql_s3_dq_integration.cpp
index 9cc5b024c6..1beda207ff 100644
--- a/ydb/library/yql/providers/s3/provider/yql_s3_dq_integration.cpp
+++ b/ydb/library/yql/providers/s3/provider/yql_s3_dq_integration.cpp
@@ -286,6 +286,7 @@ public:
             if (const auto useCoro = State_->Configuration->SourceCoroActor.Get(); (!useCoro || *useCoro) && format != "raw" && format != "json_list") {
                 return Build<TDqSourceWrap>(ctx, read->Pos())
                     .Input<TS3ParseSettings>()
+                        .World(s3ReadObject.World())
                         .Paths(s3ReadObject.Object().Paths())
                         .Token<TCoSecureParam>()
                             .Name().Build(token)
@@ -331,6 +332,7 @@ public:
                 auto emptyNode = Build<TCoVoid>(ctx, read->Pos()).Done().Ptr();
                 return Build<TDqSourceWrap>(ctx, read->Pos())
                     .Input<TS3SourceSettings>()
+                        .World(s3ReadObject.World())
                         .Paths(s3ReadObject.Object().Paths())
                         .Token<TCoSecureParam>()
                             .Name().Build(token)
diff --git a/ydb/library/yql/providers/solomon/expr_nodes/yql_solomon_expr_nodes.json b/ydb/library/yql/providers/solomon/expr_nodes/yql_solomon_expr_nodes.json
index a4db8e885d..74050a30cc 100644
--- a/ydb/library/yql/providers/solomon/expr_nodes/yql_solomon_expr_nodes.json
+++ b/ydb/library/yql/providers/solomon/expr_nodes/yql_solomon_expr_nodes.json
@@ -48,18 +48,19 @@
             "Base": "TCallable",
             "Match": {"Type": "Callable", "Name": "SoSourceSettings"},
             "Children": [
-                {"Index": 0, "Name": "Project", "Type": "TCoAtom"},
-                {"Index": 1, "Name": "Token", "Type": "TCoSecureParam"},
-                {"Index": 2, "Name": "RowType", "Type": "TExprBase"},
-                {"Index": 3, "Name": "SystemColumns", "Type": "TCoAtomList"},
-                {"Index": 4, "Name": "LabelNames", "Type": "TCoAtomList"},
-                {"Index": 5, "Name": "From", "Type": "TCoAtom"},
-                {"Index": 6, "Name": "To", "Type": "TCoAtom"},
-                {"Index": 7, "Name": "Program", "Type": "TCoAtom"},
-                {"Index": 8, "Name": "DownsamplingDisabled", "Type": "TCoBool"},
-                {"Index": 9, "Name": "DownsamplingAggregation", "Type": "TCoAtom"},
-                {"Index": 10, "Name": "DownsamplingFill", "Type": "TCoAtom"},
-                {"Index": 11, "Name": "DownsamplingGridSec", "Type": "TCoUint32"}
+                {"Index": 0, "Name": "World", "Type": "TExprBase"},
+                {"Index": 1, "Name": "Project", "Type": "TCoAtom"},
+                {"Index": 2, "Name": "Token", "Type": "TCoSecureParam"},
+                {"Index": 3, "Name": "RowType", "Type": "TExprBase"},
+                {"Index": 4, "Name": "SystemColumns", "Type": "TCoAtomList"},
+                {"Index": 5, "Name": "LabelNames", "Type": "TCoAtomList"},
+                {"Index": 6, "Name": "From", "Type": "TCoAtom"},
+                {"Index": 7, "Name": "To", "Type": "TCoAtom"},
+                {"Index": 8, "Name": "Program", "Type": "TCoAtom"},
+                {"Index": 9, "Name": "DownsamplingDisabled", "Type": "TCoBool"},
+                {"Index": 10, "Name": "DownsamplingAggregation", "Type": "TCoAtom"},
+                {"Index": 11, "Name": "DownsamplingFill", "Type": "TCoAtom"},
+                {"Index": 12, "Name": "DownsamplingGridSec", "Type": "TCoUint32"}
             ]
         },
         {
diff --git a/ydb/library/yql/providers/solomon/provider/yql_solomon_datasource_type_ann.cpp b/ydb/library/yql/providers/solomon/provider/yql_solomon_datasource_type_ann.cpp
index 72fe1df5ca..c09a053189 100644
--- a/ydb/library/yql/providers/solomon/provider/yql_solomon_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/solomon/provider/yql_solomon_datasource_type_ann.cpp
@@ -34,7 +34,11 @@ public:
     }
 
     TStatus HandleSoSourceSettings(const TExprNode::TPtr& input, TExprContext& ctx) {
-        if (!EnsureArgsCount(*input, 12U, ctx)) {
+        if (!EnsureArgsCount(*input, 13, ctx)) {
+            return TStatus::Error;
+        }
+
+        if (!EnsureWorldType(*input->Child(TSoSourceSettings::idx_World), ctx)) {
             return TStatus::Error;
         }
 
diff --git a/ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp b/ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp
index 7cf47fa189..af4a567e6d 100644
--- a/ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp
+++ b/ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp
@@ -200,6 +200,7 @@ public:
 
             return Build<TDqSourceWrap>(ctx, read->Pos())
                 .Input<TSoSourceSettings>()
+                    .World(soReadObject.World())
                     .Project(soReadObject.Object().Project())
                     .Token<TCoSecureParam>()
                         .Name().Build(token)
diff --git a/ydb/library/yql/providers/ydb/expr_nodes/yql_ydb_expr_nodes.json b/ydb/library/yql/providers/ydb/expr_nodes/yql_ydb_expr_nodes.json
index 534f097ec3..62557c87f9 100644
--- a/ydb/library/yql/providers/ydb/expr_nodes/yql_ydb_expr_nodes.json
+++ b/ydb/library/yql/providers/ydb/expr_nodes/yql_ydb_expr_nodes.json
@@ -30,9 +30,10 @@
             "Base": "TCallable",
             "Match": {"Type": "Callable", "Name": "YdbSourceSettings"},
             "Children": [
-                {"Index": 0, "Name": "Table", "Type": "TCoAtom"},
-                {"Index": 1, "Name": "Token", "Type": "TCoSecureParam"},
-                {"Index": 2, "Name": "Columns", "Type": "TCoAtomList"}
+                {"Index": 0, "Name": "World", "Type": "TExprBase"},
+                {"Index": 1, "Name": "Table", "Type": "TCoAtom"},
+                {"Index": 2, "Name": "Token", "Type": "TCoSecureParam"},
+                {"Index": 3, "Name": "Columns", "Type": "TCoAtomList"}
             ]
         },
         {
diff --git a/ydb/library/yql/providers/ydb/provider/yql_ydb_datasource_type_ann.cpp b/ydb/library/yql/providers/ydb/provider/yql_ydb_datasource_type_ann.cpp
index 466d4aa36c..e074f7e911 100644
--- a/ydb/library/yql/providers/ydb/provider/yql_ydb_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/ydb/provider/yql_ydb_datasource_type_ann.cpp
@@ -29,7 +29,11 @@ public:
 
     TStatus HandleYdbSourceSettings(const TExprNode::TPtr& input, TExprNode::TPtr& output, TExprContext& ctx) {
         Y_UNUSED(output);
-        if (!EnsureArgsCount(*input, 3U, ctx)) {
+        if (!EnsureArgsCount(*input, 4, ctx)) {
+            return TStatus::Error;
+        }
+
+        if (!EnsureWorldType(*input->Child(TYdbSourceSettings::idx_World), ctx)) {
             return TStatus::Error;
         }
 
diff --git a/ydb/library/yql/providers/ydb/provider/yql_ydb_dq_integration.cpp b/ydb/library/yql/providers/ydb/provider/yql_ydb_dq_integration.cpp
index 7ff7fc1c5c..9f165f0ba3 100644
--- a/ydb/library/yql/providers/ydb/provider/yql_ydb_dq_integration.cpp
+++ b/ydb/library/yql/providers/ydb/provider/yql_ydb_dq_integration.cpp
@@ -101,6 +101,7 @@ public:
 
             return Build<TDqSourceWrap>(ctx, read->Pos())
                 .Input<TYdbSourceSettings>()
+                    .World(ydbReadTable.World())
                     .Table(ydbReadTable.Table())
                     .Token<TCoSecureParam>()
                         .Name().Build(token)
diff --git a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Basic-default.txt_/opt.yql b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Basic-default.txt_/opt.yql
index 90de75a243..30c87e68a6 100644
--- a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Basic-default.txt_/opt.yql
+++ b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Basic-default.txt_/opt.yql
@@ -4,9 +4,9 @@
 (let $3 (DataType 'String))
 (let $4 (StructType '('"kind" $3) '('"labels" (DictType $3 $3)) '('"ts" (DataType 'Datetime)) '('type $3) '('"value" (DataType 'Double))))
 (let $5 '('"kind" '"labels" '"value" '"ts" 'type))
-(let $6 (SoSourceSettings '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"false") '"AVG" '"PREVIOUS" (Uint32 '"15")))
+(let $6 (SoSourceSettings world '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"false") '"AVG" '"PREVIOUS" (Uint32 '"15")))
 (let $7 (DqStage '((DqSource (DataSource '"solomon" '"local_solomon") $6)) (lambda '($10) $10) '('('"_logical_id" '199948))))
-(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($11) $11) '('('"_logical_id" '199969))))
+(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($11) $11) '('('"_logical_id" '199975))))
 (let $9 (ResPull! $1 $2 (Key) (DqCnResult (TDqOutput $8 '"0") '()) '('('type) '('autoref)) '"dq"))
 (return (Commit! (Commit! $9 $2) (DataSink '"solomon" '"local_solomon")))
 )
diff --git a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Downsampling-default.txt_/opt.yql b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Downsampling-default.txt_/opt.yql
index 7043ceed81..f88446f879 100644
--- a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Downsampling-default.txt_/opt.yql
+++ b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Downsampling-default.txt_/opt.yql
@@ -4,9 +4,9 @@
 (let $3 (DataType 'String))
 (let $4 (StructType '('"kind" $3) '('"labels" (DictType $3 $3)) '('"ts" (DataType 'Datetime)) '('type $3) '('"value" (DataType 'Double))))
 (let $5 '('"kind" '"labels" '"value" '"ts" 'type))
-(let $6 (SoSourceSettings '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"true") '"SUM" '"PREVIOUS" (Uint32 '"25")))
+(let $6 (SoSourceSettings world '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"true") '"SUM" '"PREVIOUS" (Uint32 '"25")))
 (let $7 (DqStage '((DqSource (DataSource '"solomon" '"local_solomon") $6)) (lambda '($10) $10) '('('"_logical_id" '199972))))
-(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($11) $11) '('('"_logical_id" '199993))))
+(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($11) $11) '('('"_logical_id" '199999))))
 (let $9 (ResPull! $1 $2 (Key) (DqCnResult (TDqOutput $8 '"0") '()) '('('type) '('autoref)) '"dq"))
 (return (Commit! (Commit! $9 $2) (DataSink '"solomon" '"local_solomon")))
 )
diff --git a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-DownsamplingValidSettings-default.txt_/opt.yql b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-DownsamplingValidSettings-default.txt_/opt.yql
index 1cbdb28578..37b96d2c4b 100644
--- a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-DownsamplingValidSettings-default.txt_/opt.yql
+++ b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-DownsamplingValidSettings-default.txt_/opt.yql
@@ -4,9 +4,9 @@
 (let $3 (DataType 'String))
 (let $4 (StructType '('"kind" $3) '('"labels" (DictType $3 $3)) '('"ts" (DataType 'Datetime)) '('type $3) '('"value" (DataType 'Double))))
 (let $5 '('"kind" '"labels" '"value" '"ts" 'type))
-(let $6 (SoSourceSettings '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"false") '"SUM" '"PREVIOUS" (Uint32 '"15")))
+(let $6 (SoSourceSettings world '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"false") '"SUM" '"PREVIOUS" (Uint32 '"15")))
 (let $7 (DqStage '((DqSource (DataSource '"solomon" '"local_solomon") $6)) (lambda '($10) $10) '('('"_logical_id" '199960))))
-(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($11) $11) '('('"_logical_id" '199981))))
+(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($11) $11) '('('"_logical_id" '199987))))
 (let $9 (ResPull! $1 $2 (Key) (DqCnResult (TDqOutput $8 '"0") '()) '('('type) '('autoref)) '"dq"))
 (return (Commit! (Commit! $9 $2) (DataSink '"solomon" '"local_solomon")))
 )
diff --git a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-LabelColumns-default.txt_/opt.yql b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-LabelColumns-default.txt_/opt.yql
index e700b9849f..3bdac9c0e0 100644
--- a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-LabelColumns-default.txt_/opt.yql
+++ b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-LabelColumns-default.txt_/opt.yql
@@ -8,9 +8,9 @@
 (let $7 (StructType '($3 $4) '($5 $4) '($6 $4) '('"kind" $4) '('"project" $4) '('"ts" (DataType 'Datetime)) '('type $4) '('"value" (DataType 'Double))))
 (let $8 '('"kind" '"value" '"ts" 'type))
 (let $9 '($3 $5 '"project" $6))
-(let $10 (SoSourceSettings '"my_project" (SecureParam '"cluster:default_local_solomon") $7 $8 $9 '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"false") '"AVG" '"PREVIOUS" (Uint32 '"15")))
+(let $10 (SoSourceSettings world '"my_project" (SecureParam '"cluster:default_local_solomon") $7 $8 $9 '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"false") '"AVG" '"PREVIOUS" (Uint32 '"15")))
 (let $11 (DqStage '((DqSource (DataSource '"solomon" '"local_solomon") $10)) (lambda '($14) $14) '('('"_logical_id" '199986))))
-(let $12 (DqStage '((DqCnUnionAll (TDqOutput $11 '"0"))) (lambda '($15) $15) '('('"_logical_id" '200007))))
+(let $12 (DqStage '((DqCnUnionAll (TDqOutput $11 '"0"))) (lambda '($15) $15) '('('"_logical_id" '200013))))
 (let $13 (ResPull! $1 $2 (Key) (DqCnResult (TDqOutput $12 '"0") '()) '('('type) '('autoref)) '"dq"))
 (return (Commit! (Commit! $13 $2) (DataSink '"solomon" '"local_solomon")))
 )
diff --git a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Subquery-default.txt_/opt.yql b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Subquery-default.txt_/opt.yql
index 3175699f34..e9c55c6442 100644
--- a/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Subquery-default.txt_/opt.yql
+++ b/ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-Subquery-default.txt_/opt.yql
@@ -4,9 +4,9 @@
 (let $3 (DataType 'String))
 (let $4 (StructType '('"kind" $3) '('"labels" (DictType $3 $3)) '('"ts" (DataType 'Datetime)) '('type $3) '('"value" (DataType 'Double))))
 (let $5 '('"kind" '"labels" '"value" '"ts" 'type))
-(let $6 (SoSourceSettings '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"true") '"SUM" '"PREVIOUS" (Uint32 '"25")))
+(let $6 (SoSourceSettings world '"my_project" (SecureParam '"cluster:default_local_solomon") $4 $5 '() '"2023-12-08T14:40:39Z" '"2023-12-08T14:45:39Z" '"{}" (Bool '"true") '"SUM" '"PREVIOUS" (Uint32 '"25")))
 (let $7 (DqStage '((DqSource (DataSource '"solomon" '"local_solomon") $6)) (lambda '($11) $11) '('('"_logical_id" '200295))))
-(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($12) $12) '('('"_logical_id" '200331))))
+(let $8 (DqStage '((DqCnUnionAll (TDqOutput $7 '"0"))) (lambda '($12) $12) '('('"_logical_id" '200337))))
 (let $9 '('('type) '('autoref) '('unordered)))
 (let $10 (ResPull! $1 $2 (Key) (DqCnResult (TDqOutput $8 '"0") '()) $9 '"dq"))
 (return (Commit! (Commit! $10 $2) (DataSink '"solomon" '"local_solomon")))