Parse and pass streamlookup parameters, YQL part

commit_hash:1ad24832c82e465562c48bea1189a0187f691cd4

(cherry picked from commit 338a2a1)

Author: yumkam7

yql/essentials/core/type_ann/type_ann_join.cpp

@@ -475,7 +475,7 @@ namespace NTypeAnnImpl {
 
     IGraphTransformer::TStatus MapJoinCoreWrapper(const TExprNode::TPtr& input, TExprNode::TPtr& output, TContext& ctx) {
         Y_UNUSED(output);
-        
+
         if (!EnsureArgsCount(*input, 9, ctx.Expr)) {
             return IGraphTransformer::TStatus::Error;
         }

yql/essentials/core/yql_join.cpp

@@ -319,8 +319,16 @@ namespace {
                 }
             }
             else if (option.IsAtom("forceSortedMerge") || option.IsAtom("forceStreamLookup")) {
-                if (!EnsureTupleSize(*child, 1, ctx)) {
-                    return IGraphTransformer::TStatus::Error;
+                if (option.IsAtom("forceStreamLookup")) {
+                    if (child->ChildrenSize() % 2 == 0) {
+                        ctx.AddError(TIssue(ctx.GetPosition(option.Pos()), TStringBuilder() <<
+                                    "streamlookup() expects KEY VALUE... pairs"));
+                        return IGraphTransformer::TStatus::Error;
+                    }
+                } else {
+                    if (!EnsureTupleSize(*child, 1, ctx)) {
+                        return IGraphTransformer::TStatus::Error;
+                    }
                 }
                 if (hasJoinStrategyHint) {
                     ctx.AddError(TIssue(ctx.GetPosition(option.Pos()), TStringBuilder() <<
@@ -1351,9 +1359,14 @@ TEquiJoinLinkSettings GetEquiJoinLinkSettings(const TExprNode& linkSettings) {
     }
 
     result.ForceSortedMerge = HasSetting(linkSettings, "forceSortedMerge");
-    
-    if (HasSetting(linkSettings, "forceStreamLookup")) {
+
+    if (auto streamlookup = GetSetting(linkSettings, "forceStreamLookup")) {
+        YQL_ENSURE(result.JoinAlgoOptions.empty());
         result.JoinAlgo = EJoinAlgoType::StreamLookupJoin;
+        auto size = streamlookup->ChildrenSize();
+        for (decltype(size) i = 1; i < size; ++i) {
+            result.JoinAlgoOptions.push_back(TString(streamlookup->Child(i)->Content()));
+        }
     }
 
     if (HasSetting(linkSettings, "compact")) {

yql/essentials/core/yql_join.h

@@ -148,6 +148,7 @@ struct TEquiJoinLinkSettings {
     // JOIN implementation may ignore this flags if SortedMerge strategy is not supported
     bool ForceSortedMerge = false;
     bool Compact = false;
+    TVector<TString> JoinAlgoOptions;
 };
 
 TEquiJoinLinkSettings GetEquiJoinLinkSettings(const TExprNode& linkSettings);

yql/essentials/sql/v1/join.cpp

@@ -502,7 +502,11 @@ class TEquiJoin: public TJoinBase {
             if (TJoinLinkSettings::EStrategy::SortedMerge == descr.LinkSettings.Strategy) {
                 linkOptions = L(linkOptions, Q(Y(Q("forceSortedMerge"))));
             } else if (TJoinLinkSettings::EStrategy::StreamLookup == descr.LinkSettings.Strategy) {
-                linkOptions = L(linkOptions, Q(Y(Q("forceStreamLookup"))));
+                auto streamlookup = Y(Q("forceStreamLookup"));
+                for (auto&& option: descr.LinkSettings.Values) {
+                    streamlookup = L(streamlookup, Q(option));
+                }
+                linkOptions = L(linkOptions, Q(streamlookup));
             } else if (TJoinLinkSettings::EStrategy::ForceMap == descr.LinkSettings.Strategy) {
                 linkOptions = L(linkOptions, Q(Y(Q("join_algo"), Q("MapJoin"))));
             } else if (TJoinLinkSettings::EStrategy::ForceGrace == descr.LinkSettings.Strategy) {

yql/essentials/sql/v1/source.h

@@ -172,6 +172,7 @@ namespace NSQLTranslationV1 {
             ForceGrace
         };
         EStrategy Strategy = EStrategy::Default;
+        TVector<TString> Values;
         bool Compact = false;
     };
 

yql/essentials/sql/v1/sql_select.cpp

@@ -43,6 +43,7 @@ bool CollectJoinLinkSettings(TPosition pos, TJoinLinkSettings& linkSettings, TCo
 
         if (TJoinLinkSettings::EStrategy::Default == linkSettings.Strategy) {
             linkSettings.Strategy = newStrategy;
+            linkSettings.Values = hint.Values;
         } else if (newStrategy == linkSettings.Strategy) {
             ctx.Error() << "Duplicate join strategy hint";
             return false;