YQL: Add option to get yt annotaions from user attributes (#7209)

Author: mxkovalev

ydb/library/yql/providers/yt/gateway/native/yql_yt_spec.cpp

@@ -187,8 +187,31 @@ void FillSpec(NYT::TNode& spec,
         }
     }
 
+    NYT::TNode annotations;
     if (auto val = settings->Annotations.Get(cluster)) {
-        spec["annotations"] = *val;
+        annotations = NYT::TNode::CreateMap(val.Get()->AsMap());
+    } else {
+        annotations = NYT::TNode::CreateMap();
+    }
+
+    // merge annotations from attributes
+    if (auto attrs = execCtx.Session_->OperationOptions_.AttrsYson.GetOrElse(TString())) {
+        NYT::TNode node = NYT::NodeFromYsonString(attrs);
+        if (auto attrAnnotations = node.AsMap().FindPtr("yt_annotations")) {
+            if (!attrAnnotations->IsMap()) {
+                throw yexception() << "Operation attribute \"yt_annotations\" should be a map";
+            }
+            for (const auto& [k, v] : attrAnnotations->AsMap()) {
+                auto it = annotations.AsMap().find(k);
+                if (it == annotations.AsMap().end()) {
+                    annotations[k] = v;
+                }
+            }
+        }
+    }
+
+    if (!annotations.Empty()) {
+        spec["annotations"] = std::move(annotations);
     }
 
     if (auto val = settings->StartedBy.Get(cluster)) {

ydb/library/yql/tools/dqrun/dqrun.cpp

@@ -507,6 +507,7 @@ int RunMain(int argc, const char* argv[])
     TString opId;
     IQStoragePtr qStorage;
     TQContext qContext;
+    TString ysonAttrs;
 
     NLastGetopt::TOpts opts = NLastGetopt::TOpts::Default();
     opts.AddLongOption('p', "program", "Program to execute (use '-' to read from stdin)")
@@ -670,6 +671,7 @@ int RunMain(int argc, const char* argv[])
         .Optional()
         .RequiredArgument("ENDPOINT")
         .StoreResult(&tokenAccessorEndpoint);
+    opts.AddLongOption("yson-attrs", "Provide operation yson attribues").StoreResult(&ysonAttrs);
     opts.AddHelpOption('h');
 
     opts.SetFreeArgsNum(0);
@@ -1069,6 +1071,10 @@ int RunMain(int argc, const char* argv[])
         runOptions.LineageStream = &Cout;
     }
 
+    if (ysonAttrs) {
+        program->SetOperationAttrsYson(ysonAttrs);
+    }
+
     int result = RunProgram(std::move(program), runOptions, clusters, sqlFlags);
     if (res.Has("metrics")) {
         NProto::TMetricsRegistrySnapshot snapshot;