[RFC] Parse and pass streamlookup parameters

ydb/library/yql/core/expr_nodes/yql_expr_nodes.json

@@ -1626,7 +1626,7 @@
                 {"Index": 6, "Name": "RightRenames", "Type": "TCoAtomList"},
                 {"Index": 7, "Name": "LeftKeysColumnNames", "Type": "TCoAtomList"},
                 {"Index": 8, "Name": "RightKeysColumnNames", "Type": "TCoAtomList"},
-                {"Index": 9, "Name": "Flags", "Type": "TCoAtomList"}
+                {"Index": 9, "Name": "Flags", "Type": "TCoNameValueTupleList"}
             ]
         },
         {
@@ -1642,7 +1642,7 @@
                 {"Index": 5, "Name": "RightRenames", "Type": "TCoAtomList"},
                 {"Index": 6, "Name": "LeftKeysColumnNames", "Type": "TCoAtomList"},
                 {"Index": 7, "Name": "RightKeysColumnNames", "Type": "TCoAtomList"},
-                {"Index": 8, "Name": "Flags", "Type": "TCoAtomList"}
+                {"Index": 8, "Name": "Flags", "Type": "TCoNameValueTupleList"}
             ]
         },
         {

ydb/library/yql/core/type_ann/type_ann_join.cpp

@@ -617,7 +617,8 @@ namespace NTypeAnnImpl {
         }
 
         for (auto i = 0U; i < input->Tail().ChildrenSize(); ++i) {
-            if (const auto& flag = *input->Tail().Child(i); !flag.IsAtom({"LeftAny", "RightAny"})) {
+            const auto& flag = *input->Tail().Child(i)->Child(0);
+            if (!flag.IsAtom({"LeftAny","RightAny"})) {
                 ctx.Expr.AddError(TIssue(ctx.Expr.GetPosition(flag.Pos()), TStringBuilder() << "Unsupported grace join option: " << flag.Content()));
                 return IGraphTransformer::TStatus::Error;
             }

ydb/library/yql/core/ut/yql_expr_constraint_ut.cpp

@@ -1939,7 +1939,7 @@ Y_UNIT_TEST_SUITE(TYqlExprConstraints) {
     (let flow1 (ExpandMap (ToFlow list1) (lambda '(item) (Member item 'key1) (Member item 'subkey1) (Member item 'value1))))
     (let flow2 (ExpandMap (ToFlow list2) (lambda '(item) (Member item 'key2) (Member item 'subkey2) (Member item 'value2))))
 
-    (let join (GraceJoinCore flow1 flow2 'Inner '('0 '1) '('0 '1) '('0 '0 '1 '1 '2 '2) '('0 '3 '1 '4 '2 '5) '() '() '('LeftAny 'RightAny)))
+    (let join (GraceJoinCore flow1 flow2 'Inner '('0 '1) '('0 '1) '('0 '0 '1 '1 '2 '2) '('0 '3 '1 '4 '2 '5) '() '() '('('"LeftAny") '('"RightAny"))))
     (let list (Collect (NarrowMap join (lambda '(lk ls lv rk rs rv) (AsStruct '('lk lk) '('ls ls) '('lv lv) '('rk rk) '('rs rs) '('rv rv))))))
 
     (let res_sink (DataSink 'yt (quote plato)))

ydb/library/yql/core/yql_expr_constraint.cpp

@@ -2266,12 +2266,13 @@ class TCallableConstraintTransformer : public TCallableTransformerBase<TCallable
         }
 
         bool leftAny = false, rigthAny = false;
-        core.Flags().Ref().ForEachChild([&](const TExprNode& flag) {
-            if (flag.IsAtom("LeftAny"))
+        for (const auto &flag: core.Flags()) {
+            auto&& name = flag.Name().Value();
+            if (name == "LeftAny"sv)
                leftAny = true;
-            else if (flag.IsAtom("RightAny"))
+            else if (name == "RightAny"sv)
                 rigthAny = true;
-        });
+        }
 
         const TUniqueConstraintNode* lUnique = leftInput->GetConstraint<TUniqueConstraintNode>();
         const TUniqueConstraintNode* rUnique = rightInput->GetConstraint<TUniqueConstraintNode>();

ydb/library/yql/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() <<
+                                    "Odd number of options, streamlookup() supports only 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() <<
@@ -1352,8 +1360,12 @@ TEquiJoinLinkSettings GetEquiJoinLinkSettings(const TExprNode& linkSettings) {
 
     result.ForceSortedMerge = HasSetting(linkSettings, "forceSortedMerge");
 
-    if (HasSetting(linkSettings, "forceStreamLookup")) {
+    if(auto streamlookup = GetSetting(linkSettings, "forceStreamLookup")) {
         result.JoinAlgo = EJoinAlgoType::StreamLookupJoin;
+        auto size = streamlookup->ChildrenSize();
+        for (decltype(size) i = 1; i < size; ++i) {
+            result.Options.push_back(TString(streamlookup->Child(i)->Content()));
+        }
     }
 
     if (HasSetting(linkSettings, "compact")) {

ydb/library/yql/core/yql_join.h

@@ -147,6 +147,7 @@ struct TEquiJoinLinkSettings {
     EJoinAlgoType JoinAlgo = EJoinAlgoType::Undefined;
     // JOIN implementation may ignore this flags if SortedMerge strategy is not supported
     bool ForceSortedMerge = false;
+    TVector<TString> Options;
     bool Compact = false;
 };
 

ydb/library/yql/dq/expr_nodes/dq_expr_nodes.json

@@ -44,7 +44,7 @@
         "Match": {"Type": "Callable", "Name": "DqJoin"},
         "Children": [
             {"Index": 8, "Name": "JoinAlgo", "Type": "TCoAtom"},
-            {"Index": 9, "Name": "Flags", "Type": "TCoAtomList", "Optional": true}
+            {"Index": 9, "Name": "Flags", "Type": "TCoNameValueTupleList", "Optional": true}
         ]
     },
     {

ydb/library/yql/dq/opt/dq_opt_join.cpp

@@ -155,6 +155,13 @@ TMaybe<TJoinInputDesc> BuildDqJoin(const TCoEquiJoinTuple& joinTuple,
     }
 
     TStringBuf joinType = joinTuple.Type().Value();
+
+    if (linkSettings.JoinAlgo == EJoinAlgoType::StreamLookupJoin) {
+        YQL_ENSURE(joinType == TStringBuf("Left"), "Streamlookup supports only LEFT JOIN ... ANY");
+        YQL_ENSURE(!leftAny, "Streamlookup ANY LEFT join is not implemented");
+        YQL_ENSURE(rightAny, "Streamlookup supports only LEFT JOIN ... ANY");
+    }
+
     TSet<std::pair<TStringBuf, TStringBuf>> resultKeys;
     if (joinType != TStringBuf("RightOnly") && joinType != TStringBuf("RightSemi")) {
         resultKeys.insert(left->Keys.begin(), left->Keys.end());
@@ -219,7 +226,7 @@ TMaybe<TJoinInputDesc> BuildDqJoin(const TCoEquiJoinTuple& joinTuple,
         rightJoinKeyNames.emplace_back(rightColumnName);
     }
 
-    if (EHashJoinMode::Off == mode || EHashJoinMode::Map == mode || !(leftAny || rightAny)) {
+    if (EHashJoinMode::Off == mode || EHashJoinMode::Map == mode || !(leftAny || rightAny || !linkSettings.Options.empty())) {
         auto dqJoin = Build<TDqJoin>(ctx, joinTuple.Pos())
             .LeftInput(BuildDqJoinInput(ctx, joinTuple.Pos(), left->Input, leftJoinKeys, leftAny))
             .LeftLabel(leftTableLabel)
@@ -237,11 +244,26 @@ TMaybe<TJoinInputDesc> BuildDqJoin(const TCoEquiJoinTuple& joinTuple,
             .Done();
         return TJoinInputDesc(Nothing(), dqJoin, std::move(resultKeys));
     } else {
-        TExprNode::TListType flags;
-        if (leftAny)
-            flags.emplace_back(ctx.NewAtom(joinTuple.Pos(), "LeftAny", TNodeFlags::Default));
-        if (rightAny)
-            flags.emplace_back(ctx.NewAtom(joinTuple.Pos(), "RightAny", TNodeFlags::Default));
+        TVector<TCoNameValueTuple> flags;
+
+        if (leftAny) {
+            flags.push_back(
+                    Build<TCoNameValueTuple>(ctx, joinTuple.Pos())
+                        .Name().Build("LeftAny"sv)
+                        .Done());
+        }
+        if (rightAny) {
+            flags.push_back(
+                    Build<TCoNameValueTuple>(ctx, joinTuple.Pos())
+                        .Name().Build("RightAny"sv)
+                        .Done());
+        }
+        for (ui32 i = 0; i + 1 < linkSettings.Options.size(); i += 2)
+            flags.push_back(
+                    Build<TCoNameValueTuple>(ctx, joinTuple.Pos())
+                        .Name().Build(linkSettings.Options[i])
+                        .Value<TCoAtom>().Build(linkSettings.Options[i + 1])
+                        .Done());
 
         auto dqJoin = Build<TDqJoin>(ctx, joinTuple.Pos())
             .LeftInput(BuildDqJoinInput(ctx, joinTuple.Pos(), left->Input, leftJoinKeys, false))
@@ -522,20 +544,26 @@ TExprBase DqRewriteRightJoinToLeft(const TExprBase node, TExprContext& ctx) {
         return node;
     }
 
-    TMaybeNode<TCoAtomList> newFlags;
-    if (TMaybeNode<TCoAtomList> flags = dqJoin.Flags()) {
-        auto flagsBuilder = Build<TCoAtomList>(ctx, flags.Cast().Pos());
+    TMaybeNode<TCoNameValueTupleList> newFlags;
+    if (TMaybeNode<TCoNameValueTupleList> flags = dqJoin.Flags()) {
+        TVector<TCoNameValueTuple> list;
         for (auto flag: flags.Cast()) {
             TStringBuf tail;
-            if( flag.Value().AfterPrefix("Left", tail)) {
-                flagsBuilder.Add().Value("Right" + TString(tail)).Build();
-            } else if ( flag.Value().AfterPrefix("Right", tail)) {
-                flagsBuilder.Add().Value("Left" + TString(tail)).Build();
+            if( flag.Name().Value().AfterPrefix("Left", tail)) {
+                list.push_back(Build<TCoNameValueTuple>(ctx, flag.Pos())
+                        .Name().Build("Right" + TString(tail))
+                    .Done());
+            } else if ( flag.Name().Value().AfterPrefix("Right", tail)) {
+                list.push_back(Build<TCoNameValueTuple>(ctx, flag.Pos())
+                        .Name().Build("Left" + TString(tail))
+                    .Done());
             } else {
-                flagsBuilder.Add(flag);
+                list.push_back(flag);
             }
         }
-        newFlags = flagsBuilder.Done();
+        newFlags = Build<TCoNameValueTupleList>(ctx, flags.Cast().Pos())
+            .Add(list)
+            .Done();
     }
 
     auto joinKeysBuilder = Build<TDqJoinKeyTupleList>(ctx, dqJoin.Pos());
@@ -640,15 +668,13 @@ TExprBase DqBuildPhyJoin(const TDqJoin& join, bool pushLeftStage, TExprContext&
         return join;
     }
 
-    TExprNode::TListType flags;
     if (const auto maybeFlags = join.Flags()) {
-        flags = maybeFlags.Cast().Ref().ChildrenList();
-    }
-
-    for (auto& flag : flags) {
-        if (flag->IsAtom("LeftAny") || flag->IsAtom("RightAny")) {
-            ctx.AddError(TIssue(ctx.GetPosition(join.Ptr()->Pos()), "ANY join kind is not currently supported"));
-            return join;
+        for (const auto& flag : maybeFlags.Cast()) {
+            auto name = flag.Name().Value();
+            if (name == "LeftAny"sv || name == "RightAny"sv) {
+                ctx.AddError(TIssue(ctx.GetPosition(join.Ptr()->Pos()), "ANY join kind is not currently supported"));
+                return join;
+            }
         }
     }
 
@@ -1514,13 +1540,15 @@ TExprBase DqBuildHashJoin(const TDqJoin& join, EHashJoinMode mode, TExprContext&
             [[fallthrough]];
         case EHashJoinMode::Dict: {
             bool leftAny = false, rightAny = false;
-            for (auto& flag : flags) {
-                if (flag->IsAtom("LeftAny")) {
+            TExprNode::TListType dictFlags;
+            for (const auto& flag : flags) {
+                const auto name = flag->Child(0)->Content();
+                if (name == "LeftAny"sv) {
                     leftAny = true;
-                    flag = ctx.NewAtom(flag->Pos(), "LeftUnique", TNodeFlags::Default);
-                } else if (flag->IsAtom("RightAny")) {
+                    dictFlags.push_back(ctx.NewAtom(flag->Pos(), "LeftUnique", TNodeFlags::Default));
+                } else if (name == "RightAny"sv) {
                     rightAny = true;
-                    flag = ctx.NewAtom(flag->Pos(), "RightUnique", TNodeFlags::Default);
+                    dictFlags.push_back(ctx.NewAtom(flag->Pos(), "RightUnique", TNodeFlags::Default));
                 }
             }
 
@@ -1538,7 +1566,7 @@ TExprBase DqBuildHashJoin(const TDqJoin& join, EHashJoinMode mode, TExprContext&
                                         .Arg(0, "left")
                                         .Arg(1, "right")
                                         .Add(2, join.JoinType().Ptr())
-                                        .List(3).Add(std::move(flags)).Seal()
+                                        .List(3).Add(std::move(dictFlags)).Seal()
                                     .Seal()
                                 .Seal()
                             .Seal()

ydb/library/yql/dq/type_ann/dq_type_ann.cpp

@@ -444,7 +444,8 @@ const TStructExprType* GetDqJoinResultType(const TExprNode::TPtr& input, bool st
         }
     }
 
-    if (!EnsureAtom(*input->Child(TDqJoin::idx_JoinType), ctx)) {
+    const auto& joinType = *input->Child(TDqJoin::idx_JoinType);
+    if (!EnsureAtom(joinType, ctx)) {
         return nullptr;
     }
 
@@ -501,15 +502,35 @@ const TStructExprType* GetDqJoinResultType(const TExprNode::TPtr& input, bool st
     auto rightTableLabel = join.RightLabel().Maybe<TCoAtom>()
         ? join.RightLabel().Cast<TCoAtom>().Value()
         : TStringBuf("");
+    if (input->ChildrenSize() > TDqJoin::idx_JoinAlgo) {
+    const auto& joinAlgo = *input->Child(TDqJoin::idx_JoinAlgo);
+    if (joinAlgo.IsAtom("StreamLooup") && !joinType.IsAtom("Left")) {
+        ctx.AddError(TIssue(ctx.GetPosition(joinType.Pos()), TStringBuilder() << "DqJoin: Unsupported DQ join type " << joinType.Content() << " for join algorithm " << joinAlgo.Content()));
+        return nullptr;
+    }
 
-    if (input->ChildrenSize() > 9U) {
-        for (auto i = 0U; i < input->Tail().ChildrenSize(); ++i) {
-            if (const auto& flag = *input->Tail().Child(i); !flag.IsAtom({"LeftAny", "RightAny"})) {
-                ctx.AddError(TIssue(ctx.GetPosition(flag.Pos()), TStringBuilder() << "Unsupported DQ join option: " << flag.Content()));
+    if (input->ChildrenSize() > TDqJoin::idx_Flags) {
+        auto&& flags = input->Tail();
+        for (ui32 i = 0; i < flags.ChildrenSize(); ++i) {
+            auto&& flag = *flags.Child(i);
+            if ( !EnsureTupleOfAtoms(flag, ctx) || !EnsureTupleMinSize(flag, 1, ctx))
                 return nullptr;
+            auto&& name = *flag.Child(TCoNameValueTuple::idx_Name);
+            if (name.IsAtom({"TTL", "MaxCachedRows", "MaxDelayedRows"})) {
+                if (joinAlgo.IsAtom("StreamLookupJoin")) {
+                   if (!EnsureTupleSize(flag, 2, ctx))
+                       return nullptr;
+                   continue;
+                }
+            } else
+            if (name.IsAtom({"LeftAny", "RightAny"})) {
+                continue;
             }
+            ctx.AddError(TIssue(ctx.GetPosition(flag.Pos()), TStringBuilder() << "DqJoin: Unsupported DQ join option: " << name.Content()));
+            return nullptr;
         }
     }
+    }
 
     return GetDqJoinResultType<IsMapJoin>(join.Pos(), *leftStructType, leftTableLabel, *rightStructType,
         rightTableLabel, join.JoinType(), join.JoinKeys(), ctx);

ydb/library/yql/providers/common/mkql/yql_provider_mkql.cpp

@@ -1700,9 +1700,9 @@ TMkqlCommonCallableCompiler::TShared::TShared() {
 
         auto anyJoinSettings = EAnyJoinSettings::None;
         node.Tail().ForEachChild([&](const TExprNode& flag) {
-            if (flag.IsAtom("LeftAny"))
+            if (flag.Child(0)->IsAtom("LeftAny"))
                anyJoinSettings = EAnyJoinSettings::Right == anyJoinSettings ? EAnyJoinSettings::Both : EAnyJoinSettings::Left;
-            else if (flag.IsAtom("RightAny"))
+            else if (flag.Child(0)->IsAtom("RightAny"))
                anyJoinSettings = EAnyJoinSettings::Left == anyJoinSettings ? EAnyJoinSettings::Both : EAnyJoinSettings::Right;
         });
 

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

@@ -252,6 +252,38 @@ class TDqsPhysicalOptProposalTransformer : public TOptimizeTransformerBase {
         if (!left) {
             return node;
         }
+        TExprNode::TPtr ttl = nullptr;
+        TExprNode::TPtr maxCachedRows = nullptr;
+        TExprNode::TPtr maxDelayedRows = nullptr;
+        if (const auto maybeFlags = join.Flags()) {
+            for (auto&& flag: maybeFlags.Cast()) {
+                auto&& name = flag.Name().Value();
+                if (name == "TTL"sv) {
+                    YQL_ENSURE(flag.Ref().ChildrenSize() == 2);
+                    ttl = flag.Ref().Child(1);
+                } else if (name == "MaxCachedRows"sv) {
+                    YQL_ENSURE(flag.Ref().ChildrenSize() == 2);
+                    maxCachedRows = flag.Ref().Child(1);
+                } else if (name == "MaxDelayedRows"sv) {
+                    YQL_ENSURE(flag.Ref().ChildrenSize() == 2);
+                    maxDelayedRows = flag.Ref().Child(1);
+                } else if (name == "RightAny"sv) {
+                    YQL_ENSURE(flag.Ref().ChildrenSize() == 1);
+                    continue;
+                } else {
+                    Y_DEBUG_ABORT("Impossible flag"); // should have been checked by dq_type_ann
+                }
+            }
+        }
+        if (!ttl) {
+            ttl = ctx.NewAtom(pos, 300);
+        }
+        if (!maxCachedRows) {
+            maxCachedRows = ctx.NewAtom(pos, 1'000'000);
+        }
+        if (!maxDelayedRows) {
+            maxDelayedRows = ctx.NewAtom(pos, 1'000'000);
+        }
         auto cn = Build<TDqCnStreamLookup>(ctx, pos)
             .Output(left.Output().Cast())
             .LeftLabel(join.LeftLabel().Cast<NNodes::TCoAtom>())
@@ -261,9 +293,9 @@ class TDqsPhysicalOptProposalTransformer : public TOptimizeTransformerBase {
             .JoinType(join.JoinType())
             .LeftJoinKeyNames(join.LeftJoinKeyNames())
             .RightJoinKeyNames(join.RightJoinKeyNames())
-            .TTL(ctx.NewAtom(pos, 300)) //TODO configure me
-            .MaxCachedRows(ctx.NewAtom(pos, 1'000'000)) //TODO configure me
-            .MaxDelayedRows(ctx.NewAtom(pos, 1'000'000)) //Configure me
+            .TTL(ttl)
+            .MaxCachedRows(maxCachedRows)
+            .MaxDelayedRows(maxDelayedRows)
         .Done();
 
         auto lambda = Build<TCoLambda>(ctx, pos)

ydb/library/yql/providers/dq/planner/execution_planner.cpp

@@ -613,8 +613,10 @@ namespace NYql::NDqs {
         const auto narrowOutputRowType = GetSeqItemType(streamLookup.Ptr()->GetTypeAnn());
         Y_ABORT_UNLESS(narrowOutputRowType->GetKind() == ETypeAnnotationKind::Struct);
         settings.SetNarrowOutputRowType(NYql::NCommon::GetSerializedTypeAnnotation(narrowOutputRowType));
-        settings.SetCacheLimit(1'000'000); //TODO configure me
-        settings.SetCacheTtlSeconds(60); //TODO configure me
+        // FIXME RESOLVE BEFORE MERGE
+        settings.SetCacheLimit(atoll(streamLookup.MaxCachedRows().StringValue().c_str()));
+        settings.SetCacheTtlSeconds(atoll(streamLookup.TTL().StringValue().c_str()));
+        settings.SetMaxDelayedRows(atoll(streamLookup.MaxDelayedRows().StringValue().c_str()));
 
         const auto inputRowType = GetSeqItemType(streamLookup.Output().Stage().Program().Ref().GetTypeAnn());
         const auto outputRowType = GetSeqItemType(stage.Program().Args().Arg(inputIndex).Ref().GetTypeAnn());

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

@@ -144,12 +144,13 @@ class TDqDataSinkConstraintTransformer : public TVisitorTransformerBase {
         bool leftAny = false, rightAny = false;
         if (const auto maybeJoin = join.Maybe<TDqJoin>()) {
             if (const auto maybeFlags = maybeJoin.Cast().Flags()) {
-                maybeFlags.Cast().Ref().ForEachChild([&](const TExprNode& flag) {
-                    if (flag.IsAtom("LeftAny"))
+                for (const auto &flag: maybeFlags.Cast()) {
+                    const auto name = flag.Name().Value();
+                    if (name == "LeftAny"sv)
                         leftAny = true;
-                    else if (flag.IsAtom("RightAny"))
+                    else if (name == "RightAny"sv)
                         rightAny = true;
-                });
+                }
             }
         }