[HashJoin] move block/scalar common parts (#47728)

3 files changed, 322 insertions, 400 deletions

diff --git a/ydb/library/yql/dq/comp_nodes/dq_block_hash_join.cpp b/ydb/library/yql/dq/comp_nodes/dq_block_hash_join.cpp
index 78c7ea6a99f..64eea184600 100644
--- a/ydb/library/yql/dq/comp_nodes/dq_block_hash_join.cpp
+++ b/ydb/library/yql/dq/comp_nodes/dq_block_hash_join.cpp
@@ -21,7 +21,7 @@ using TDqJoinImplRenames = TDqRenames<ESide>;
 
 struct TDqBlockJoinContext {
     TSides<TVector<TBlockType*>> InputTypes;
-    TSides<TVector<int>> KeyColumns;
+    TSides<TVector<ui32>> KeyColumns;
     TVector<TBlockType*> ResultItemTypes;
     TDqJoinImplRenames Renames;
     EJoinKind Kind;
@@ -130,14 +130,14 @@ class TBlockPackedTupleSource : public NNonCopyable::TMoveOnly {
 };
 
 template<EJoinKind Kind>
-struct TRenamesPackedTupleOutput : NNonCopyable::TMoveOnly {
+struct TRenamesPackedTupleOutput : TPackedTupleOutputBase<Kind, IBlockLayoutConverter> {
+    using TBase = TPackedTupleOutputBase<Kind, IBlockLayoutConverter>;
+
     TRenamesPackedTupleOutput(const TDqBlockJoinContext* meta, TSides<IBlockLayoutConverter*> converters,
                               const TVector<TType*>& userNullTypes, arrow::MemoryPool& arrowPool)
-        : Renames_(&meta->Renames)
-        , Converters_(converters)
-        , LeftIsBuild_(meta->Settings.LeftIsBuild())
+        : TBase(&meta->Renames, converters, meta->Settings.LeftIsBuild())
     {
-        if constexpr (!std::is_same_v<decltype(Nulls_), Empty>) {
+        if constexpr (!std::is_same_v<typename TBase::BuildNullIfNeeded, typename TBase::Empty>) {
             TVector<arrow::Datum> nulls;
             for(auto* type:userNullTypes) {
                 auto strname = type->GetKindAsStr();
@@ -147,60 +147,33 @@ struct TRenamesPackedTupleOutput : NNonCopyable::TMoveOnly {
                 builder->Add(NYql::NUdf::TBlockItem{});
                 nulls.push_back(builder->Build(true));
             }
-            if (LeftIsBuild_) {
-                Converters_.Probe->Pack(nulls, Nulls_);
+            if (this->LeftIsBuild_) {
+                this->Converters_.Probe->Pack(nulls, this->Nulls_);
             } else {
-                Converters_.Build->Pack(nulls, Nulls_);
+                this->Converters_.Build->Pack(nulls, this->Nulls_);
             }
         }
     }
 
-    int Columns() const {
-        return Renames_->size();
-    }
-
-    i64 SizeTuples() const {
-        AssertSizeIsSane();
-        return Output_.Probe.NTuples;
-    }
-
-
-
-    using TuplePairs = TSides<TPackResult>;
-    struct Empty {};
-    using BuildNullIfNeeded = std::conditional_t<Kind==EJoinKind::Left, TPackResult, Empty>;
+    struct TFlushResult {
+        TVector<arrow::Datum> Columns;
+        i64 Rows;
+    };
 
-    auto MakeConsumeFn() {
-        struct ConsumeFn {
-            TRenamesPackedTupleOutput& self;
-            void operator()(TSides<TSingleTuple> tuples) {
-                for(ESide side: EachSide) {
-                    self.Output_.SelectSide(side).AppendTuple(tuples.SelectSide(side), self.Converters_.SelectSide(side)->GetTupleLayout());
-                }
-            }
-            void operator()(TSingleTuple tuple) {
-                if constexpr (Kind == EJoinKind::Left) {
-                    TSingleTuple null{.PackedData = self.Nulls_.PackedTuples.data(), .OverflowBegin = self.Nulls_.Overflow.data() };
-                    if (self.LeftIsBuild_) {
-                        this->operator()(TSides<TSingleTuple>{.Build = tuple, .Probe = null});
-                    } else {
-                        this->operator()(TSides<TSingleTuple>{.Build = null, .Probe = tuple});
-                    }
-                } else if constexpr(SemiOrOnlyJoin(Kind)) {
-                    self.Output_.Probe.AppendTuple(tuple, self.Converters_.Probe->GetTupleLayout());
-                }
-            }
-        };
-        return ConsumeFn{*this};
+    TFlushResult Flush() {
+        TFlushResult res;
+        res.Rows = this->SizeTuples();
+        res.Columns = FlushAndApplyRenames();
+        return res;
     }
 
     TVector<arrow::Datum> FlushAndApplyRenames() {
         if constexpr(LeftSemiOrOnly(Kind)) {
             TVector<arrow::Datum> out;
-            Converters_.Probe->Unpack(Output_.Probe, out);
-            Output_.Probe.Clear();
+            this->Converters_.Probe->Unpack(this->Output_.Probe, out);
+            this->Output_.Probe.Clear();
             TVector<arrow::Datum> renamed;
-            for(auto rename: *Renames_){
+            for(auto rename: *this->Renames_){
                 MKQL_ENSURE(rename.Side == ESide::Probe, "renames in Semi or Only Left Join shouldn't contain columns from right side");
                 renamed.push_back(out[rename.Index]);
             }
@@ -208,42 +181,16 @@ struct TRenamesPackedTupleOutput : NNonCopyable::TMoveOnly {
         } else {
             TSides<TVector<arrow::Datum>> sides;
             for(ESide side: EachSide) {
-                Converters_.SelectSide(side)->Unpack(Output_.SelectSide(side), sides.SelectSide(side));
-                Output_.SelectSide(side).Clear();
+                this->Converters_.SelectSide(side)->Unpack(this->Output_.SelectSide(side), sides.SelectSide(side));
+                this->Output_.SelectSide(side).Clear();
             }
             TVector<arrow::Datum> renamed;
-            for (auto rename : *Renames_) {
+            for (auto rename : *this->Renames_) {
                 renamed.push_back(sides.SelectSide(rename.Side)[rename.Index]);
             }
             return renamed;
         }
     }
-
-  private:
-    TSides<TVector<arrow::Datum>> Flush() {
-        TSides<TVector<arrow::Datum>> out;
-        for(ESide side: EachSide) {
-
-            Converters_.SelectSide(side)->Unpack(Output_.SelectSide(side), out.SelectSide(side));
-            Output_.SelectSide(side).Clear();
-
-        }
-
-        return out;
-    }
-    void AssertSizeIsSane() const{
-        if constexpr (Kind == EJoinKind::LeftOnly || Kind==EJoinKind::LeftSemi) {
-            MKQL_ENSURE(Output_.Build.NTuples == 0, "Left Only and Left Semi join types shouldn't collect any Build(right) tuples");
-        } else if constexpr (Kind == EJoinKind::Left || Kind == EJoinKind::Inner) {
-            MKQL_ENSURE(Output_.Build.NTuples == Output_.Probe.NTuples, "Inner and Left join types must collect same amount of tuples from build and probe");
-        }
-    }
-
-    TuplePairs Output_;
-    const TDqJoinImplRenames* Renames_;
-    TSides<IBlockLayoutConverter*> Converters_;
-    BuildNullIfNeeded Nulls_;
-    bool LeftIsBuild_;
 };
 
 template <EJoinKind Kind> class TBlockHashJoinWrapper : public TMutableComputationNode<TBlockHashJoinWrapper<Kind>> {
@@ -262,10 +209,7 @@ template <EJoinKind Kind> class TBlockHashJoinWrapper : public TMutableComputati
         TSides<std::unique_ptr<IBlockLayoutConverter>> layouts;
         const auto& userTypes = Meta_->UserTypes;
         for(ESide side: EachSide) {
-            TVector<NPackedTuple::EColumnRole> roles(userTypes.SelectSide(side).size(), NPackedTuple::EColumnRole::Payload);
-            for (int column : Meta_->KeyColumns.SelectSide(side)) {
-                roles[column] = NPackedTuple::EColumnRole::Key;
-            }
+            const auto roles = MakeColumnRoles(userTypes.SelectSide(side).size(), Meta_->KeyColumns.SelectSide(side));
             layouts.SelectSide(side) = MakeBlockLayoutConverter(helper, userTypes.SelectSide(side), roles, &ctx.ArrowMemoryPool);
         }
         const auto& userNullTypes = (Kind == EJoinKind::Left && Meta_->Settings.LeftIsBuild()) ? userTypes.Probe : userTypes.Build;
@@ -295,17 +239,12 @@ template <EJoinKind Kind> class TBlockHashJoinWrapper : public TMutableComputati
             , Output_(meta, {.Build = Converters_.Build.get(), .Probe = Converters_.Probe.get()}, userBuildTypes, ctx.ArrowMemoryPool)
         {}
 
-        NUdf::EFetchStatus FlushTo(NUdf::TUnboxedValue* output) {
-            MKQL_ENSURE(Output_.SizeTuples() != 0, "make sure we are flushing something, not empty set of tuples");
-            i64 rows = Output_.SizeTuples();
-            TVector<arrow::Datum> arrowOutput = Output_.FlushAndApplyRenames();
-            for (int colIndex = 0; colIndex < Output_.Columns(); ++colIndex) {
-                output[colIndex] = Ctx_->HolderFactory.CreateArrowBlock(std::move(arrowOutput[colIndex]), Ctx_->RuntimeSettings.DatumValidation.Get());
+        void WriteFlushToOutput(NUdf::TUnboxedValue* output, typename TRenamesPackedTupleOutput<Kind>::TFlushResult flush) {
+            const int cols = Output_.Columns();
+            for (int colIndex = 0; colIndex < cols; ++colIndex) {
+                output[colIndex] = Ctx_->HolderFactory.CreateArrowBlock(std::move(flush.Columns[colIndex]), Ctx_->RuntimeSettings.DatumValidation.Get());
             }
-            output[Output_.Columns()] = Ctx_->HolderFactory.CreateArrowBlock(arrow::Datum(static_cast<uint64_t>(rows)), Ctx_->RuntimeSettings.DatumValidation.Get());
-
-            MKQL_ENSURE(Output_.SizeTuples() == 0, "something left after flush??");
-            return NYql::NUdf::EFetchStatus::Ok;
+            output[cols] = Ctx_->HolderFactory.CreateArrowBlock(arrow::Datum(static_cast<uint64_t>(flush.Rows)), Ctx_->RuntimeSettings.DatumValidation.Get());
         }
 
       private:
@@ -313,29 +252,18 @@ template <EJoinKind Kind> class TBlockHashJoinWrapper : public TMutableComputati
             size_t expectedSize = Meta_->Renames.size() + 1;
             MKQL_ENSURE(width == expectedSize,
                         Sprintf("runtime(%i) vs compile-time(%i) tuple width mismatch", width, expectedSize));
-            if (Finished_) {
+            switch (RunPackedHashJoinBatch<MaxOutputRows_>(
+                *Ctx_, Join_, Output_, [&](auto flush) { WriteFlushToOutput(output, std::move(flush)); })) {
+            case EFetchResult::One:
+                return NYql::NUdf::EFetchStatus::Ok;
+            case EFetchResult::Yield:
+                return NYql::NUdf::EFetchStatus::Yield;
+            case EFetchResult::Finish:
                 return NYql::NUdf::EFetchStatus::Finish;
+            default:
+                MKQL_ENSURE(false, "unexpected fetch result");
             }
-            auto outputIsFull = [&]() {
-                return Output_.SizeTuples() >= MaxOutputRows_;
-            };
-            while (!outputIsFull()) {
-                auto res = Join_.MatchRows(*Ctx_, Output_.MakeConsumeFn(), outputIsFull);
-                switch (res) {
-                case EFetchResult::Finish: {
-                    if (Output_.SizeTuples() == 0) {
-                        return NYql::NUdf::EFetchStatus::Finish;
-                    }
-                    Finished_ = true;
-                    return FlushTo(output);
-                }
-                case EFetchResult::Yield:
-                    return NYql::NUdf::EFetchStatus::Yield;
-                case EFetchResult::One:
-                    break;
-                }
-            }
-            return FlushTo(output);
+            Y_UNREACHABLE();
         }
 
       private:
@@ -345,7 +273,6 @@ template <EJoinKind Kind> class TBlockHashJoinWrapper : public TMutableComputati
         TComputationContext* Ctx_;
         TRenamesPackedTupleOutput<Kind> Output_;
         static constexpr i64 MaxOutputRows_ = 10000;
-        bool Finished_ = false;
     };
 
     void RegisterDependencies() const final {
@@ -395,32 +322,14 @@ IComputationNode* WrapDqBlockHashJoin(TCallable& callable, const TComputationNod
         MKQL_ENSURE(blockType->IsBlock(), "Expected block types as wide components of right stream");
         meta.InputTypes.Build.push_back(AS_TYPE(TBlockType, blockType));
     }
-    const auto joinKindNode = callable.GetInput(2);
-    const auto rawKind = AS_VALUE(TDataLiteral, joinKindNode)->AsValue().Get<ui32>();
-    const auto joinKind = GetJoinKind(rawKind);
+    const auto parsed = ParseCommonHashJoinArgs(callable);
+    const auto joinKind = parsed.Kind;
     meta.Kind = joinKind;
-
-    const auto leftKeyColumnsLiteral = callable.GetInput(3);
-    const auto leftKeyColumnsTuple = AS_VALUE(TTupleLiteral, leftKeyColumnsLiteral);
-    for (ui32 i = 0; i < leftKeyColumnsTuple->GetValuesCount(); i++) {
-        const auto item = AS_VALUE(TDataLiteral, leftKeyColumnsTuple->GetValue(i));
-        meta.KeyColumns.Probe.emplace_back(item->AsValue().Get<ui32>());
-    }
-
-    const auto rightKeyColumnsLiteral = callable.GetInput(4);
-    const auto rightKeyColumnsTuple = AS_VALUE(TTupleLiteral, rightKeyColumnsLiteral);
-    for (ui32 i = 0; i < rightKeyColumnsTuple->GetValuesCount(); i++) {
-        const auto item = AS_VALUE(TDataLiteral, rightKeyColumnsTuple->GetValue(i));
-        meta.KeyColumns.Build.emplace_back(item->AsValue().Get<ui32>());
-    }
-    TDqUserRenames userRenames =
-        FromGraceFormat(TGraceJoinRenames::FromRuntimeNodes(callable.GetInput(5), callable.GetInput(6)));
-
-    MKQL_ENSURE(meta.KeyColumns.Build.size() == meta.KeyColumns.Probe.size(), "Key columns mismatch");
+    meta.KeyColumns = parsed.KeyColumns;
 
     const auto leftStream = LocateNode(ctx.NodeLocator, callable, 0);
     const auto rightStream = LocateNode(ctx.NodeLocator, callable, 1);
-    ValidateRenames(userRenames, joinKind, std::ssize(meta.InputTypes.Probe) - 1,
+    ValidateRenames(parsed.UserRenames, joinKind, std::ssize(meta.InputTypes.Probe) - 1,
                     std::ssize(meta.InputTypes.Build) - 1);
     for(ESide side: EachSide) {
         int size = std::ssize(meta.InputTypes.SelectSide(side));
@@ -434,16 +343,7 @@ IComputationNode* WrapDqBlockHashJoin(TCallable& callable, const TComputationNod
         }
     }
 
-    for (auto rename : userRenames) {
-        ESide thisSide = [&] {
-            if (rename.Side == EJoinSide::kLeft) {
-                return ESide::Probe;
-            } else {
-                return ESide::Build;
-            }
-        }();
-        meta.Renames.push_back({.Index = rename.Index, .Side = thisSide});
-    }
+    meta.Renames = BuildImplRenames(parsed.UserRenames);
 
     {
         const auto settingsTuple = AS_VALUE(TTupleLiteral, callable.GetInput(7));
@@ -459,84 +359,28 @@ IComputationNode* WrapDqBlockHashJoin(TCallable& callable, const TComputationNod
         }
     }
 
-    for (ESide side : EachSide) {
-        auto& keyColumns = meta.KeyColumns.SelectSide(side);
-        int numDataCols = meta.InputTypes.SelectSide(side).size() - 1;
-        int numKeys = keyColumns.size();
-
-        bool needsReorder = false;
-        for (int i = 0; i < numKeys; ++i) {
-            if (keyColumns[i] != i) {
-                needsReorder = true;
-                break;
-            }
-        }
-        if (!needsReorder) {
-            continue;
-        }
-
-        TVector<int> perm(numDataCols);
-        std::iota(perm.begin(), perm.end(), 0);
-        for (int i = 0; i < numKeys; ++i) {
-            auto it = std::find(perm.begin(), perm.end(), keyColumns[i]);
-            std::swap(perm[i], *it);
-        }
-
-        meta.ColumnPermutation.SelectSide(side) = perm;
-
-        auto origTypes = TVector<TBlockType*>(meta.InputTypes.SelectSide(side).begin(),
-                                              meta.InputTypes.SelectSide(side).begin() + numDataCols);
-        for (int i = 0; i < numDataCols; ++i) {
-            meta.InputTypes.SelectSide(side)[i] = origTypes[perm[i]];
-        }
-
-        TVector<int> inv(numDataCols);
-        for (int i = 0; i < numDataCols; ++i) {
-            inv[perm[i]] = i;
-        }
-        for (auto& rename : meta.Renames) {
-            if (rename.Side == side) {
-                rename.Index = inv[rename.Index];
-            }
-        }
-
-        for (int i = 0; i < numKeys; ++i) {
-            keyColumns[i] = i;
-        }
-    }
+    ApplyKeyColumnPermutation(meta.KeyColumns, meta.InputTypes, /* trailingColumns */ 1, meta.Renames,
+                              meta.ColumnPermutation);
 
     for(ESide side: EachSide) {
         meta.TempStateIndes.SelectSide(side) = std::exchange(ctx.Mutables.CurValueIndex, meta.InputTypes.SelectSide(side).size() + ctx.Mutables.CurValueIndex);
     }
 
-    const ESide nullableSide = meta.Settings.LeftIsBuild() ? ESide::Probe : ESide::Build;
+    TSides<TVector<TType*>> itemTypes;
     for (ESide side : EachSide) {
         for (int index = 0; index < std::ssize(meta.InputTypes.SelectSide(side)) - 1; ++index) {
-            TType* thisType = meta.InputTypes.SelectSide(side)[index]->GetItemType();
-            if (meta.Kind == EJoinKind::Left && side == nullableSide && !thisType->IsOptional()) {
-                meta.UserTypes.SelectSide(side).push_back(TOptionalType::Create(thisType, ctx.Env));
-            } else {
-                meta.UserTypes.SelectSide(side).push_back(thisType);
-            }
+            itemTypes.SelectSide(side).push_back(meta.InputTypes.SelectSide(side)[index]->GetItemType());
         }
     }
+    const ESide nullableSide = meta.Settings.LeftIsBuild() ? ESide::Probe : ESide::Build;
+    meta.UserTypes = ForceOptionalOnNullableSide(itemTypes, meta.Kind, nullableSide, ctx.Env);
 
     const auto streams = meta.Settings.LeftIsBuild()
         ? TSides<IComputationNode*>{.Build = leftStream, .Probe = rightStream}
         : TSides<IComputationNode*>{.Build = rightStream, .Probe = leftStream};
 
-    using enum EJoinKind;
-    if (joinKind == Inner) {
-        return new TBlockHashJoinWrapper<Inner>(ctx.Mutables, meta, streams);
-    } else if (joinKind == LeftOnly) {
-        return new TBlockHashJoinWrapper<LeftOnly>(ctx.Mutables, meta, streams);
-    } else if (joinKind == LeftSemi) {
-        return new TBlockHashJoinWrapper<LeftSemi>(ctx.Mutables, meta, streams);
-    } else if (joinKind == Left) {
-        return new TBlockHashJoinWrapper<Left>(ctx.Mutables, meta, streams);
-    } else {
-        MKQL_ENSURE(false, "unsupported join type in block hash join");
-    }
+    return DispatchHashJoinByKind<TBlockHashJoinWrapper, IComputationNode>(
+        joinKind, "unsupported join type in block hash join", ctx.Mutables, std::move(meta), streams);
 }
 
 } // namespace NKikimr::NMiniKQL
diff --git a/ydb/library/yql/dq/comp_nodes/dq_join_common.h b/ydb/library/yql/dq/comp_nodes/dq_join_common.h
index 1701383a97b..d2641af6432 100644
--- a/ydb/library/yql/dq/comp_nodes/dq_join_common.h
+++ b/ydb/library/yql/dq/comp_nodes/dq_join_common.h
@@ -1,6 +1,8 @@
 #pragma once
 #include "dq_hash_join_table.h"
 #include "dq_block_hash_join_settings.h"
+#include <algorithm>
+#include <numeric>
 #include <vector>
 #include <ydb/library/yql/dq/comp_nodes/hash_join_utils/alloc.h>
 #include <ydb/library/yql/dq/comp_nodes/hash_join_utils/layout_converter_common.h>
@@ -9,6 +11,7 @@
 #include <yql/essentials/minikql/computation/mkql_block_reader.h>
 #include <yql/essentials/minikql/computation/mkql_computation_node.h>
 #include <yql/essentials/minikql/computation/mkql_computation_node_holders.h>
+#include <yql/essentials/minikql/mkql_node_cast.h>
 #include <yql/essentials/minikql/mkql_program_builder.h>
 
 namespace NKikimr::NMiniKQL {
@@ -257,6 +260,8 @@ template <typename Source> class TInMemoryHashJoin {
   public:
     using TTable = NJoinTable::TNeumannJoinTable;
 
+    static constexpr bool FlushOnYield = false;
+
     TInMemoryHashJoin(TSides<Source> sources, TComputationContext& ctx, TString componentName,
                     TSides<const NPackedTuple::TTupleLayout*> layouts)
         : Logger_(ctx.MakeLogger())
@@ -369,6 +374,8 @@ template <typename Source, TSpillerSettings Settings, EJoinKind Kind> class THyb
   public:
     using TTable = NJoinTable::TNeumannJoinTable;
 
+    static constexpr bool FlushOnYield = false;
+
     struct Init {};
 
     struct FetchingBuild {
@@ -831,4 +838,235 @@ template <typename Source, TSpillerSettings Settings, EJoinKind Kind> class THyb
 };
 } // namespace NJoinPackedTuples
 
+struct TParsedHashJoinArgs {
+    EJoinKind Kind;
+    TSides<TVector<ui32>> KeyColumns;
+    TDqUserRenames UserRenames;
+};
+
+inline TParsedHashJoinArgs ParseCommonHashJoinArgs(TCallable& callable) {
+    TParsedHashJoinArgs res;
+    res.Kind = GetJoinKind(AS_VALUE(TDataLiteral, callable.GetInput(2))->AsValue().Get<ui32>());
+
+    const auto parseKeys = [](TRuntimeNode node) {
+        TVector<ui32> keys;
+        const auto tuple = AS_VALUE(TTupleLiteral, node);
+        for (ui32 i = 0; i < tuple->GetValuesCount(); ++i) {
+            keys.push_back(AS_VALUE(TDataLiteral, tuple->GetValue(i))->AsValue().Get<ui32>());
+        }
+        return keys;
+    };
+    res.KeyColumns.Probe = parseKeys(callable.GetInput(3));
+    res.KeyColumns.Build = parseKeys(callable.GetInput(4));
+    MKQL_ENSURE(res.KeyColumns.Build.size() == res.KeyColumns.Probe.size(), "Key columns mismatch");
+
+    res.UserRenames = FromGraceFormat(TGraceJoinRenames::FromRuntimeNodes(callable.GetInput(5), callable.GetInput(6)));
+    return res;
+}
+
+inline TDqRenames<ESide> BuildImplRenames(const TDqUserRenames& userRenames) {
+    TDqRenames<ESide> renames;
+    for (auto rename : userRenames) {
+        const ESide side = rename.Side == EJoinSide::kLeft ? ESide::Probe : ESide::Build;
+        renames.push_back({.Index = rename.Index, .Side = side});
+    }
+    return renames;
+}
+
+template <typename TKeyColumns>
+TVector<NPackedTuple::EColumnRole> MakeColumnRoles(size_t width, const TKeyColumns& keyColumns) {
+    TVector<NPackedTuple::EColumnRole> roles(width, NPackedTuple::EColumnRole::Payload);
+    for (auto column : keyColumns) {
+        roles[column] = NPackedTuple::EColumnRole::Key;
+    }
+    return roles;
+}
+
+template <template <EJoinKind> class Wrapper, typename TResult, typename... Args>
+TResult* DispatchHashJoinByKind(EJoinKind kind, TStringBuf unsupportedMessage, Args&&... args) {
+    using enum EJoinKind;
+    switch (kind) {
+    case Inner:
+        return new Wrapper<Inner>(std::forward<Args>(args)...);
+    case LeftOnly:
+        return new Wrapper<LeftOnly>(std::forward<Args>(args)...);
+    case LeftSemi:
+        return new Wrapper<LeftSemi>(std::forward<Args>(args)...);
+    case Left:
+        return new Wrapper<Left>(std::forward<Args>(args)...);
+    default:
+        break;
+    }
+    MKQL_ENSURE(false, unsupportedMessage);
+    Y_UNREACHABLE();
+}
+
+template <typename TKeyCols, typename TInputTypes>
+void ApplyKeyColumnPermutation(TSides<TKeyCols>& keyColumns, TSides<TInputTypes>& inputTypes, int trailingColumns,
+                               TDqRenames<ESide>& renames, TSides<TVector<int>>& outColumnPermutation) {
+    for (ESide side : EachSide) {
+        auto& keyCols = keyColumns.SelectSide(side);
+        auto& types = inputTypes.SelectSide(side);
+        const int numDataCols = std::ssize(types) - trailingColumns;
+        const int numKeys = std::ssize(keyCols);
+
+        bool needsReorder = false;
+        for (int i = 0; i < numKeys; ++i) {
+            if (static_cast<int>(keyCols[i]) != i) {
+                needsReorder = true;
+                break;
+            }
+        }
+        if (!needsReorder) {
+            continue;
+        }
+
+        TVector<int> perm(numDataCols);
+        std::iota(perm.begin(), perm.end(), 0);
+        for (int i = 0; i < numKeys; ++i) {
+            const int keyColumn = static_cast<int>(keyCols[i]);
+            MKQL_ENSURE(keyColumn >= 0 && keyColumn < numDataCols,
+                        Sprintf("key column index %i on %s side is out of range [0, %i)", keyColumn, AsString(side),
+                                numDataCols));
+            auto it = std::find(perm.begin() + i, perm.end(), keyColumn);
+            MKQL_ENSURE(it != perm.end(),
+                        Sprintf("key column index %i on %s side is duplicated or could not be placed", keyColumn,
+                                AsString(side)));
+            std::swap(perm[i], *it);
+        }
+
+        outColumnPermutation.SelectSide(side) = perm;
+
+        using TElem = std::decay_t<decltype(types[0])>;
+        const TVector<TElem> orig(types.begin(), types.begin() + numDataCols);
+        for (int i = 0; i < numDataCols; ++i) {
+            types[i] = orig[perm[i]];
+        }
+
+        TVector<int> inv(numDataCols);
+        for (int i = 0; i < numDataCols; ++i) {
+            inv[perm[i]] = i;
+        }
+        for (auto& rename : renames) {
+            if (rename.Side == side) {
+                rename.Index = inv[rename.Index];
+            }
+        }
+
+        for (int i = 0; i < numKeys; ++i) {
+            keyCols[i] = i;
+        }
+    }
+}
+
+inline TSides<TVector<TType*>> ForceOptionalOnNullableSide(const TSides<TVector<TType*>>& itemTypes, EJoinKind kind,
+                                                           ESide nullableSide, const TTypeEnvironment& env) {
+    TSides<TVector<TType*>> userTypes;
+    for (ESide side : EachSide) {
+        for (TType* thisType : itemTypes.SelectSide(side)) {
+            if (kind == EJoinKind::Left && side == nullableSide && !thisType->IsOptional()) {
+                userTypes.SelectSide(side).push_back(TOptionalType::Create(thisType, env));
+            } else {
+                userTypes.SelectSide(side).push_back(thisType);
+            }
+        }
+    }
+    return userTypes;
+}
+
+template <EJoinKind Kind, typename Converter>
+struct TPackedTupleOutputBase : NNonCopyable::TMoveOnly {
+    struct Empty {};
+    using BuildNullIfNeeded = std::conditional_t<Kind == EJoinKind::Left, TPackResult, Empty>;
+
+    TPackedTupleOutputBase(const TDqRenames<ESide>* renames, TSides<Converter*> converters, bool leftIsBuild)
+        : Renames_(renames)
+        , Converters_(converters)
+        , LeftIsBuild_(leftIsBuild)
+    {}
+
+    int Columns() const {
+        return Renames_->size();
+    }
+
+    i64 SizeTuples() const {
+        AssertSizeIsSane();
+        return Output_.Probe.NTuples;
+    }
+
+    auto MakeConsumeFn() {
+        struct ConsumeFn {
+            TPackedTupleOutputBase& Self;
+
+            void operator()(TSides<TSingleTuple> tuples) {
+                for (ESide side : EachSide) {
+                    Self.Output_.SelectSide(side).AppendTuple(tuples.SelectSide(side),
+                                                              Self.Converters_.SelectSide(side)->GetTupleLayout());
+                }
+            }
+
+            void operator()(TSingleTuple tuple) {
+                if constexpr (Kind == EJoinKind::Left) {
+                    const TSingleTuple null{.PackedData = Self.Nulls_.PackedTuples.data(),
+                                            .OverflowBegin = Self.Nulls_.Overflow.data()};
+                    if (Self.LeftIsBuild_) {
+                        (*this)(TSides<TSingleTuple>{.Build = tuple, .Probe = null});
+                    } else {
+                        (*this)(TSides<TSingleTuple>{.Build = null, .Probe = tuple});
+                    }
+                } else if constexpr (SemiOrOnlyJoin(Kind)) {
+                    Self.Output_.Probe.AppendTuple(tuple, Self.Converters_.Probe->GetTupleLayout());
+                }
+            }
+        };
+        return ConsumeFn{*this};
+    }
+
+protected:
+    void AssertSizeIsSane() const {
+        if constexpr (LeftSemiOrOnly(Kind)) {
+            MKQL_ENSURE(Output_.Build.NTuples == 0,
+                        "Left Only and Left Semi join types shouldn't collect any Build(right) tuples");
+        } else if constexpr (Kind == EJoinKind::Left || Kind == EJoinKind::Inner) {
+            MKQL_ENSURE(Output_.Build.NTuples == Output_.Probe.NTuples,
+                        "Inner and Left join types must collect same amount of tuples from build and probe");
+        }
+    }
+
+    const TDqRenames<ESide>* Renames_;
+    TSides<Converter*> Converters_;
+    TSides<TPackResult> Output_;
+    BuildNullIfNeeded Nulls_;
+    bool LeftIsBuild_;
+};
+
+template <i64 MaxOutputRows, typename JoinType, typename OutputType, typename FlushSink>
+EFetchResult RunPackedHashJoinBatch(TComputationContext& ctx, JoinType& join, OutputType& output, FlushSink&& onFlush) {
+    auto outputIsFull = [&]() { return output.SizeTuples() >= MaxOutputRows; };
+    while (!outputIsFull()) {
+        switch (join.MatchRows(ctx, output.MakeConsumeFn(), outputIsFull)) {
+        case EFetchResult::Finish:
+            if (output.SizeTuples() == 0) {
+                return EFetchResult::Finish;
+            }
+            onFlush(output.Flush());
+            return EFetchResult::One;
+        case EFetchResult::Yield:
+            if constexpr (JoinType::FlushOnYield) {
+                if (output.SizeTuples() > 0) {
+                    onFlush(output.Flush());
+                    return EFetchResult::One;
+                }
+            }
+            return EFetchResult::Yield;
+        case EFetchResult::One:
+            break;
+        default:
+            MKQL_ENSURE(false, "unexpected fetch result");
+        }
+    }
+    onFlush(output.Flush());
+    return EFetchResult::One;
+}
+
 } // namespace NKikimr::NMiniKQL
diff --git a/ydb/library/yql/dq/comp_nodes/dq_scalar_hash_join.cpp b/ydb/library/yql/dq/comp_nodes/dq_scalar_hash_join.cpp
index 47c8a30ce26..34b17632ba6 100644
--- a/ydb/library/yql/dq/comp_nodes/dq_scalar_hash_join.cpp
+++ b/ydb/library/yql/dq/comp_nodes/dq_scalar_hash_join.cpp
@@ -10,9 +10,6 @@
 #include <ydb/library/yql/dq/comp_nodes/dq_join_common.h>
 #include <ydb/library/yql/dq/comp_nodes/hash_join_utils/scalar_layout_converter.h>
 
-#include <algorithm>
-#include <numeric>
-
 namespace NKikimr::NMiniKQL {
 
 namespace {
@@ -115,72 +112,38 @@ private:
 };
 
 template <EJoinKind Kind>
-struct TRenamesScalarOutput : NNonCopyable::TMoveOnly {
+struct TRenamesScalarOutput : TPackedTupleOutputBase<Kind, IScalarLayoutConverter> {
+    using TBase = TPackedTupleOutputBase<Kind, IScalarLayoutConverter>;
+
     struct TFlushResult {
         TVector<NUdf::TUnboxedValue> Buffer;
         TSides<TPackResult> Packs;
     };
 
-    struct Empty {};
-    using BuildNullIfNeeded = std::conditional_t<Kind == EJoinKind::Left, TPackResult, Empty>;
-
     TRenamesScalarOutput(const TDqScalarJoinMetadata* meta, TSides<IScalarLayoutConverter*> converters)
-        : Renames_(&meta->Renames)
-        , Converters_(converters)
+        : TBase(&meta->Renames, converters, /* leftIsBuild */ false)
         , BuildWidth_(std::ssize(meta->InputTypes.Build))
         , ProbeWidth_(std::ssize(meta->InputTypes.Probe))
     {
-        if constexpr (!std::is_same_v<decltype(Nulls_), Empty>) {
+        if constexpr (!std::is_same_v<typename TBase::BuildNullIfNeeded, typename TBase::Empty>) {
             TMKQLVector<NUdf::TUnboxedValue> nulls(BuildWidth_);
-            Converters_.Build->Pack(nulls.data(), Nulls_);
+            this->Converters_.Build->Pack(nulls.data(), this->Nulls_);
         }
     }
 
-    int Columns() const {
-        return Renames_->size();
-    }
-
-    i64 SizeTuples() const {
-        return Output_.Probe.NTuples;
-    }
-
-    auto MakeConsumeFn() {
-        struct ConsumeFn {
-            TRenamesScalarOutput& Self;
-
-            void operator()(TSides<TSingleTuple> tuples) {
-                for (ESide side : EachSide) {
-                    Self.Output_.SelectSide(side).AppendTuple(
-                        tuples.SelectSide(side), Self.Converters_.SelectSide(side)->GetTupleLayout());
-                }
-            }
-
-            void operator()(TSingleTuple tuple) {
-                if constexpr (Kind == EJoinKind::Left) {
-                    TSingleTuple null{.PackedData = Self.Nulls_.PackedTuples.data(),
-                                      .OverflowBegin = Self.Nulls_.Overflow.data()};
-                    this->operator()(TSides<TSingleTuple>{.Build = null, .Probe = tuple});
-                } else if constexpr (SemiOrOnlyJoin(Kind)) {
-                    Self.Output_.Probe.AppendTuple(tuple, Self.Converters_.Probe->GetTupleLayout());
-                }
-            }
-        };
-        return ConsumeFn{*this};
-    }
-
     TFlushResult Flush() {
         TFlushResult res;
-        const i64 nItems = Output_.Probe.NTuples;
-        res.Packs.Build = std::move(Output_.Build);
-        res.Packs.Probe = std::move(Output_.Probe);
+        const i64 nItems = this->Output_.Probe.NTuples;
+        res.Packs.Build = std::move(this->Output_.Build);
+        res.Packs.Probe = std::move(this->Output_.Probe);
 
-        res.Buffer.reserve(nItems * Columns());
+        res.Buffer.reserve(nItems * this->Columns());
 
         if constexpr (LeftSemiOrOnly(Kind)) {
             TMKQLVector<NUdf::TUnboxedValue> probeValues(ProbeWidth_);
             for (i64 tupleIndex = 0; tupleIndex < nItems; ++tupleIndex) {
-                Converters_.Probe->Unpack(res.Packs.Probe, tupleIndex, probeValues.data());
-                for (auto rename : *Renames_) {
+                this->Converters_.Probe->Unpack(res.Packs.Probe, tupleIndex, probeValues.data());
+                for (auto rename : *this->Renames_) {
                     MKQL_ENSURE(rename.Side == ESide::Probe,
                                 "renames in Semi or Only Left Join shouldn't contain columns from right side");
                     res.Buffer.push_back(probeValues[rename.Index]);
@@ -190,9 +153,9 @@ struct TRenamesScalarOutput : NNonCopyable::TMoveOnly {
             TMKQLVector<NUdf::TUnboxedValue> buildValues(BuildWidth_);
             TMKQLVector<NUdf::TUnboxedValue> probeValues(ProbeWidth_);
             for (i64 tupleIndex = 0; tupleIndex < nItems; ++tupleIndex) {
-                Converters_.Build->Unpack(res.Packs.Build, tupleIndex, buildValues.data());
-                Converters_.Probe->Unpack(res.Packs.Probe, tupleIndex, probeValues.data());
-                for (auto rename : *Renames_) {
+                this->Converters_.Build->Unpack(res.Packs.Build, tupleIndex, buildValues.data());
+                this->Converters_.Probe->Unpack(res.Packs.Probe, tupleIndex, probeValues.data());
+                for (auto rename : *this->Renames_) {
                     if (rename.Side == ESide::Build) {
                         res.Buffer.push_back(buildValues[rename.Index]);
                     } else {
@@ -206,12 +169,8 @@ struct TRenamesScalarOutput : NNonCopyable::TMoveOnly {
     }
 
 private:
-    const TDqJoinImplRenames* Renames_;
-    TSides<IScalarLayoutConverter*> Converters_;
     const int BuildWidth_;
     const int ProbeWidth_;
-    TSides<TPackResult> Output_;
-    BuildNullIfNeeded Nulls_;
 };
 
 template <EJoinKind Kind>
@@ -291,35 +250,8 @@ private:
         }
 
         EFetchResult FillBuffer() {
-            auto outputIsFull = [&]() {
-                return Output_.SizeTuples() >= Threshold_;
-            };
-            while (!outputIsFull()) {
-                auto res = Join_.MatchRows(*JoinCtx_, Output_.MakeConsumeFn(), outputIsFull);
-                switch (res) {
-                case EFetchResult::Finish: {
-                    if (Output_.SizeTuples() == 0) {
-                        return EFetchResult::Finish;
-                    }
-                    Buffer_ = Output_.Flush();
-                    return EFetchResult::One;
-                }
-                case EFetchResult::Yield: {
-                    if (Output_.SizeTuples() == 0) {
-                        return EFetchResult::Yield;
-                    }
-                    Buffer_ = Output_.Flush();
-                    return EFetchResult::One;
-                }
-                case EFetchResult::One: {
-                    break;
-                }
-                default:
-                    MKQL_ENSURE(false, "unexpected fetch result");
-                }
-            }
-            Buffer_ = Output_.Flush();
-            return EFetchResult::One;
+            return RunPackedHashJoinBatch<OutputThreshold_>(
+                *JoinCtx_, Join_, Output_, [&](auto flush) { Buffer_ = std::move(flush); });
         }
 
     private:
@@ -330,18 +262,15 @@ private:
         TRenamesScalarOutput<Kind> Output_;
         std::optional<typename TRenamesScalarOutput<Kind>::TFlushResult> Buffer_;
         size_t BufferPos_ = 0;
-        const int Threshold_ = 10000;
+        static constexpr i64 OutputThreshold_ = 10000;
     };
 
     void MakeState(TComputationContext& ctx, NUdf::TUnboxedValue& state) const {
         TSides<std::unique_ptr<IScalarLayoutConverter>> converters;
         TTypeInfoHelper helper;
         for(ESide side: EachSide) {
-            TVector<NPackedTuple::EColumnRole> roles(std::ssize(Meta_->UserTypes.SelectSide(side)),
-                                                    NPackedTuple::EColumnRole::Payload);
-            for (int column : Meta_->KeyColumns.SelectSide(side)) {
-                roles[column] = NPackedTuple::EColumnRole::Key;
-            }
+            const auto roles =
+                MakeColumnRoles(Meta_->UserTypes.SelectSide(side).size(), Meta_->KeyColumns.SelectSide(side));
             converters.SelectSide(side) =
                 MakeScalarLayoutConverter(helper, Meta_->UserTypes.SelectSide(side), roles, ctx.HolderFactory);
         }
@@ -393,116 +322,27 @@ IComputationWideFlowNode* WrapDqScalarHashJoin(TCallable& callable, const TCompu
         meta.InputTypes.Build.push_back(type);
     }
 
-    const auto joinKindNode = callable.GetInput(2);
-    const auto rawKind = AS_VALUE(TDataLiteral, joinKindNode)->AsValue().Get<ui32>();
-    const auto joinKind = GetJoinKind(rawKind);
+    const auto parsed = ParseCommonHashJoinArgs(callable);
+    const auto joinKind = parsed.Kind;
     meta.Kind = joinKind;
-
-    const auto leftKeyColumnsLiteral = callable.GetInput(3);
-    const auto leftKeyColumnsTuple = AS_VALUE(TTupleLiteral, leftKeyColumnsLiteral);
-    for (ui32 i = 0; i < leftKeyColumnsTuple->GetValuesCount(); i++) {
-        const auto item = AS_VALUE(TDataLiteral, leftKeyColumnsTuple->GetValue(i));
-        meta.KeyColumns.Probe.emplace_back(item->AsValue().Get<ui32>());
-    }
-
-    const auto rightKeyColumnsLiteral = callable.GetInput(4);
-    const auto rightKeyColumnsTuple = AS_VALUE(TTupleLiteral, rightKeyColumnsLiteral);
-    for (ui32 i = 0; i < rightKeyColumnsTuple->GetValuesCount(); i++) {
-        const auto item = AS_VALUE(TDataLiteral, rightKeyColumnsTuple->GetValue(i));
-        meta.KeyColumns.Build.emplace_back(item->AsValue().Get<ui32>());
-    }
-
-    MKQL_ENSURE(meta.KeyColumns.Build.size() == meta.KeyColumns.Probe.size(), "Key columns mismatch");
+    meta.KeyColumns = parsed.KeyColumns;
 
     const auto leftFlow = dynamic_cast<IComputationWideFlowNode*>(LocateNode(ctx.NodeLocator, callable, 0));
     const auto rightFlow = dynamic_cast<IComputationWideFlowNode*>(LocateNode(ctx.NodeLocator, callable, 1));
     MKQL_ENSURE(leftFlow, "Expected WideFlow as a left input");
     MKQL_ENSURE(rightFlow, "Expected WideFlow as a right input");
 
-    TDqUserRenames userRenames =
-        FromGraceFormat(TGraceJoinRenames::FromRuntimeNodes(callable.GetInput(5), callable.GetInput(6)));
-    ValidateRenames(userRenames, joinKind, std::ssize(meta.InputTypes.Probe), std::ssize(meta.InputTypes.Build));
-
-    for (auto rename : userRenames) {
-        ESide side = rename.Side == EJoinSide::kLeft ? ESide::Probe : ESide::Build;
-        meta.Renames.push_back({.Index = rename.Index, .Side = side});
-    }
-
-    for (ESide side : EachSide) {
-        auto& keyColumns = meta.KeyColumns.SelectSide(side);
-        const int numDataCols = std::ssize(meta.InputTypes.SelectSide(side));
-        const int numKeys = std::ssize(keyColumns);
-
-        bool needsReorder = false;
-        for (int i = 0; i < numKeys; ++i) {
-            if (keyColumns[i] != static_cast<ui32>(i)) {
-                needsReorder = true;
-                break;
-            }
-        }
-        if (!needsReorder) {
-            continue;
-        }
-
-        TVector<int> perm(numDataCols);
-        std::iota(perm.begin(), perm.end(), 0);
-        for (int i = 0; i < numKeys; ++i) {
-            const int keyColumn = static_cast<int>(keyColumns[i]);
-            MKQL_ENSURE(keyColumn >= 0 && keyColumn < numDataCols,
-                        Sprintf("key column index %i on %s side is out of range [0, %i)", keyColumn,
-                                AsString(side), numDataCols));
-            auto it = std::find(perm.begin() + i, perm.end(), keyColumn);
-            MKQL_ENSURE(it != perm.end(),
-                        Sprintf("key column index %i on %s side is duplicated or could not be placed",
-                                keyColumn, AsString(side)));
-            std::swap(perm[i], *it);
-        }
-
-        meta.ColumnPermutation.SelectSide(side) = perm;
+    ValidateRenames(parsed.UserRenames, joinKind, std::ssize(meta.InputTypes.Probe), std::ssize(meta.InputTypes.Build));
+    meta.Renames = BuildImplRenames(parsed.UserRenames);
 
-        const auto origTypes = meta.InputTypes.SelectSide(side);
-        for (int i = 0; i < numDataCols; ++i) {
-            meta.InputTypes.SelectSide(side)[i] = origTypes[perm[i]];
-        }
+    ApplyKeyColumnPermutation(meta.KeyColumns, meta.InputTypes, /* trailingColumns */ 0, meta.Renames,
+                              meta.ColumnPermutation);
+    meta.UserTypes = ForceOptionalOnNullableSide(meta.InputTypes, joinKind, ESide::Build, ctx.Env);
 
-        TVector<int> inv(numDataCols);
-        for (int i = 0; i < numDataCols; ++i) {
-            inv[perm[i]] = i;
-        }
-        for (auto& rename : meta.Renames) {
-            if (rename.Side == side) {
-                rename.Index = inv[rename.Index];
-            }
-        }
-
-        for (int i = 0; i < numKeys; ++i) {
-            keyColumns[i] = i;
-        }
-    }
-
-    for (ESide side : EachSide) {
-        for (TType* thisType : meta.InputTypes.SelectSide(side)) {
-            if (joinKind == EJoinKind::Left && side == ESide::Build && !thisType->IsOptional()) {
-                meta.UserTypes.SelectSide(side).push_back(TOptionalType::Create(thisType, ctx.Env));
-            } else {
-                meta.UserTypes.SelectSide(side).push_back(thisType);
-            }
-        }
-    }
-
-    using enum EJoinKind;
     const TSides<IComputationWideFlowNode*> flows{.Build = rightFlow, .Probe = leftFlow};
-    if (joinKind == Inner) {
-        return new TScalarHashJoinWrapper<Inner>(ctx.Mutables, std::move(meta), flows);
-    } else if (joinKind == LeftOnly) {
-        return new TScalarHashJoinWrapper<LeftOnly>(ctx.Mutables, std::move(meta), flows);
-    } else if (joinKind == LeftSemi) {
-        return new TScalarHashJoinWrapper<LeftSemi>(ctx.Mutables, std::move(meta), flows);
-    } else if (joinKind == Left) {
-        return new TScalarHashJoinWrapper<Left>(ctx.Mutables, std::move(meta), flows);
-    } else {
-        MKQL_ENSURE(false, "unsupported join type in scalar hash join, see gh#26780 for details.");
-    }
+    return DispatchHashJoinByKind<TScalarHashJoinWrapper, IComputationWideFlowNode>(
+        joinKind, "unsupported join type in scalar hash join, see gh#26780 for details.", ctx.Mutables,
+        std::move(meta), flows);
 }
 
 } // namespace NKikimr::NMiniKQL