YQL-12536: Join Any

ref:d26d1cf74808a40d05f5739b597932f07bf1e341

2 files changed, 62 insertions, 39 deletions

diff --git a/ydb/library/yql/dq/opt/dq_opt_join.cpp b/ydb/library/yql/dq/opt/dq_opt_join.cpp
index 548ad89c9a..ab440d1d5e 100644
--- a/ydb/library/yql/dq/opt/dq_opt_join.cpp
+++ b/ydb/library/yql/dq/opt/dq_opt_join.cpp
@@ -56,14 +56,71 @@ TExprBase BuildSkipNullKeys(TExprContext& ctx, TPositionHandle pos,
         .Done();
 };
 
+TExprBase BuildDqJoinInput(TExprContext& ctx, TPositionHandle pos, const TExprBase& input, const TVector<TCoAtom>& keys, bool any) {
+    if (!any) {
+        return input;
+    }
+
+    auto keyExtractor = ctx.Builder(pos)
+        .Lambda()
+            .Param("item")
+            .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
+                auto listBuilder = parent.List();
+                int pos = 0;
+                for (const auto& key : keys) {
+                    listBuilder
+                        .Callable(pos++, "Member")
+                            .Arg(0, "item")
+                            .Add(1, key.Ptr())
+                        .Seal();
+                }
+
+                return listBuilder.Seal();
+            })
+        .Seal()
+        .Build();
+
+    auto condense = Build<TCoLambda>(ctx, pos)
+        .Args({"list"})
+        .Body<TCoCondense1>()
+            .Input("list")
+            .InitHandler(BuildIdentityLambda(pos, ctx))
+            .SwitchHandler()
+                .Args({"item", "state"})
+                .Body<TCoAggrNotEqual>()
+                    .Left<TExprApplier>().Apply(TCoLambda(keyExtractor)).With(0, "item")
+                        .Build()
+                    .Right<TExprApplier>().Apply(TCoLambda(keyExtractor)).With(0, "state")
+                        .Build()
+                    .Build()
+                .Build()
+            .UpdateHandler()
+                .Args({"item", "state"})
+                .Body("state")
+            .Build()
+        .Build()
+        .Done();
+
+    auto partition = Build<TCoPartitionsByKeys>(ctx, pos)
+        .Input(input)
+        .KeySelectorLambda(keyExtractor)
+        .SortDirections<TCoVoid>()
+            .Build()
+        .SortKeySelectorLambda<TCoVoid>()
+            .Build()
+        .ListHandlerLambda(condense)
+        .Done();
+
+    return partition;
+}
+
 TMaybe<TJoinInputDesc> BuildDqJoin(const TCoEquiJoinTuple& joinTuple,
     const THashMap<TStringBuf, TJoinInputDesc>& inputs, TExprContext& ctx)
 {
     auto options = joinTuple.Options();
     auto linkSettings = GetEquiJoinLinkSettings(options.Ref());
-    if (linkSettings.LeftHints.contains("any") || linkSettings.RightHints.contains("any")) {
-        return {};
-    }
+    bool leftAny = linkSettings.LeftHints.contains("any");
+    bool rightAny = linkSettings.RightHints.contains("any");
 
     TMaybe<TJoinInputDesc> left;
     if (joinTuple.LeftScope().Maybe<TCoAtom>()) {
@@ -141,9 +198,9 @@ TMaybe<TJoinInputDesc> BuildDqJoin(const TCoEquiJoinTuple& joinTuple,
     }
 
     auto dqJoin = Build<TDqJoin>(ctx, joinTuple.Pos())
-        .LeftInput(left->Input)
+        .LeftInput(BuildDqJoinInput(ctx, joinTuple.Pos(), left->Input, leftJoinKeys, leftAny))
         .LeftLabel(leftTableLabel)
-        .RightInput(right->Input)
+        .RightInput(BuildDqJoinInput(ctx, joinTuple.Pos(), right->Input, rightJoinKeys, rightAny))
         .RightLabel(rightTableLabel)
         .JoinType(joinTuple.Type())
         .JoinKeys(joinKeysBuilder.Done())
@@ -265,37 +322,6 @@ bool CheckJoinColumns(const TExprBase& node) {
     }
 }
 
-bool CheckJoinTupleLinkSettings(const TCoEquiJoinTuple& joinTuple) {
-    auto options = joinTuple.Options();
-    auto linkSettings = GetEquiJoinLinkSettings(options.Ref());
-    if (linkSettings.LeftHints.contains("any") || linkSettings.RightHints.contains("any")) {
-        return false;
-    }
-
-    bool result = true;
-    if (!joinTuple.LeftScope().Maybe<TCoAtom>()) {
-        result &= CheckJoinTupleLinkSettings(joinTuple.LeftScope().Cast<TCoEquiJoinTuple>());
-    }
-    if (!result) {
-        return result;
-    }
-
-    if (!joinTuple.RightScope().Maybe<TCoAtom>()) {
-        result &= CheckJoinTupleLinkSettings(joinTuple.RightScope().Cast<TCoEquiJoinTuple>());
-    }
-    return result;
-}
-
-bool CheckJoinLinkSettings(const TExprBase& node) {
-    if (!node.Maybe<TCoEquiJoin>()) {
-        return true;
-    }
-    auto equiJoin = node.Cast<TCoEquiJoin>();
-    YQL_ENSURE(equiJoin.ArgCount() >= 4);
-    auto joinTuple = equiJoin.Arg(equiJoin.ArgCount() - 2).Cast<TCoEquiJoinTuple>();
-    return CheckJoinTupleLinkSettings(joinTuple);
-}
-
 /**
  * Rewrite `EquiJoin` to a number of `DqJoin` callables. This is done to simplify next step of building
  * physical stages with join operators.
diff --git a/ydb/library/yql/providers/dq/provider/yql_dq_recapture.cpp b/ydb/library/yql/providers/dq/provider/yql_dq_recapture.cpp
index 79b9c9ff09..058c99a9fc 100644
--- a/ydb/library/yql/providers/dq/provider/yql_dq_recapture.cpp
+++ b/ydb/library/yql/providers/dq/provider/yql_dq_recapture.cpp
@@ -173,9 +173,6 @@ private:
         } else if (TMaybeNode<TCoEquiJoin>(&node) && !NDq::CheckJoinColumns(expr)) {
             AddInfo(ctx, TStringBuilder() << "unsupported join column");
             good = false;
-        } else if (TMaybeNode<TCoEquiJoin>(&node) && !NDq::CheckJoinLinkSettings(expr)) {
-            AddInfo(ctx, TStringBuilder() << "unsupported join any");
-            good = false;
         } else if (node.ChildrenSize() > 1 && TCoDataSource::Match(node.Child(1))) {
             auto dataSourceName = node.Child(1)->Child(0)->Content();
             if (dataSourceName != DqProviderName && !node.IsCallable(ConfigureName)) {