PR from branch users/spuchin/KIKIMR-14325

Support ExtractMembers in SqlIn optimizer. (KIKIMR-14325)
REVIEW: 2423187

Do not split reads on ExtractMembers. (KIKIMR-14325, KIKIMR-14498)
REVIEW: 2416041

REVIEW: 2426829
x-ydb-stable-ref: 920e89b18455b2437fb171af35e1ef0575108f50

14 files changed, 463 insertions, 230 deletions

diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log.cpp
index 4a2d137926..e696d5a50c 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log.cpp
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log.cpp
@@ -28,10 +28,6 @@ public:
         AddHandler(0, &TCoFlatMap::Match, HNDL(PushExtractedPredicateToReadTable));
         AddHandler(0, &TCoFlatMap::Match, HNDL(PushPredicateToReadTable));
         AddHandler(0, &TCoAggregate::Match, HNDL(RewriteAggregate));
-        AddHandler(0, &TCoExtractMembers::Match, HNDL(ApplyExtractMembersToReadTable));
-        AddHandler(0, &TCoExtractMembers::Match, HNDL(ApplyExtractMembersToReadTableRanges));
-        AddHandler(0, &TCoExtractMembers::Match, HNDL(ApplyExtractMembersToReadOlapTable));
-        AddHandler(0, &TCoExtractMembers::Match, HNDL(ApplyExtractMembersToLookupTable));
         AddHandler(0, &TCoTake::Match, HNDL(RewriteTakeSortToTopSort));
         AddHandler(0, &TCoFlatMap::Match, HNDL(RewriteSqlInToEquiJoin));
         AddHandler(0, &TCoFlatMap::Match, HNDL(RewriteSqlInCompactToJoin));
@@ -45,10 +41,21 @@ public:
         AddHandler(0, &TKqlDeleteRows::Match, HNDL(DeleteOverLookup));
         AddHandler(0, &TKqlUpsertRowsBase::Match, HNDL(ExcessUpsertInputColumns));
         AddHandler(0, &TCoTake::Match, HNDL(DropTakeOverLookupTable));
+        AddHandler(0, &TKqlReadTableBase::Match, HNDL(ApplyExtractMembersToReadTable<false>));
+        AddHandler(0, &TKqlReadTableRangesBase::Match, HNDL(ApplyExtractMembersToReadTableRanges<false>));
+        AddHandler(0, &TKqpReadOlapTableRangesBase::Match, HNDL(ApplyExtractMembersToReadOlapTable<false>));
+        AddHandler(0, &TKqlLookupTableBase::Match, HNDL(ApplyExtractMembersToLookupTable<false>));
 
         AddHandler(1, &TKqlReadTableIndex::Match, HNDL(RewriteIndexRead));
         AddHandler(1, &TKqlLookupIndex::Match, HNDL(RewriteLookupIndex));
+
+        AddHandler(2, &TKqlReadTableBase::Match, HNDL(ApplyExtractMembersToReadTable<true>));
+        AddHandler(2, &TKqlReadTableRangesBase::Match, HNDL(ApplyExtractMembersToReadTableRanges<true>));
+        AddHandler(2, &TKqpReadOlapTableRangesBase::Match, HNDL(ApplyExtractMembersToReadOlapTable<true>));
+        AddHandler(2, &TKqlLookupTableBase::Match, HNDL(ApplyExtractMembersToLookupTable<true>));
 #undef HNDL
+
+        SetGlobal(2u);
     }
 
 protected:
@@ -71,30 +78,6 @@ protected:
         return output;
     }
 
-    TMaybeNode<TExprBase> ApplyExtractMembersToReadTable(TExprBase node, TExprContext& ctx) {
-        TExprBase output = KqpApplyExtractMembersToReadTable(node, ctx);
-        DumpAppliedRule("ApplyExtractMembersToReadTable", node.Ptr(), output.Ptr(), ctx);
-        return output;
-    }
-
-    TMaybeNode<TExprBase> ApplyExtractMembersToReadTableRanges(TExprBase node, TExprContext& ctx) {
-        TExprBase output = KqpApplyExtractMembersToReadTableRanges(node, ctx);
-        DumpAppliedRule("ApplyExtractMembersToReadTableRanges", node.Ptr(), output.Ptr(), ctx);
-        return output;
-    }
-
-    TMaybeNode<TExprBase> ApplyExtractMembersToReadOlapTable(TExprBase node, TExprContext& ctx) {
-        TExprBase output = KqpApplyExtractMembersToReadOlapTable(node, ctx);
-        DumpAppliedRule("ApplyExtractMembersToReadOlapTable", node.Ptr(), output.Ptr(), ctx);
-        return output;
-    }
-
-    TMaybeNode<TExprBase> ApplyExtractMembersToLookupTable(TExprBase node, TExprContext& ctx) {
-        TExprBase output = KqpApplyExtractMembersToLookupTable(node, ctx);
-        DumpAppliedRule("ApplyExtractMembersToLookupTable", node.Ptr(), output.Ptr(), ctx);
-        return output;
-    }
-
     TMaybeNode<TExprBase> RewriteTakeSortToTopSort(TExprBase node, TExprContext& ctx, const TGetParents& getParents) {
         TExprBase output = DqRewriteTakeSortToTopSort(node, ctx, *getParents());
         DumpAppliedRule("RewriteTakeSortToTopSort", node.Ptr(), output.Ptr(), ctx);
@@ -179,6 +162,42 @@ protected:
         return output;
     }
 
+    template <bool IsGlobal>
+    TMaybeNode<TExprBase> ApplyExtractMembersToReadTable(TExprBase node, TExprContext& ctx,
+        const TGetParents& getParents)
+    {
+        TExprBase output = KqpApplyExtractMembersToReadTable(node, ctx, *getParents(), IsGlobal);
+        DumpAppliedRule("ApplyExtractMembersToReadTable", node.Ptr(), output.Ptr(), ctx);
+        return output;
+    }
+
+    template <bool IsGlobal>
+    TMaybeNode<TExprBase> ApplyExtractMembersToReadTableRanges(TExprBase node, TExprContext& ctx,
+        const TGetParents& getParents)
+    {
+        TExprBase output = KqpApplyExtractMembersToReadTableRanges(node, ctx, *getParents(), IsGlobal);
+        DumpAppliedRule("ApplyExtractMembersToReadTableRanges", node.Ptr(), output.Ptr(), ctx);
+        return output;
+    }
+
+    template <bool IsGlobal>
+    TMaybeNode<TExprBase> ApplyExtractMembersToReadOlapTable(TExprBase node, TExprContext& ctx,
+        const TGetParents& getParents)
+    {
+        TExprBase output = KqpApplyExtractMembersToReadOlapTable(node, ctx, *getParents(), IsGlobal);
+        DumpAppliedRule("ApplyExtractMembersToReadOlapTable", node.Ptr(), output.Ptr(), ctx);
+        return output;
+    }
+
+    template <bool IsGlobal>
+    TMaybeNode<TExprBase> ApplyExtractMembersToLookupTable(TExprBase node, TExprContext& ctx,
+        const TGetParents& getParents)
+    {
+        TExprBase output = KqpApplyExtractMembersToLookupTable(node, ctx, *getParents(), IsGlobal);
+        DumpAppliedRule("ApplyExtractMembersToLookupTable", node.Ptr(), output.Ptr(), ctx);
+        return output;
+    }
+
 private:
     TTypeAnnotationContext& TypesCtx;
     const TKqpOptimizeContext& KqpCtx;
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_extract.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log_extract.cpp
index fcfd8a9e92..24d63381c7 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log_extract.cpp
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_extract.cpp
@@ -15,123 +15,166 @@ using namespace NYql::NCommon;
 using namespace NYql::NDq;
 using namespace NYql::NNodes;
 
+namespace {
 
-TExprBase KqpApplyExtractMembersToReadTable(TExprBase node, TExprContext& ctx) {
-    if (!node.Maybe<TCoExtractMembers>()) {
-        return node;
+TMaybeNode<TCoAtomList> GetUsedColumns(TExprBase read, TCoAtomList columns, const TParentsMap& parentsMap,
+    bool allowMultiUsage, TExprContext& ctx)
+{
+    TSet<TString> usedColumnsSet;
+
+    auto consumers = GetConsumers(read, parentsMap);
+    if (!allowMultiUsage && consumers.size() > 1) {
+        return {};
+    }
+
+    for (const auto& consumer : consumers) {
+        auto maybeExtractMembers = TMaybeNode<TCoExtractMembers>(consumer);
+        if (!maybeExtractMembers) {
+            return {};
+        }
+
+        auto columns = maybeExtractMembers.Cast().Members();
+        for (const auto& column : columns) {
+            usedColumnsSet.emplace(column);
+        }
     }
 
-    auto extract = node.Cast<TCoExtractMembers>();
-    auto input = extract.Input();
+    YQL_ENSURE(usedColumnsSet.size() <= columns.Size());
+
+    if (usedColumnsSet.size() == columns.Size()) {
+        return {};
+    }
 
-    if (!input.Maybe<TKqlReadTableBase>()) {
+    TVector<TExprNode::TPtr> usedColumns;
+    usedColumns.reserve(usedColumnsSet.size());
+    for (const auto& column : usedColumnsSet) {
+        usedColumns.emplace_back(ctx.NewAtom(columns.Pos(), column));
+    }
+
+    return Build<TCoAtomList>(ctx, columns.Pos())
+        .Add(usedColumns)
+        .Done();
+}
+
+} // namespace
+
+TExprBase KqpApplyExtractMembersToReadTable(TExprBase node, TExprContext& ctx, const TParentsMap& parentsMap,
+    bool allowMultiUsage)
+{
+    if (!node.Maybe<TKqlReadTableBase>()) {
         return node;
     }
 
-    auto read = extract.Input().Cast<TKqlReadTableBase>();
+    auto read = node.Cast<TKqlReadTableBase>();
+
+    auto usedColumns = GetUsedColumns(read, read.Columns(), parentsMap, allowMultiUsage, ctx);
+    if (!usedColumns) {
+        return node;
+    }
 
     if (auto maybeIndexRead = read.Maybe<TKqlReadTableIndex>()) {
         auto indexRead = maybeIndexRead.Cast();
 
-        return Build<TKqlReadTableIndex>(ctx, extract.Pos())
+        return Build<TKqlReadTableIndex>(ctx, read.Pos())
             .Table(indexRead.Table())
             .Range(indexRead.Range())
-            .Columns(extract.Members())
+            .Columns(usedColumns.Cast())
             .Index(indexRead.Index())
             .Settings(indexRead.Settings())
             .Done();
     }
 
-    return Build<TKqlReadTableBase>(ctx, extract.Pos())
+    return Build<TKqlReadTableBase>(ctx, read.Pos())
         .CallableName(read.CallableName())
         .Table(read.Table())
         .Range(read.Range())
-        .Columns(extract.Members())
+        .Columns(usedColumns.Cast())
         .Settings(read.Settings())
         .Done();
 }
 
-TExprBase KqpApplyExtractMembersToReadTableRanges(TExprBase node, TExprContext& ctx) {
-    if (!node.Maybe<TCoExtractMembers>()) {
+TExprBase KqpApplyExtractMembersToReadTableRanges(TExprBase node, TExprContext& ctx, const TParentsMap& parentsMap,
+    bool allowMultiUsage)
+{
+    if (!node.Maybe<TKqlReadTableRangesBase>()) {
         return node;
     }
 
-    auto extract = node.Cast<TCoExtractMembers>();
-    auto input = extract.Input();
-
-    if (!input.Maybe<TKqlReadTableRangesBase>()) {
+    // TKqpReadOlapTableRangesBase is derived from TKqlReadTableRangesBase, but should be handled separately
+    if (node.Maybe<TKqpReadOlapTableRangesBase>()) {
         return node;
     }
 
-    // TKqpReadOlapTableRangesBase is derived from TKqlReadTableRangesBase, but should be handled separately
-    if (input.Maybe<TKqpReadOlapTableRangesBase>()) {
+    auto read = node.Cast<TKqlReadTableRangesBase>();
+
+    auto usedColumns = GetUsedColumns(read, read.Columns(), parentsMap, allowMultiUsage, ctx);
+    if (!usedColumns) {
         return node;
     }
 
-    auto read = extract.Input().Cast<TKqlReadTableRangesBase>();
-
-    return Build<TKqlReadTableRangesBase>(ctx, extract.Pos())
+    return Build<TKqlReadTableRangesBase>(ctx, read.Pos())
         .CallableName(read.CallableName())
         .Table(read.Table())
         .Ranges(read.Ranges())
-        .Columns(extract.Members())
+        .Columns(usedColumns.Cast())
         .Settings(read.Settings())
         .ExplainPrompt(read.ExplainPrompt())
         .Done();
 }
 
-TExprBase KqpApplyExtractMembersToReadOlapTable(TExprBase node, TExprContext& ctx) {
-    if (!node.Maybe<TCoExtractMembers>()) {
-        return node;
-    }
-
-    auto extract = node.Cast<TCoExtractMembers>();
-    auto input = extract.Input();
-
-    if (!input.Maybe<TKqlReadTableRangesBase>()) {
+TExprBase KqpApplyExtractMembersToReadOlapTable(TExprBase node, TExprContext& ctx, const TParentsMap& parentsMap,
+    bool allowMultiUsage)
+{
+    if (!node.Maybe<TKqpReadOlapTableRangesBase>()) {
         return node;
     }
 
-    auto read = extract.Input().Cast<TKqpReadOlapTableRangesBase>();
+    auto read = node.Cast<TKqpReadOlapTableRangesBase>();
 
     // When process is set it may use columns in read.Columns() but those columns may not be present
     // in the results. Thus do not apply extract members if process is not empty lambda
+    // TODO: Support process lambda in this rule.
     if (read.Process().Body().Raw() != read.Process().Args().Arg(0).Raw()) {
         return node;
     }
 
-    return Build<TKqpReadOlapTableRangesBase>(ctx, extract.Pos())
+    auto usedColumns = GetUsedColumns(read, read.Columns(), parentsMap, allowMultiUsage, ctx);
+    if (!usedColumns) {
+        return node;
+    }
+
+    return Build<TKqpReadOlapTableRangesBase>(ctx, read.Pos())
         .CallableName(read.CallableName())
         .Table(read.Table())
         .Ranges(read.Ranges())
-        .Columns(extract.Members())
+        .Columns(usedColumns.Cast())
         .Settings(read.Settings())
         .ExplainPrompt(read.ExplainPrompt())
         .Process(read.Process())
         .Done();
 }
 
-TExprBase KqpApplyExtractMembersToLookupTable(TExprBase node, TExprContext& ctx) {
-    if (!node.Maybe<TCoExtractMembers>()) {
+TExprBase KqpApplyExtractMembersToLookupTable(TExprBase node, TExprContext& ctx, const TParentsMap& parentsMap,
+    bool allowMultiUsage)
+{
+    if (!node.Maybe<TKqlLookupTableBase>()) {
         return node;
     }
 
-    auto extract = node.Cast<TCoExtractMembers>();
-    auto input = extract.Input();
+    auto lookup = node.Cast<TKqlLookupTableBase>();
 
-    if (!input.Maybe<TKqlLookupTableBase>()) {
+    auto usedColumns = GetUsedColumns(lookup, lookup.Columns(), parentsMap, allowMultiUsage, ctx);
+    if (!usedColumns) {
         return node;
     }
 
-    auto lookup = extract.Input().Cast<TKqlLookupTableBase>();
-
     if (auto maybeIndexLookup = lookup.Maybe<TKqlLookupIndex>()) {
         auto indexLookup = maybeIndexLookup.Cast();
 
         return Build<TKqlLookupIndex>(ctx, lookup.Pos())
             .Table(indexLookup.Table())
             .LookupKeys(indexLookup.LookupKeys())
-            .Columns(extract.Members())
+            .Columns(usedColumns.Cast())
             .Index(indexLookup.Index())
             .Done();
     }
@@ -140,7 +183,7 @@ TExprBase KqpApplyExtractMembersToLookupTable(TExprBase node, TExprContext& ctx)
         .CallableName(lookup.CallableName())
         .Table(lookup.Table())
         .LookupKeys(lookup.LookupKeys())
-        .Columns(extract.Members())
+        .Columns(usedColumns.Cast())
         .Done();
 }
 
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_helpers.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log_helpers.cpp
new file mode 100644
index 0000000000..17d577c2f3
--- /dev/null
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_helpers.cpp
@@ -0,0 +1,84 @@
+#include "kqp_opt_log_impl.h"
+
+#include <ydb/library/yql/core/yql_opt_utils.h>
+
+namespace NKikimr::NKqp::NOpt {
+
+using namespace NYql;
+using namespace NYql::NNodes;
+
+TExprBase TKqpMatchReadResult::BuildProcessNodes(TExprBase input, TExprContext& ctx) const {
+    auto expr = input;
+
+    if (ExtractMembers) {
+        expr = Build<TCoExtractMembers>(ctx, ExtractMembers.Cast().Pos())
+            .Input(expr)
+            .Members(ExtractMembers.Cast().Members())
+            .Done();
+    }
+
+    if (FilterNullMembers) {
+        expr = Build<TCoFilterNullMembers>(ctx, FilterNullMembers.Cast().Pos())
+            .Input(expr)
+            .Members(FilterNullMembers.Cast().Members())
+            .Done();
+    }
+
+    if (SkipNullMembers) {
+        expr = Build<TCoSkipNullMembers>(ctx, SkipNullMembers.Cast().Pos())
+            .Input(expr)
+            .Members(SkipNullMembers.Cast().Members())
+            .Done();
+    }
+
+    if (FlatMap) {
+        expr = Build<TCoFlatMap>(ctx, FlatMap.Cast().Pos())
+            .Input(expr)
+            .Lambda(FlatMap.Cast().Lambda())
+            .Done();
+    }
+
+    return expr;
+}
+
+TMaybe<TKqpMatchReadResult> MatchRead(TExprBase node, std::function<bool(TExprBase)> matchFunc) {
+    auto expr = node;
+
+    TMaybeNode<TCoFlatMap> flatmap;
+    if (auto maybeNode = expr.Maybe<TCoFlatMap>()) {
+        flatmap = maybeNode;
+        expr = maybeNode.Cast().Input();
+    }
+
+    TMaybeNode<TCoSkipNullMembers> skipNullMembers;
+    if (auto maybeNode = expr.Maybe<TCoSkipNullMembers>()) {
+        skipNullMembers = maybeNode;
+        expr = maybeNode.Cast().Input();
+    }
+
+    TMaybeNode<TCoFilterNullMembers> filterNullMembers;
+    if (auto maybeNode = expr.Maybe<TCoFilterNullMembers>()) {
+        filterNullMembers = maybeNode;
+        expr = maybeNode.Cast().Input();
+    }
+
+    TMaybeNode<TCoExtractMembers> extractMembers;
+    if (auto maybeNode = expr.Maybe<TCoExtractMembers>()) {
+        extractMembers = maybeNode;
+        expr = maybeNode.Cast().Input();
+    }
+
+    if (!matchFunc(expr)) {
+        return {};
+    }
+
+    return TKqpMatchReadResult {
+        .Read = expr,
+        .ExtractMembers = extractMembers,
+        .FilterNullMembers = filterNullMembers,
+        .SkipNullMembers = skipNullMembers,
+        .FlatMap = flatmap
+    };
+}
+
+} // namespace NKikimr::NKqp::NOpt
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_impl.h b/ydb/core/kqp/opt/logical/kqp_opt_log_impl.h
new file mode 100644
index 0000000000..49f5c74e03
--- /dev/null
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_impl.h
@@ -0,0 +1,28 @@
+#pragma once
+
+#include <ydb/core/kqp/opt/kqp_opt_impl.h>
+
+namespace NKikimr::NKqp::NOpt {
+
+struct TKqpMatchReadResult {
+    NYql::NNodes::TExprBase Read;
+    NYql::NNodes::TMaybeNode<NYql::NNodes::TCoExtractMembers> ExtractMembers;
+    NYql::NNodes::TMaybeNode<NYql::NNodes::TCoFilterNullMembers> FilterNullMembers;
+    NYql::NNodes::TMaybeNode<NYql::NNodes::TCoSkipNullMembers> SkipNullMembers;
+    NYql::NNodes::TMaybeNode<NYql::NNodes::TCoFlatMap> FlatMap;
+
+    NYql::NNodes::TExprBase BuildProcessNodes(NYql::NNodes::TExprBase input, NYql::TExprContext& ctx) const;
+};
+
+TMaybe<TKqpMatchReadResult> MatchRead(NYql::NNodes::TExprBase node,
+    std::function<bool(NYql::NNodes::TExprBase)> matchFunc);
+
+template<typename TRead>
+TMaybe<TKqpMatchReadResult> MatchRead(NYql::NNodes::TExprBase node) {
+    return MatchRead(node, [] (NYql::NNodes::TExprBase node) { return node.Maybe<TRead>().IsValid(); });
+}
+
+} // NKikimr::NKqp::NOpt
+
+
+
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_join.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log_join.cpp
index ac0fe0bc39..802c6e6e73 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log_join.cpp
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_join.cpp
@@ -1,3 +1,4 @@
+#include "kqp_opt_log_impl.h"
 #include "kqp_opt_log_rules.h"
 
 #include <ydb/core/kqp/opt/kqp_opt_impl.h>
@@ -155,49 +156,25 @@ TMaybeNode<TExprBase> KqpJoinToIndexLookupImpl(const TDqJoin& join, TExprContext
         return {};
     }
 
-    TMaybeNode<TKqlReadTableBase> rightRead;
-    TMaybeNode<TCoFlatMap> rightFlatmap;
-    TMaybeNode<TCoFilterNullMembers> rightFilterNull;
-    TMaybeNode<TCoSkipNullMembers> rightSkipNull;
-
-    if (auto readTable = join.RightInput().Maybe<TKqlReadTableBase>()) {
-        rightRead = readTable;
-    }
-
-    if (auto readTable = join.RightInput().Maybe<TCoFlatMap>().Input().Maybe<TKqlReadTableBase>()) {
-        rightRead = readTable;
-        rightFlatmap = join.RightInput().Maybe<TCoFlatMap>();
-    }
-
-    if (auto readTable = join.RightInput().Maybe<TCoFlatMap>().Input().Maybe<TCoFilterNullMembers>().Input().Maybe<TKqlReadTableBase>()) {
-        rightRead = readTable;
-        rightFlatmap = join.RightInput().Maybe<TCoFlatMap>();
-        rightFilterNull = rightFlatmap.Input().Cast<TCoFilterNullMembers>();
-    }
-
-    if (auto readTable = join.RightInput().Maybe<TCoFlatMap>().Input().Maybe<TCoSkipNullMembers>().Input().Maybe<TKqlReadTableBase>()) {
-        rightRead = readTable;
-        rightFlatmap = join.RightInput().Maybe<TCoFlatMap>();
-        rightSkipNull = rightFlatmap.Input().Cast<TCoSkipNullMembers>();
+    auto rightReadMatch = MatchRead<TKqlReadTableBase>(join.RightInput());
+    if (!rightReadMatch) {
+        return {};
     }
 
-    if (!rightRead) {
+    if (rightReadMatch->FlatMap && !IsPassthroughFlatMap(rightReadMatch->FlatMap.Cast(), nullptr)) {
         return {};
     }
 
+    auto rightRead = rightReadMatch->Read.Cast<TKqlReadTableBase>();
+
     Y_ENSURE(rightRead.Maybe<TKqlReadTable>() || rightRead.Maybe<TKqlReadTableIndex>());
 
-    const TKqlReadTableBase read = rightRead.Cast();
+    const TKqlReadTableBase read = rightRead;
     if (!read.Table().SysView().Value().empty()) {
         // Can't lookup in system views
         return {};
     }
 
-    if (rightFlatmap && !IsPassthroughFlatMap(rightFlatmap.Cast(), nullptr)) {
-        // Can't lookup in modified table
-        return {};
-    }
-
     auto maybeRightTableKeyPrefix = GetRightTableKeyPrefix(read.Range());
     if (!maybeRightTableKeyPrefix) {
         return {};
@@ -373,32 +350,15 @@ TMaybeNode<TExprBase> KqpJoinToIndexLookupImpl(const TDqJoin& join, TExprContext
             .Build()
         .Done();
 
-    if (rightFilterNull) {
-        lookup = Build<TCoFilterNullMembers>(ctx, join.Pos())
-            .Input(lookup)
-            .Members(rightFilterNull.Cast().Members())
-            .Done();
-    }
-
-    if (rightSkipNull) {
-        lookup = Build<TCoSkipNullMembers>(ctx, join.Pos())
-            .Input(lookup)
-            .Members(rightSkipNull.Cast().Members())
-            .Done();
-    }
-
-    if (rightFlatmap) {
-        lookup = Build<TCoFlatMap>(ctx, join.Pos())
-            .Input(lookup)
-            .Lambda(rightFlatmap.Cast().Lambda())
-            .Done();
-    }
+    rightReadMatch->ExtractMembers = {}; // We already fetching only required columns
+    lookup = rightReadMatch->BuildProcessNodes(lookup, ctx);
 
     if (join.JoinType().Value() == "RightSemi") {
         auto arg = TCoArgument(ctx.NewArgument(join.Pos(), "row"));
         auto rightLabel = join.RightLabel().Cast<TCoAtom>().Value();
 
-        TVector<TExprBase> renames = CreateRenames(rightFlatmap, read.Columns(), arg, rightLabel, join.Pos(), ctx);
+        TVector<TExprBase> renames = CreateRenames(rightReadMatch->FlatMap, read.Columns(), arg, rightLabel,
+            join.Pos(), ctx);
 
         lookup = Build<TCoMap>(ctx, join.Pos())
             .Input(lookup)
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_ranges.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log_ranges.cpp
index 16da537a40..beb2a12705 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log_ranges.cpp
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_ranges.cpp
@@ -1,3 +1,4 @@
+#include "kqp_opt_log_impl.h"
 #include "kqp_opt_log_rules.h"
 
 #include <ydb/core/kqp/common/kqp_yql.h>
@@ -125,48 +126,30 @@ TExprBase KqpPushPredicateToReadTable(TExprBase node, TExprContext& ctx, const T
         return node;
     }
 
-    TMaybeNode<TKqlReadTableBase> readTable;
-    TMaybeNode<TCoFilterNullMembers> filterNull;
-    TMaybeNode<TCoSkipNullMembers> skipNull;
-
-    TMaybeNode<TCoAtom> indexName;
-
-    if (auto maybeRead = flatmap.Input().Maybe<TKqlReadTable>()) {
-        readTable = maybeRead.Cast();
-    }
-
-    if (auto maybeRead = flatmap.Input().Maybe<TKqlReadTableIndex>()) {
-        readTable = maybeRead.Cast();
-        indexName = maybeRead.Cast().Index();
-    }
-
-    if (auto maybeRead = flatmap.Input().Maybe<TCoFilterNullMembers>().Input().Maybe<TKqlReadTable>()) {
-        readTable = maybeRead.Cast();
-        filterNull = flatmap.Input().Cast<TCoFilterNullMembers>();
+    auto readMatch = MatchRead<TKqlReadTableBase>(flatmap.Input());
+    if (!readMatch) {
+        return node;
     }
 
-    if (auto maybeRead = flatmap.Input().Maybe<TCoFilterNullMembers>().Input().Maybe<TKqlReadTableIndex>()) {
-        readTable = maybeRead.Cast();
-        filterNull = flatmap.Input().Cast<TCoFilterNullMembers>();
-        indexName = maybeRead.Cast().Index();
+    if (readMatch->FlatMap) {
+        return node;
     }
 
-    if (auto maybeRead = flatmap.Input().Maybe<TCoSkipNullMembers>().Input().Maybe<TKqlReadTable>()) {
-        readTable = maybeRead.Cast();
-        skipNull = flatmap.Input().Cast<TCoSkipNullMembers>();
-    }
+    auto read = readMatch->Read.Cast<TKqlReadTableBase>();
 
-    if (auto maybeRead = flatmap.Input().Maybe<TCoSkipNullMembers>().Input().Maybe<TKqlReadTableIndex>()) {
-        readTable = maybeRead.Cast();
-        skipNull = flatmap.Input().Cast<TCoSkipNullMembers>();
-        indexName = maybeRead.Cast().Index();
-    }
+    static const std::set<TStringBuf> supportedReads {
+        TKqlReadTable::CallableName(),
+        TKqlReadTableIndex::CallableName(),
+    };
 
-    if (!readTable) {
+    if (!supportedReads.contains(read.CallableName())) {
         return node;
     }
 
-    auto read = readTable.Cast();
+    TMaybeNode<TCoAtom> indexName;
+    if (auto maybeIndexRead = read.Maybe<TKqlReadTableIndex>()) {
+        indexName = maybeIndexRead.Cast().Index();
+    }
 
     if (read.Range().From().ArgCount() > 0 || read.Range().To().ArgCount() > 0) {
         return node;
@@ -260,19 +243,7 @@ TExprBase KqpPushPredicateToReadTable(TExprBase node, TExprContext& ctx, const T
 
         auto newBody = ctx.ChangeChild(flatmap.Lambda().Body().Ref(), 0, std::move(residualPredicate));
 
-        if (filterNull) {
-            input = Build<TCoFilterNullMembers>(ctx, node.Pos())
-                .Input(input)
-                .Members(filterNull.Cast().Members())
-                .Done();
-        }
-
-        if (skipNull) {
-            input = Build<TCoSkipNullMembers>(ctx, node.Pos())
-                .Input(input)
-                .Members(skipNull.Cast().Members())
-                .Done();
-        }
+        input = readMatch->BuildProcessNodes(input, ctx);
 
         auto fetch = Build<TCoFlatMap>(ctx, node.Pos())
             .Input(input)
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_ranges_predext.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log_ranges_predext.cpp
index 0bbb0d71cd..3713a1bfa1 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log_ranges_predext.cpp
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_ranges_predext.cpp
@@ -1,3 +1,4 @@
+#include "kqp_opt_log_impl.h"
 #include "kqp_opt_log_rules.h"
 
 #include <ydb/core/kqp/common/kqp_yql.h>
@@ -19,8 +20,8 @@ using namespace NYql::NNodes;
 namespace {
 
 TMaybeNode<TExprBase> TryBuildTrivialReadTable(TCoFlatMap& flatmap, TKqlReadTableRanges readTable,
-    TMaybeNode<TCoFilterNullMembers>& filterNull, TMaybeNode<TCoSkipNullMembers>& skipNull,
-    const TKikimrTableDescription& tableDesc, TExprContext& ctx, const TKqpOptimizeContext& kqpCtx)
+    const TKqpMatchReadResult& readMatch, const TKikimrTableDescription& tableDesc, TExprContext& ctx,
+    const TKqpOptimizeContext& kqpCtx)
 {
     switch (tableDesc.Metadata->Kind) {
         case EKikimrTableKind::Datashard:
@@ -108,19 +109,7 @@ TMaybeNode<TExprBase> TryBuildTrivialReadTable(TCoFlatMap& flatmap, TKqlReadTabl
             .Settings(settings.BuildNode(ctx, readTable.Pos()))
             .Done();
 
-        if (filterNull) {
-            input = Build<TCoFilterNullMembers>(ctx, readTable.Pos())
-                .Input(input)
-                .Members(filterNull.Cast().Members())
-                .Done();
-        }
-
-        if (skipNull) {
-            input = Build<TCoSkipNullMembers>(ctx, readTable.Pos())
-                .Input(input)
-                .Members(skipNull.Cast().Members())
-                .Done();
-        }
+        input = readMatch.BuildProcessNodes(input, ctx);
 
         input = Build<TCoFlatMap>(ctx, readTable.Pos())
             .Input(input)
@@ -160,28 +149,17 @@ TExprBase KqpPushExtractedPredicateToReadTable(TExprBase node, TExprContext& ctx
         return node;
     }
 
-    TMaybeNode<TKqlReadTableRanges> readTable;
-    TMaybeNode<TCoFilterNullMembers> filterNull;
-    TMaybeNode<TCoSkipNullMembers> skipNull;
-
-    if (auto maybeRead = flatmap.Input().Maybe<TKqlReadTableRanges>()) {
-        readTable = maybeRead.Cast();
-    }
-
-    if (auto maybeRead = flatmap.Input().Maybe<TCoFilterNullMembers>().Input().Maybe<TKqlReadTableRanges>()) {
-        readTable = maybeRead.Cast();
-        filterNull = flatmap.Input().Cast<TCoFilterNullMembers>();
-    }
-
-    if (auto maybeRead = flatmap.Input().Maybe<TCoSkipNullMembers>().Input().Maybe<TKqlReadTableRanges>()) {
-        readTable = maybeRead.Cast();
-        skipNull = flatmap.Input().Cast<TCoSkipNullMembers>();
+    auto readMatch = MatchRead<TKqlReadTableRanges>(flatmap.Input());
+    if (!readMatch) {
+        return node;
     }
 
-    if (!readTable) {
+    if (readMatch->FlatMap) {
         return node;
     }
 
+    auto read = readMatch->Read.Cast<TKqlReadTableRanges>();
+
     /*
      * ReadTableRanges supported predicate extraction, but it may be disabled via flag. For example to force
      * pushdown predicates to OLAP SSA program.
@@ -192,8 +170,6 @@ TExprBase KqpPushExtractedPredicateToReadTable(TExprBase node, TExprContext& ctx
         return node;
     }
 
-    auto read = readTable.Cast();
-
     if (!read.Ranges().Maybe<TCoVoid>()) {
         return node;
     }
@@ -201,7 +177,7 @@ TExprBase KqpPushExtractedPredicateToReadTable(TExprBase node, TExprContext& ctx
     const auto& tableDesc = kqpCtx.Tables->ExistingTable(kqpCtx.Cluster, read.Table().Path());
 
     // test for trivial cases (explicit literals or parameters)
-    if (auto expr = TryBuildTrivialReadTable(flatmap, read, filterNull, skipNull, tableDesc, ctx, kqpCtx)) {
+    if (auto expr = TryBuildTrivialReadTable(flatmap, read, *readMatch, tableDesc, ctx, kqpCtx)) {
         return expr.Cast();
     }
 
@@ -247,19 +223,7 @@ TExprBase KqpPushExtractedPredicateToReadTable(TExprBase node, TExprContext& ctx
         .ExplainPrompt(prompt.BuildNode(ctx, read.Pos()))
         .Done();
 
-    if (filterNull) {
-        input = Build<TCoFilterNullMembers>(ctx, node.Pos())
-            .Input(input)
-            .Members(filterNull.Cast().Members())
-            .Done();
-    }
-
-    if (skipNull) {
-        input = Build<TCoSkipNullMembers>(ctx, node.Pos())
-            .Input(input)
-            .Members(skipNull.Cast().Members())
-            .Done();
-    }
+    input = readMatch->BuildProcessNodes(input, ctx);
 
     return Build<TCoFlatMap>(ctx, node.Pos())
         .Input(input)
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_rules.h b/ydb/core/kqp/opt/logical/kqp_opt_log_rules.h
index a0080e9363..e61e3cc4af 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log_rules.h
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_rules.h
@@ -24,14 +24,6 @@ NYql::NNodes::TExprBase KqpPushPredicateToReadTable(NYql::NNodes::TExprBase node
 NYql::NNodes::TExprBase KqpPushExtractedPredicateToReadTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx,
     const TKqpOptimizeContext& kqpCtx, NYql::TTypeAnnotationContext& typesCtx);
 
-NYql::NNodes::TExprBase KqpApplyExtractMembersToReadTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx);
-
-NYql::NNodes::TExprBase KqpApplyExtractMembersToReadOlapTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx);
-
-NYql::NNodes::TExprBase KqpApplyExtractMembersToReadTableRanges(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx);
-
-NYql::NNodes::TExprBase KqpApplyExtractMembersToLookupTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx);
-
 NYql::NNodes::TExprBase KqpJoinToIndexLookup(const NYql::NNodes::TExprBase& node, NYql::TExprContext& ctx,
     const TKqpOptimizeContext& kqpCtx, const NYql::TKikimrConfiguration::TPtr& config);
 
@@ -60,4 +52,16 @@ NYql::NNodes::TExprBase KqpExcessUpsertInputColumns(const NYql::NNodes::TExprBas
 NYql::NNodes::TExprBase KqpDropTakeOverLookupTable(const NYql::NNodes::TExprBase& node, NYql::TExprContext& ctx,
     const TKqpOptimizeContext& kqpCtx);
 
+NYql::NNodes::TExprBase KqpApplyExtractMembersToReadTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx,
+    const NYql::TParentsMap& parentsMap, bool allowMultiUsage);
+
+NYql::NNodes::TExprBase KqpApplyExtractMembersToReadOlapTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx,
+    const NYql::TParentsMap& parentsMap, bool allowMultiUsage);
+
+NYql::NNodes::TExprBase KqpApplyExtractMembersToReadTableRanges(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx,
+    const NYql::TParentsMap& parentsMap, bool allowMultiUsage);
+
+NYql::NNodes::TExprBase KqpApplyExtractMembersToLookupTable(NYql::NNodes::TExprBase node, NYql::TExprContext& ctx,
+    const NYql::TParentsMap& parentsMap, bool allowMultiUsage);
+
 } // namespace NKikimr::NKqp::NOpt
diff --git a/ydb/core/kqp/opt/logical/kqp_opt_log_sqlin.cpp b/ydb/core/kqp/opt/logical/kqp_opt_log_sqlin.cpp
index fd5d213001..a09a982c56 100644
--- a/ydb/core/kqp/opt/logical/kqp_opt_log_sqlin.cpp
+++ b/ydb/core/kqp/opt/logical/kqp_opt_log_sqlin.cpp
@@ -1,3 +1,4 @@
+#include "kqp_opt_log_impl.h"
 #include "kqp_opt_log_rules.h"
 
 #include <ydb/core/kqp/opt/kqp_opt_impl.h>
@@ -40,19 +41,32 @@ TExprBase KqpRewriteSqlInToEquiJoin(const TExprBase& node, TExprContext& ctx, co
         return node;
     }
 
-    if (!flatMap.Input().Maybe<TKqlReadTable>() && !flatMap.Input().Maybe<TKqlReadTableIndex>()) {
+    auto readMatch = MatchRead<TKqlReadTableBase>(flatMap.Input());
+    if (!readMatch) {
         return node;
     }
 
-    const auto readTable = flatMap.Input().Cast<TKqlReadTableBase>();
+    if (readMatch->FlatMap) {
+        return node;
+    }
+
+    auto readTable = readMatch->Read.Cast<TKqlReadTableBase>();
+
+    static const std::set<TStringBuf> supportedReads {
+        TKqlReadTable::CallableName(),
+        TKqlReadTableIndex::CallableName(),
+    };
+
+    if (!supportedReads.contains(readTable.CallableName())) {
+        return node;
+    }
 
     if (!readTable.Table().SysView().Value().empty()) {
         return node;
     }
 
     TString lookupTable;
-
-    if (auto indexRead = flatMap.Input().Maybe<TKqlReadTableIndex>()) {
+    if (auto indexRead = readTable.Maybe<TKqlReadTableIndex>()) {
         lookupTable = GetIndexMetadata(indexRead.Cast(), *kqpCtx.Tables, kqpCtx.Cluster)->Name;
     } else {
         lookupTable = readTable.Table().Path().StringValue();
diff --git a/ydb/core/kqp/opt/logical/ya.make b/ydb/core/kqp/opt/logical/ya.make
index b9811606c1..b4ddde4091 100644
--- a/ydb/core/kqp/opt/logical/ya.make
+++ b/ydb/core/kqp/opt/logical/ya.make
@@ -8,6 +8,7 @@ OWNER(
 SRCS(
     kqp_opt_log_effects.cpp
     kqp_opt_log_extract.cpp
+    kqp_opt_log_helpers.cpp
     kqp_opt_log_join.cpp
     kqp_opt_log_indexes.cpp
     kqp_opt_log_ranges.cpp
diff --git a/ydb/core/kqp/ut/kqp_indexes_ut.cpp b/ydb/core/kqp/ut/kqp_indexes_ut.cpp
index c2580fb2aa..b33b433e42 100644
--- a/ydb/core/kqp/ut/kqp_indexes_ut.cpp
+++ b/ydb/core/kqp/ut/kqp_indexes_ut.cpp
@@ -5226,6 +5226,45 @@ R"([[#;#;["Primary1"];[41u]];[["Secondary2"];[2u];["Primary2"];[42u]];[["Seconda
             [[2];[40u];["Two"];["Value4"]]
         ])", FormatResultSetYson(result.GetResultSet(0)));
     }
+
+    Y_UNIT_TEST_TWIN(IndexMultipleRead, UseNewEngine) {
+        TKikimrRunner kikimr;
+
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+        CreateSampleTablesWithIndex(session);
+
+        NYdb::NTable::TExecDataQuerySettings execSettings;
+        execSettings.CollectQueryStats(ECollectQueryStatsMode::Basic);
+
+        auto params = db.GetParamsBuilder()
+            .AddParam("$fks")
+                .BeginList()
+                .AddListItem().Int32(5)
+                .AddListItem().Int32(10)
+                .EndList()
+                .Build()
+            .Build();
+
+        auto result = session.ExecuteDataQuery(Q1_(R"(
+            DECLARE $fks AS List<Int32>;
+
+            SELECT * FROM SecondaryKeys VIEW Index WHERE Fk IN $fks;
+            SELECT COUNT(*) FROM SecondaryKeys VIEW Index WHERE Fk IN $fks;
+        )"), TTxControl::BeginTx().CommitTx(), params, execSettings).ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+        AssertTableStats(result, "/Root/SecondaryKeys", {
+            .ExpectedReads = UseNewEngine ? 2 : 1, // TODO: Looks like missing SkipNullMembers in NewEngine
+        });
+
+        AssertTableStats(result, "/Root/SecondaryKeys/Index/indexImplTable", {
+            .ExpectedReads = UseNewEngine ? 1 : 2,
+        });
+
+        CompareYson(R"([[[5];[5];["Payload5"]]])", FormatResultSetYson(result.GetResultSet(0)));
+        CompareYson(R"([[1u]])", FormatResultSetYson(result.GetResultSet(1)));
+    }
 }
 
 }
diff --git a/ydb/core/kqp/ut/kqp_join_ut.cpp b/ydb/core/kqp/ut/kqp_join_ut.cpp
index f934b6d88d..31e85f2755 100644
--- a/ydb/core/kqp/ut/kqp_join_ut.cpp
+++ b/ydb/core/kqp/ut/kqp_join_ut.cpp
@@ -301,6 +301,26 @@ Y_UNIT_TEST_SUITE(KqpJoin) {
         AssertTableReads(result, "/Root/Join1_2", 3);
     }
 
+    Y_UNIT_TEST_NEW_ENGINE(IdxLookupSelf) {
+        TKikimrRunner kikimr(SyntaxV1Settings());
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        CreateSampleTables(session);
+
+        const TString query = Q_(R"(
+            SELECT t1.Fk21 AS Key, t2.Value AS Value
+            FROM Join1_1 AS t1
+            LEFT JOIN Join1_1 AS t2
+            ON t1.Fk21 == t2.Key
+            WHERE t1.Key == 2
+            ORDER BY Key;
+        )");
+
+        auto result = ExecQueryAndTestResult(session, query, R"([[[102];#]])");
+        AssertTableReads(result, "/Root/Join1_1", 1);
+    }
+
     Y_UNIT_TEST_NEW_ENGINE(LeftJoinWithNull) {
         TKikimrRunner kikimr;
         auto db = kikimr.GetTableClient();
diff --git a/ydb/core/kqp/ut/kqp_ne_effects_ut.cpp b/ydb/core/kqp/ut/kqp_ne_effects_ut.cpp
index 17b315ac4d..038e59551c 100644
--- a/ydb/core/kqp/ut/kqp_ne_effects_ut.cpp
+++ b/ydb/core/kqp/ut/kqp_ne_effects_ut.cpp
@@ -505,6 +505,44 @@ Y_UNIT_TEST_SUITE(KqpNewEngineEffects) {
             [[4000000003u];["Updated"];[6]]
         ])", FormatResultSetYson(result.GetResultSet(0)));
     }
+
+    Y_UNIT_TEST(DeletePkPrefixWithIndex) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        auto schemeResult = session.ExecuteSchemeQuery(R"(
+            --!syntax_v1
+
+            CREATE TABLE Tmp (
+                Key1 Uint32,
+                Key2 String,
+                Value String,
+                PRIMARY KEY (Key1, Key2),
+                INDEX Index GLOBAL ON (Value, Key1)
+            )
+        )").ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(schemeResult.GetStatus(), EStatus::SUCCESS, schemeResult.GetIssues().ToString());
+
+        auto result = session.ExplainDataQuery(R"(
+            --!syntax_v1
+            PRAGMA kikimr.UseNewEngine = 'true';
+
+            DECLARE $key1 as Uint32;
+
+            DELETE FROM Tmp WHERE Key1 = $key1;
+        )").ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+        NJson::TJsonValue plan;
+        NJson::ReadJsonTree(result.GetPlan(), &plan, true);
+        auto table = plan["tables"][0];
+        UNIT_ASSERT_VALUES_EQUAL(table["name"], "/Root/Tmp");
+        auto reads = table["reads"].GetArraySafe();
+        UNIT_ASSERT_VALUES_EQUAL(reads.size(), 1);
+        UNIT_ASSERT_VALUES_EQUAL(reads[0]["type"], "Lookup");
+        UNIT_ASSERT_VALUES_EQUAL(reads[0]["columns"].GetArraySafe().size(), 3);
+    }
 }
 
 } // namespace NKqp
diff --git a/ydb/core/kqp/ut/kqp_ne_ut.cpp b/ydb/core/kqp/ut/kqp_ne_ut.cpp
index 925fd21954..7c513c994c 100644
--- a/ydb/core/kqp/ut/kqp_ne_ut.cpp
+++ b/ydb/core/kqp/ut/kqp_ne_ut.cpp
@@ -2995,6 +2995,8 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
         )").ExtractValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
 
+        //Cerr << result.GetPlan() << Endl;
+
         NJson::TJsonValue plan;
         NJson::ReadJsonTree(result.GetPlan(), &plan, true);
         auto reads = plan["tables"][0]["reads"].GetArraySafe();
@@ -3063,6 +3065,52 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
         UNIT_ASSERT_VALUES_EQUAL(1, stats.query_phases()[0].affected_shards());
         UNIT_ASSERT_VALUES_EQUAL("/Root/Logs", stats.query_phases()[0].table_access()[0].name());
     }
+
+    Y_UNIT_TEST(ReadDifferentColumns) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        auto result = session.ExplainDataQuery(R"(
+            --!syntax_v1
+            PRAGMA kikimr.UseNewEngine = 'true';
+
+            SELECT Fk21 FROM Join1 WHERE Value = "Value1";
+            SELECT Fk22 FROM Join1 WHERE Value = "Value2";
+        )").ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+        NJson::TJsonValue plan;
+        NJson::ReadJsonTree(result.GetPlan(), &plan, true);
+        auto reads = plan["tables"][0]["reads"].GetArraySafe();
+        UNIT_ASSERT_VALUES_EQUAL(reads.size(), 1);
+        UNIT_ASSERT_VALUES_EQUAL(reads[0]["columns"].GetArraySafe().size(), 3);
+    }
+
+    Y_UNIT_TEST(ReadDifferentColumnsPk) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        auto result = session.ExplainDataQuery(R"(
+            --!syntax_v1
+            PRAGMA kikimr.UseNewEngine = 'true';
+
+            SELECT Fk21 FROM Join1 WHERE Key = 1;
+            SELECT Fk22 FROM Join1 WHERE Value = "Value2";
+        )").ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+        NJson::TJsonValue plan;
+        NJson::ReadJsonTree(result.GetPlan(), &plan, true);
+        auto reads = plan["tables"][0]["reads"].GetArraySafe();
+        UNIT_ASSERT_VALUES_EQUAL(reads.size(), 2);
+
+        TSet<TString> readTypes;
+        readTypes.insert(reads[0]["type"].GetString());
+        readTypes.insert(reads[1]["type"].GetString());
+        UNIT_ASSERT(readTypes.contains("Lookup"));
+    }
 }
 
 } // namespace NKikimr::NKqp