Add base class for simple stateless wide nodes (#5183)

4 files changed, 156 insertions, 77 deletions

diff --git a/ydb/library/yql/minikql/comp_nodes/mkql_skip.cpp b/ydb/library/yql/minikql/comp_nodes/mkql_skip.cpp
index b1bd436740b..411994de69f 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_skip.cpp
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_skip.cpp
@@ -118,8 +118,8 @@ private:
     IComputationNode* const Count;
 };
 
-class TWideSkipWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWrapper, ui64> {
-using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWrapper, ui64>;
+class TWideSkipWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWrapper> {
+using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWrapper>;
 public:
      TWideSkipWrapper(TComputationMutables& mutables, IComputationWideFlowNode* flow, IComputationNode* count, ui32 size)
         : TBaseComputation(mutables, flow, size, size)
diff --git a/ydb/library/yql/minikql/comp_nodes/mkql_wide_filter.cpp b/ydb/library/yql/minikql/comp_nodes/mkql_wide_filter.cpp
index d2d26230369..36dbcf37e2e 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_wide_filter.cpp
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_wide_filter.cpp
@@ -77,58 +77,57 @@ protected:
     const ui32 WideFieldsIndex;
 };
 
-class TWideFilterWrapper : public TStatelessWideFlowCodegeneratorNode<TWideFilterWrapper>,  public TBaseWideFilterWrapper {
-using TBaseComputation = TStatelessWideFlowCodegeneratorNode<TWideFilterWrapper>;
+class TWideFilterWrapper : public TSimpleStatelessWideFlowCodegeneratorNode<TWideFilterWrapper>,  public TBaseWideFilterWrapper {
+using TBaseComputation = TSimpleStatelessWideFlowCodegeneratorNode<TWideFilterWrapper>;
 public:
     TWideFilterWrapper(TComputationMutables& mutables, IComputationWideFlowNode* flow, TComputationExternalNodePtrVector&& items, IComputationNode* predicate)
-        : TBaseComputation(flow)
+        : TBaseComputation(flow, items.size(), items.size())
         , TBaseWideFilterWrapper(mutables, flow, std::move(items), predicate)
     {}
 
-    EFetchResult DoCalculate(TComputationContext& ctx, NUdf::TUnboxedValue*const* output) const {
-        auto** fields = GetFields(ctx);
-
-        while (true) {
-            PrepareArguments(ctx, output);
-
-            if (const auto result = Flow->FetchValues(ctx, fields); EFetchResult::One != result)
-                return result;
+    NUdf::TUnboxedValue*const* PrepareInput(TComputationContext& ctx, NUdf::TUnboxedValue*const* output) const {
+        return PrepareArguments(ctx, output);
+    }
 
+    TMaybeFetchResult DoProcess(TComputationContext& ctx, TMaybeFetchResult fetchRes, NUdf::TUnboxedValue*const* output) const {
+        if (fetchRes.Get() == EFetchResult::One) {
             if (Predicate->GetValue(ctx).Get<bool>()) {
                 FillOutputs(ctx, output);
                 return EFetchResult::One;
             }
+            return TMaybeFetchResult::None();
         }
+        return fetchRes;
     }
-#ifndef MKQL_DISABLE_CODEGEN
-    TGenerateResult DoGenGetValues(const TCodegenContext& ctx, BasicBlock*& block) const {
-        auto& context = ctx.Codegen.GetContext();
-
-        const auto loop = BasicBlock::Create(context, "loop", ctx.Func);
 
-        BranchInst::Create(loop, block);
-
-        block = loop;
-
-        auto status = GetNodeValues(Flow, ctx, block);
-
-        const auto good = CmpInst::Create(Instruction::ICmp, ICmpInst::ICMP_SGT, status.first, ConstantInt::get(status.first->getType(), 0), "good", block);
-
-        const auto work = BasicBlock::Create(context, "work", ctx.Func);
-        const auto pass = BasicBlock::Create(context, "pass", ctx.Func);
-
-        BranchInst::Create(work, pass, good, block);
-
-        block = work;
+#ifndef MKQL_DISABLE_CODEGEN
+    typename TBaseComputation::TGenerateResult GenFetchProcess(const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const override {
+        auto &context = ctx.Codegen.GetContext();
+        auto pass = BasicBlock::Create(context, "pass", ctx.Func);
+        auto check = BasicBlock::Create(context, "check", ctx.Func);
+        auto decr = BasicBlock::Create(context, "decr", ctx.Func);
+        auto maybeResultVal = PHINode::Create(TMaybeFetchResult::LLVMType(context), 4, "maybe_res", pass);
+        
+        auto [fetchResVal, fetchGetters] = fetchGenerator(ctx, block);
+        auto passCond = CmpInst::Create(Instruction::ICmp, ICmpInst::ICMP_NE, ConstantInt::get(fetchResVal->getType(), static_cast<i32>(EFetchResult::One)), fetchResVal, "not_one", block);
+        maybeResultVal->addIncoming(TMaybeFetchResult::LLVMFromFetchResult(fetchResVal, "fetch_res_ext", block), block);
+        BranchInst::Create(pass, check, passCond, block);
 
-        const auto predicate = GenGetPredicate(ctx, status.second, block);
+        block = check;
+        auto predicateCond = GenGetPredicate<false>(ctx, fetchGetters, block);
+        maybeResultVal->addIncoming(TMaybeFetchResult::None().LLVMConst(context), block);
+        BranchInst::Create(decr, pass, predicateCond, block);
 
-        BranchInst::Create(pass, loop, predicate, block);
+        block = decr;
+        maybeResultVal->addIncoming(TMaybeFetchResult(EFetchResult::One).LLVMConst(context), block);
+        BranchInst::Create(pass, block);
 
         block = pass;
-        return status;
+
+        return {maybeResultVal, std::move(fetchGetters)};
     }
 #endif
+
 private:
     void RegisterDependencies() const final {
         if (const auto flow = FlowDependsOn(Flow)) {
@@ -138,8 +137,8 @@ private:
     }
 };
 
-class TWideFilterWithLimitWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideFilterWithLimitWrapper, ui64>,  public TBaseWideFilterWrapper {
-using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideFilterWithLimitWrapper, ui64>;
+class TWideFilterWithLimitWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideFilterWithLimitWrapper>,  public TBaseWideFilterWrapper {
+using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideFilterWithLimitWrapper>;
 public:
     TWideFilterWithLimitWrapper(TComputationMutables& mutables, IComputationWideFlowNode* flow, IComputationNode* limit,
             TComputationExternalNodePtrVector&& items, IComputationNode* predicate)
@@ -220,8 +219,8 @@ private:
 };
 
 template<bool Inclusive>
-class TWideTakeWhileWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideTakeWhileWrapper<Inclusive>, bool>,  public TBaseWideFilterWrapper {
-using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideTakeWhileWrapper<Inclusive>, bool>;
+class TWideTakeWhileWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideTakeWhileWrapper<Inclusive>>,  public TBaseWideFilterWrapper {
+using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideTakeWhileWrapper<Inclusive>>;
 public:
      TWideTakeWhileWrapper(TComputationMutables& mutables, IComputationWideFlowNode* flow, TComputationExternalNodePtrVector&& items,
             IComputationNode* predicate)
@@ -298,8 +297,8 @@ private:
 };
 
 template<bool Inclusive>
-class TWideSkipWhileWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWhileWrapper<Inclusive>, bool>,  public TBaseWideFilterWrapper {
-using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWhileWrapper<Inclusive>, bool>;
+class TWideSkipWhileWrapper : public TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWhileWrapper<Inclusive>>,  public TBaseWideFilterWrapper {
+using TBaseComputation = TSimpleStatefulWideFlowCodegeneratorNode<TWideSkipWhileWrapper<Inclusive>>;
 public:
      TWideSkipWhileWrapper(TComputationMutables& mutables, IComputationWideFlowNode* flow, TComputationExternalNodePtrVector&& items, IComputationNode* predicate)
         : TBaseComputation(mutables, flow, items.size(), items.size())
diff --git a/ydb/library/yql/minikql/computation/mkql_simple_codegen.cpp b/ydb/library/yql/minikql/computation/mkql_simple_codegen.cpp
index 4829d23ac22..28558660e77 100644
--- a/ydb/library/yql/minikql/computation/mkql_simple_codegen.cpp
+++ b/ydb/library/yql/minikql/computation/mkql_simple_codegen.cpp
@@ -5,7 +5,7 @@ namespace NKikimr {
 namespace NMiniKQL {
 
 #ifndef MKQL_DISABLE_CODEGEN
-ICodegeneratorInlineWideNode::TGenerateResult TSimpleStatefulWideFlowCodegeneratorNodeLLVMBase::DoGenGetValues(const NKikimr::NMiniKQL::TCodegenContext &ctx, llvm::Value *statePtrVal, llvm::BasicBlock *&genToBlock) const  {
+ICodegeneratorInlineWideNode::TGenerateResult TSimpleWideFlowCodegeneratorNodeLLVMBase::DoGenGetValuesBase(const NKikimr::NMiniKQL::TCodegenContext& ctx, llvm::Value* statePtrVal, llvm::BasicBlock*& genToBlock) const  {
     // init stuff (mainly in global entry block)
 
     auto& context = ctx.Codegen.GetContext();
@@ -19,12 +19,14 @@ ICodegeneratorInlineWideNode::TGenerateResult TSimpleStatefulWideFlowCodegenerat
     const auto done = BasicBlock::Create(context, "done", ctx.Func);
     const auto entryPos = &ctx.Func->getEntryBlock().back();
 
+    const bool hasState = statePtrVal != nullptr;
+
     const auto thisType = StructType::get(context)->getPointerTo();
     const auto thisRawVal = ConstantInt::get(Type::getInt64Ty(context), PtrTable.ThisPtr);
     const auto thisVal = CastInst::Create(Instruction::IntToPtr, thisRawVal, thisType, "this", entryPos);
     const auto valuePtrType = PointerType::getUnqual(valueType);
     const auto valuePtrsPtrType = PointerType::getUnqual(valuePtrType);
-    const auto statePtrType = statePtrVal->getType();
+    const auto statePtrType = hasState ? statePtrVal->getType() : nullptr;
     const auto ctxType = ctx.Ctx->getType();
     const auto i32Type = Type::getInt32Ty(context);
     const auto valueNullptrVal = ConstantPointerNull::get(valuePtrType);
@@ -44,26 +46,32 @@ ICodegeneratorInlineWideNode::TGenerateResult TSimpleStatefulWideFlowCodegenerat
 
     auto block = genToBlock; // >>> start of main code chunk
 
-    const auto stateVal = new LoadInst(valueType, statePtrVal, "state", block);
-    BranchInst::Create(init, loop, IsInvalid(stateVal, block), block);
+    const auto stateVal = hasState ? new LoadInst(valueType, statePtrVal, "state", block) : nullptr;
+    BranchInst::Create(init, loop, hasState ? IsInvalid(stateVal, block) : ConstantInt::get(Type::getInt1Ty(context), 0), block);
 
     block = init; // state initialization block:
 
-    const auto initFuncType = FunctionType::get(Type::getVoidTy(context), {thisType, statePtrType, ctxType}, false);
-    const auto initFuncRawVal = ConstantInt::get(Type::getInt64Ty(context), PtrTable.InitStateMethPtr);
-    const auto initFuncVal = CastInst::Create(Instruction::IntToPtr, initFuncRawVal, PointerType::getUnqual(initFuncType), "init_func", block);
-    CallInst::Create(initFuncType, initFuncVal, {thisVal, statePtrVal, ctx.Ctx}, "", block);
+    if (hasState) {
+        const auto initFuncType = FunctionType::get(Type::getVoidTy(context), {thisType, statePtrType, ctxType}, false);
+        const auto initFuncRawVal = ConstantInt::get(Type::getInt64Ty(context), PtrTable.InitStateMethPtr);
+        const auto initFuncVal = CastInst::Create(Instruction::IntToPtr, initFuncRawVal, PointerType::getUnqual(initFuncType), "init_func", block);
+        CallInst::Create(initFuncType, initFuncVal, {thisVal, statePtrVal, ctx.Ctx}, "", block);
+    }
     BranchInst::Create(loop, block);
 
     block = loop; // loop head block: (prepare inputs and decide whether to calculate row or not)
 
-    const auto generated = GenFetchProcess(statePtrVal, ctx, std::bind_front(GetNodeValues, SourceFlow), block);
+    const auto generated = DispatchGenFetchProcess(statePtrVal, ctx, std::bind_front(GetNodeValues, SourceFlow), block);
     auto processResVal = generated.first;
     if (processResVal == nullptr) {
-        const auto prepareFuncType = FunctionType::get(valuePtrsPtrType, {thisType, statePtrType, ctxType, valuePtrsPtrType}, false);
+        const auto prepareFuncType = hasState
+            ? FunctionType::get(valuePtrsPtrType, {thisType, statePtrType, ctxType, valuePtrsPtrType}, false)
+            : FunctionType::get(valuePtrsPtrType, {thisType, ctxType, valuePtrsPtrType}, false);
         const auto prepareFuncRawVal = ConstantInt::get(Type::getInt64Ty(context), PtrTable.PrepareInputMethPtr);
         const auto prepareFuncVal = CastInst::Create(Instruction::IntToPtr, prepareFuncRawVal, PointerType::getUnqual(prepareFuncType), "prepare_func", block);
-        const auto inputPtrsVal = CallInst::Create(prepareFuncType, prepareFuncVal, {thisVal, statePtrVal, ctx.Ctx, outputPtrsVal}, "input_ptrs", block);
+        const auto inputPtrsVal = hasState
+            ? CallInst::Create(prepareFuncType, prepareFuncVal, {thisVal, statePtrVal, ctx.Ctx, outputPtrsVal}, "input_ptrs", block)
+            : CallInst::Create(prepareFuncType, prepareFuncVal, {thisVal, ctx.Ctx, outputPtrsVal}, "input_ptrs", block);
         const auto skipFetchCond = CmpInst::Create(Instruction::ICmp, ICmpInst::ICMP_EQ, inputPtrsVal, valuePtrNullptrVal, "skip_fetch", block);
         BranchInst::Create(loopTail, loopFetch, skipFetchCond, block);
 
@@ -109,10 +117,14 @@ ICodegeneratorInlineWideNode::TGenerateResult TSimpleStatefulWideFlowCodegenerat
         maybeFetchResVal->addIncoming(noneVal, loop);
         maybeFetchResVal->addIncoming(fetchResExtVal, fetchSourceBlock);
         maybeFetchResVal->addIncoming(fetchResExtVal, calcSourceBlock);
-        const auto processFuncType = FunctionType::get(maybeResType, {thisType, statePtrType, ctxType, maybeResType, valuePtrsPtrType}, false);
+        const auto processFuncType = hasState 
+            ? FunctionType::get(maybeResType, {thisType, statePtrType, ctxType, maybeResType, valuePtrsPtrType}, false)
+            : FunctionType::get(maybeResType, {thisType, ctxType, maybeResType, valuePtrsPtrType}, false);
         const auto processFuncRawVal = ConstantInt::get(Type::getInt64Ty(context), PtrTable.DoProcessMethPtr);
         const auto processFuncVal = CastInst::Create(Instruction::IntToPtr, processFuncRawVal, PointerType::getUnqual(processFuncType), "process_func", block);
-        processResVal = CallInst::Create(processFuncType, processFuncVal, {thisVal, statePtrVal, ctx.Ctx, maybeFetchResVal, outputPtrsVal}, "process_res", block);
+        processResVal = hasState
+            ? CallInst::Create(processFuncType, processFuncVal, {thisVal, statePtrVal, ctx.Ctx, maybeFetchResVal, outputPtrsVal}, "process_res", block)
+            : CallInst::Create(processFuncType, processFuncVal, {thisVal, ctx.Ctx, maybeFetchResVal, outputPtrsVal}, "process_res", block);
     } else {
         BranchInst::Create(loopFetch, loopFetch);
         BranchInst::Create(loopCalc, loopCalc);
@@ -143,6 +155,7 @@ ICodegeneratorInlineWideNode::TGenerateResult TSimpleStatefulWideFlowCodegenerat
     }
     return {processResTruncVal, std::move(new_getters)};
 }
+
 #endif
 
 }
diff --git a/ydb/library/yql/minikql/computation/mkql_simple_codegen.h b/ydb/library/yql/minikql/computation/mkql_simple_codegen.h
index b87509006eb..977ad2a99b5 100644
--- a/ydb/library/yql/minikql/computation/mkql_simple_codegen.h
+++ b/ydb/library/yql/minikql/computation/mkql_simple_codegen.h
@@ -34,7 +34,7 @@ public:
         return Type::getInt64Ty(context);
     }
 
-    static Value* LLVMFromFetchResult(Value *fetchRes, const Twine& name, BasicBlock* block) {
+    static Value* LLVMFromFetchResult(Value* fetchRes, const Twine& name, BasicBlock* block) {
         return new ZExtInst(fetchRes, LLVMType(fetchRes->getContext()), name, block);
     }
 
@@ -48,7 +48,7 @@ public:
 using TResultCodegenerator = std::function<ICodegeneratorInlineWideNode::TGenerateResult(const TCodegenContext&, BasicBlock*&)>;
 #endif
 
-class TSimpleStatefulWideFlowCodegeneratorNodeLLVMBase {
+class TSimpleWideFlowCodegeneratorNodeLLVMBase {
 public:
     struct TMethPtrTable {
         uintptr_t ThisPtr;
@@ -57,44 +57,52 @@ public:
         uintptr_t DoProcessMethPtr;
     };
 
-    TSimpleStatefulWideFlowCodegeneratorNodeLLVMBase(IComputationWideFlowNode* source, ui32 inWidth, ui32 outWidth, TMethPtrTable ptrTable)
+    TSimpleWideFlowCodegeneratorNodeLLVMBase(IComputationWideFlowNode* source, ui32 inWidth, ui32 outWidth, TMethPtrTable ptrTable)
         : SourceFlow(source), InWidth(inWidth), OutWidth(outWidth)
         , PtrTable(ptrTable) {}
 
+protected:
 #ifndef MKQL_DISABLE_CODEGEN
-    virtual ICodegeneratorInlineWideNode::TGenerateResult GenFetchProcess(Value* statePtrVal, const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const {
-        Y_UNUSED(statePtrVal);
-        Y_UNUSED(ctx);
-        Y_UNUSED(fetchGenerator);
-        Y_UNUSED(block);
-        return {nullptr, {}};
-    }
+    virtual ICodegeneratorInlineWideNode::TGenerateResult DispatchGenFetchProcess(Value* statePtrVal, const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const = 0;
 
-    ICodegeneratorInlineWideNode::TGenerateResult DoGenGetValues(const TCodegenContext& ctx, Value* statePtrVal, BasicBlock*& genToBlock) const;
+    ICodegeneratorInlineWideNode::TGenerateResult DoGenGetValuesBase(const NKikimr::NMiniKQL::TCodegenContext& ctx, llvm::Value* statePtrVal, llvm::BasicBlock*& genToBlock) const;
 #endif
 
-protected:
     IComputationWideFlowNode* const SourceFlow;
     const ui32 InWidth, OutWidth;
     const TMethPtrTable PtrTable;
 };
 
-template<typename TDerived, typename TState, EValueRepresentation StateKind = EValueRepresentation::Embedded>
+template<typename TDerived, EValueRepresentation StateKind = EValueRepresentation::Embedded>
 class TSimpleStatefulWideFlowCodegeneratorNode
-        : public TStatefulWideFlowCodegeneratorNode<TSimpleStatefulWideFlowCodegeneratorNode<TDerived, TState, StateKind>>
-        , public TSimpleStatefulWideFlowCodegeneratorNodeLLVMBase {
+        : public TStatefulWideFlowCodegeneratorNode<TSimpleStatefulWideFlowCodegeneratorNode<TDerived, StateKind>>
+        , public TSimpleWideFlowCodegeneratorNodeLLVMBase {
     using TBase = TStatefulWideFlowCodegeneratorNode<TSimpleStatefulWideFlowCodegeneratorNode>;
-    using TLLVMBase = TSimpleStatefulWideFlowCodegeneratorNodeLLVMBase;
+    using TLLVMBase = TSimpleWideFlowCodegeneratorNodeLLVMBase;
 
 protected:
     TSimpleStatefulWideFlowCodegeneratorNode(TComputationMutables& mutables, IComputationWideFlowNode* source, ui32 inWidth, ui32 outWidth)
-            : TBase(mutables, source, StateKind)
-            , TLLVMBase(source, inWidth, outWidth, {
-                .ThisPtr = reinterpret_cast<uintptr_t>(this),
-                .InitStateMethPtr = GetMethodPtr(&TDerived::InitState),
-                .PrepareInputMethPtr = GetMethodPtr(&TDerived::PrepareInput),
-                .DoProcessMethPtr = GetMethodPtr(&TDerived::DoProcess)
-            }) {}
+        : TBase(mutables, source, StateKind)
+        , TLLVMBase(source, inWidth, outWidth, {
+            .ThisPtr = reinterpret_cast<uintptr_t>(this),
+            .InitStateMethPtr = GetMethodPtr(&TDerived::InitState),
+            .PrepareInputMethPtr = GetMethodPtr(&TDerived::PrepareInput),
+            .DoProcessMethPtr = GetMethodPtr(&TDerived::DoProcess)
+        }) {}
+    
+#ifndef MKQL_DISABLE_CODEGEN
+    virtual ICodegeneratorInlineWideNode::TGenerateResult GenFetchProcess(Value* statePtrVal, const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const {
+        Y_UNUSED(statePtrVal);
+        Y_UNUSED(ctx);
+        Y_UNUSED(fetchGenerator);
+        Y_UNUSED(block);
+        return {nullptr, {}};
+    }
+
+    virtual ICodegeneratorInlineWideNode::TGenerateResult DispatchGenFetchProcess(Value* statePtrVal, const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const override final {
+        return GenFetchProcess(statePtrVal, ctx, fetchGenerator, block);
+    }
+#endif
 
 public:
     EFetchResult DoCalculate(NUdf::TUnboxedValue& state, TComputationContext& ctx, NUdf::TUnboxedValue*const* output) const {
@@ -114,6 +122,65 @@ public:
         }
         return result.Get();
     }
+
+#ifndef MKQL_DISABLE_CODEGEN
+    ICodegeneratorInlineWideNode::TGenerateResult DoGenGetValues(const NKikimr::NMiniKQL::TCodegenContext& ctx, llvm::Value* statePtrVal, llvm::BasicBlock*& genToBlock) const {
+        return DoGenGetValuesBase(ctx, statePtrVal, genToBlock);
+    }
+#endif
+};
+
+template<typename TDerived>
+class TSimpleStatelessWideFlowCodegeneratorNode
+        : public TStatelessWideFlowCodegeneratorNode<TSimpleStatelessWideFlowCodegeneratorNode<TDerived>> 
+        , public TSimpleWideFlowCodegeneratorNodeLLVMBase {
+    using TBase = TStatelessWideFlowCodegeneratorNode<TSimpleStatelessWideFlowCodegeneratorNode<TDerived>>;
+    using TLLVMBase = TSimpleWideFlowCodegeneratorNodeLLVMBase;
+
+protected:
+    TSimpleStatelessWideFlowCodegeneratorNode(IComputationWideFlowNode* source, ui32 inWidth, ui32 outWidth)
+        : TBase (source)
+        , TLLVMBase(source, inWidth, outWidth, {
+            .ThisPtr = reinterpret_cast<uintptr_t>(this),
+            .InitStateMethPtr = 0,
+            .PrepareInputMethPtr = GetMethodPtr(&TDerived::PrepareInput),
+            .DoProcessMethPtr = GetMethodPtr(&TDerived::DoProcess)
+        }) {}
+
+#ifndef MKQL_DISABLE_CODEGEN
+    virtual ICodegeneratorInlineWideNode::TGenerateResult GenFetchProcess(const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const {
+        Y_UNUSED(ctx);
+        Y_UNUSED(fetchGenerator);
+        Y_UNUSED(block);
+        return {nullptr, {}};
+    }
+
+    virtual ICodegeneratorInlineWideNode::TGenerateResult DispatchGenFetchProcess(Value* statePtrVal, const TCodegenContext& ctx, const TResultCodegenerator& fetchGenerator, BasicBlock*& block) const override final {
+        Y_UNUSED(statePtrVal);
+        return GenFetchProcess(ctx, fetchGenerator, block);
+    }
+#endif
+
+public:
+    EFetchResult DoCalculate(TComputationContext& ctx, NUdf::TUnboxedValue*const* output) const {
+        NUdf::TUnboxedValue *const stub = nullptr;
+        if (!output && !OutWidth) {
+            output = &stub;
+        }
+        auto result = TMaybeFetchResult::None();
+        while (result.Empty()) {
+            NUdf::TUnboxedValue*const* input = static_cast<const TDerived*>(this)->PrepareInput(ctx, output);
+            TMaybeFetchResult fetchResult = input ? SourceFlow->FetchValues(ctx, input) : TMaybeFetchResult::None();
+            result = static_cast<const TDerived*>(this)->DoProcess(ctx, fetchResult, output);
+        }
+        return result.Get();
+    }
+
+#ifndef MKQL_DISABLE_CODEGEN
+    ICodegeneratorInlineWideNode::TGenerateResult DoGenGetValues(const NKikimr::NMiniKQL::TCodegenContext& ctx, llvm::BasicBlock*& genToBlock) const {
+        return DoGenGetValuesBase(ctx, nullptr, genToBlock);
+    }
+#endif
 };
 
 }