Add HashJoinMode pragma to YDB

4 files changed, 37 insertions, 1 deletions

diff --git a/ydb/core/kqp/opt/physical/kqp_opt_phy.cpp b/ydb/core/kqp/opt/physical/kqp_opt_phy.cpp
index bea79d22d2b..8da951da3bb 100644
--- a/ydb/core/kqp/opt/physical/kqp_opt_phy.cpp
+++ b/ydb/core/kqp/opt/physical/kqp_opt_phy.cpp
@@ -363,7 +363,9 @@ protected:
     TMaybeNode<TExprBase> BuildJoin(TExprBase node, TExprContext& ctx,
         IOptimizationContext& optCtx, const TGetParents& getParents)
     {
-        TExprBase output = DqBuildJoin(node, ctx, optCtx, *getParents(), IsGlobal, /*pushLeftStage =*/ !KqpCtx.IsDataQuery() && AllowFuseJoinInputs(node));
+        TExprBase output = DqBuildJoin(node, ctx, optCtx, *getParents(), IsGlobal,
+            /*pushLeftStage =*/ !KqpCtx.IsDataQuery() && AllowFuseJoinInputs(node), KqpCtx.Config->GetHashJoinMode()
+        );
         DumpAppliedRule("BuildJoin", node.Ptr(), output.Ptr(), ctx);
         return output;
     }
diff --git a/ydb/core/kqp/provider/yql_kikimr_settings.cpp b/ydb/core/kqp/provider/yql_kikimr_settings.cpp
index 4f418f5bfec..c0ca661999b 100644
--- a/ydb/core/kqp/provider/yql_kikimr_settings.cpp
+++ b/ydb/core/kqp/provider/yql_kikimr_settings.cpp
@@ -50,6 +50,7 @@ TKikimrConfiguration::TKikimrConfiguration() {
     REGISTER_SETTING(*this, _ResultRowsLimit);
     REGISTER_SETTING(*this, EnableSystemColumns);
     REGISTER_SETTING(*this, UseLlvm);
+    REGISTER_SETTING(*this, HashJoinMode).Parser([](const TString& v) { return FromString<NDq::EHashJoinMode>(v); });
 
     REGISTER_SETTING(*this, OptDisableJoinRewrite);
     REGISTER_SETTING(*this, OptDisableJoinTableLookup);
@@ -130,6 +131,11 @@ EOptionalFlag TKikimrSettings::GetUseLlvm() const {
     return GetOptionalFlagValue(UseLlvm.Get());
 }
 
+NDq::EHashJoinMode TKikimrSettings::GetHashJoinMode() const {
+    auto maybeHashJoinMode = HashJoinMode.Get();
+    return maybeHashJoinMode ? *maybeHashJoinMode : NDq::EHashJoinMode::Off;
+}
+
 TKikimrSettings::TConstPtr TKikimrConfiguration::Snapshot() const {
     return std::make_shared<const TKikimrSettings>(*this);
 }
diff --git a/ydb/core/kqp/provider/yql_kikimr_settings.h b/ydb/core/kqp/provider/yql_kikimr_settings.h
index 1b9793a6933..4e3366023c4 100644
--- a/ydb/core/kqp/provider/yql_kikimr_settings.h
+++ b/ydb/core/kqp/provider/yql_kikimr_settings.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <ydb/library/yql/dq/common/dq_common.h>
 #include <ydb/library/yql/providers/common/config/yql_dispatch.h>
 #include <ydb/library/yql/providers/common/config/yql_setting.h>
 #include <ydb/core/protos/config.pb.h>
@@ -40,6 +41,7 @@ struct TKikimrSettings {
     NCommon::TConfSetting<ui32, false> _ResultRowsLimit;
     NCommon::TConfSetting<bool, false> EnableSystemColumns;
     NCommon::TConfSetting<bool, false> UseLlvm;
+    NCommon::TConfSetting<NDq::EHashJoinMode, false> HashJoinMode;
 
     /* Disable optimizer rules */
     NCommon::TConfSetting<bool, false> OptDisableJoinRewrite;
@@ -73,6 +75,7 @@ struct TKikimrSettings {
     bool HasOptUseFinalizeByKey() const;
     EOptionalFlag GetOptPredicateExtract() const;
     EOptionalFlag GetUseLlvm() const;
+    NDq::EHashJoinMode GetHashJoinMode() const;
 
     // WARNING: For testing purposes only, inplace update is not ready for production usage.
     bool HasOptEnableInplaceUpdate() const;
diff --git a/ydb/core/kqp/ut/join/kqp_join_ut.cpp b/ydb/core/kqp/ut/join/kqp_join_ut.cpp
index b832fbb171a..fedaa77073e 100644
--- a/ydb/core/kqp/ut/join/kqp_join_ut.cpp
+++ b/ydb/core/kqp/ut/join/kqp_join_ut.cpp
@@ -1278,6 +1278,31 @@ Y_UNIT_TEST_SUITE(KqpJoin) {
         CompareYson(R"([[2]])", FormatResultSetYson(result.GetResultSet(0)));
     }
 
+    Y_UNIT_TEST(JoinPragmaHashJoinMode) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        CreateSampleTables(session);
+
+        auto query = Q1_(R"(
+            PRAGMA ydb.HashJoinMode='grace';
+
+            SELECT t1.Value
+                FROM `/Root/Join1_1` AS t1
+                INNER JOIN `/Root/Join1_2` AS t2
+                ON t1.Value == t2.Value;
+        )");
+
+        auto result = session.ExecuteDataQuery(query, TTxControl::BeginTx().CommitTx(), BuildPureTableParams(db)).GetValueSync();
+
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        CompareYson(R"([])", FormatResultSetYson(result.GetResultSet(0)));
+
+        auto explain = session.ExplainDataQuery(query).GetValueSync();
+        UNIT_ASSERT(explain.GetAst().Contains("GraceJoinCore"));
+    }
+
 }
 
 } // namespace NKqp