DQ Fallback when udf with layers (#26687)

Author: MrLolthe1st

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

@@ -818,7 +818,7 @@ class TDqExecTransformer: public TExecTransformerBase, TCounters
         bool fallbackFlag = BuildUploadList(uploadList, localRun, explorer, typeEnv, files);
 
         if (fallbackFlag) {
-            YQL_CLOG(DEBUG, ProviderDq) << "Fallback: " << NCommon::ExprToPrettyString(ctx, *input);
+            YQL_CLOG(TRACE, ProviderDq) << "Fallback: " << NCommon::ExprToPrettyString(ctx, *input);
             return Fallback();
         } else {
             *lambda = SerializeRuntimeNode(root, typeEnv);
@@ -921,6 +921,33 @@ class TDqExecTransformer: public TExecTransformerBase, TCounters
                 }
             }
 
+            bool hasErrors = false;
+            auto fallback = [&ctx, &hasErrors](const TExprNode& n, const TString& msg) {
+                auto issues = TIssues{TIssue(ctx.GetPosition(n.Pos()), msg).SetCode(TIssuesIds::DQ_GATEWAY_NEED_FALLBACK_ERROR, TSeverityIds::S_WARNING)};
+                ctx.AssociativeIssues.emplace(&n, std::move(issues));
+                hasErrors = true;
+            };
+
+            VisitExpr(*input.Get(), [&fallback, &hasErrors](const TExprNode& n) {
+                if (TCoScriptUdf::Match(&n) && NKikimr::NMiniKQL::IsSystemPython(NKikimr::NMiniKQL::ScriptTypeFromStr(n.Head().Content()))) {
+                    fallback(n, TStringBuilder() << "Cannot execute system python udf " << n.Content() << " in DQ");
+                    return false;
+                }
+                if ((TCoScriptUdf::Match(&n) && n.ChildrenSize() > 4) || (TCoUdf::Match(&n) && n.ChildrenSize() == 8)) {
+                    for (const auto& setting: n.Child(TCoScriptUdf::Match(&n) ? 4 : 7)->Children()) {
+                        YQL_ENSURE(setting->Head().IsAtom());
+                        if (setting->Head().Content() == "layers") {
+                            fallback(n, TStringBuilder() << "Cannot execute udf " << n.Head().Content() << " with layers in DQ");
+                            return false;
+                        }
+                    }
+                }
+                return !hasErrors;
+            });
+            if (hasErrors) {
+                return Fallback();
+            }
+
             THashMap<TString, TString> secureParams;
             NCommon::FillSecureParams(resInput, *State->TypeCtx, secureParams);
 
@@ -1074,7 +1101,7 @@ class TDqExecTransformer: public TExecTransformerBase, TCounters
                         return IGraphTransformer::TStatus(IGraphTransformer::TStatus::Error);
                     }
 
-                    YQL_CLOG(DEBUG, ProviderDq) << "Fallback from gateway: " << NCommon::ExprToPrettyString(ctx, *input);
+                    YQL_CLOG(TRACE, ProviderDq) << "Fallback from gateway: " << NCommon::ExprToPrettyString(ctx, *input);
                     TIssue warning(ctx.GetPosition(input->Pos()), "DQ cannot execute the query");
                     warning.Severity = TSeverityIds::S_INFO;
                     ctx.IssueManager.RaiseIssue(warning);

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

@@ -202,8 +202,17 @@ class TDqsRecaptureTransformer : public TSyncTransformerBase {
                     Scan(*node.Child(i), ctx, good, visited);
                 }
             }
-        }
-        else if (TCoScriptUdf::Match(&node)) {
+        } else if (TCoScriptUdf::Match(&node)) {
+            if (node.ChildrenSize() > 4) {
+                for (const auto& setting: node.Child(4)->Children()) {
+                    YQL_ENSURE(setting->Head().IsAtom());
+                    if (setting->Head().Content() == "layers") {
+                        AddInfo(ctx, TStringBuilder() << "Cannot execute udf " << node.Head().Content() << " with layers in DQ");
+                        good = false;
+                    }
+                }
+            }
+
             if (good && TCoScriptUdf::Match(&node) && NKikimr::NMiniKQL::IsSystemPython(NKikimr::NMiniKQL::ScriptTypeFromStr(node.Head().Content()))) {
                 AddInfo(ctx, TStringBuilder() << "system python udf");
                 good = false;
@@ -213,8 +222,15 @@ class TDqsRecaptureTransformer : public TSyncTransformerBase {
                     Scan(*node.Child(i), ctx, good, visited);
                 }
             }
-        }
-        else {
+        } else if (TCoUdf::Match(&node) && node.ChildrenSize() == 8) {
+            for (const auto& setting: node.Child(7)->Children()) {
+                YQL_ENSURE(setting->Head().IsAtom());
+                if (setting->Head().Content() == "layers") {
+                    AddInfo(ctx, TStringBuilder() << "Cannot execute udf " << node.Head().Content() << " with layers in DQ");
+                    good = false;
+                }
+            }
+        } else {
             for (size_t i = 0; i != node.ChildrenSize() && good; ++i) {
                 Scan(*node.Child(i), ctx, good, visited);
             }

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

@@ -58,6 +58,16 @@ std::shared_ptr<TIssue> TDqExecutionValidator::ValidateDqStage(const TExprNode&
                     stageInfo.Issue = MakeErrorPtr(ctx, *n, TStringBuilder() << "Cannot execute system python udf " << n->Content() << " in DQ");
                     hasErrors = true;
                 }
+                if ((TCoScriptUdf::Match(n.Get()) && n->ChildrenSize() > 4) || (TCoUdf::Match(n.Get()) && n->ChildrenSize() == 8)) {
+                    for (const auto& setting: n->Child(TCoScriptUdf::Match(n.Get()) ? 4 : 7)->Children()) {
+                        YQL_ENSURE(setting->Head().IsAtom());
+                        if (setting->Head().Content() == "layers") {
+                            stageInfo.Issue = MakeErrorPtr(ctx, *n, TStringBuilder() << "Cannot execute udf " << n->Head().Content() << " with layers in DQ");
+                            hasErrors = true;
+                        }
+                    }
+                }
+
                 if (!typeCtx->ForceDq && TDqReadWrapBase::Match(n.Get())) {
                     auto readNode = n->Child(0);
                     auto dataSourceName = readNode->Child(1)->Child(0)->Content();