fix a bug (#45184)

15 files changed, 131 insertions, 27 deletions

diff --git a/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp b/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp
index 21d2b9758d4..5c28408666f 100644
--- a/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp
@@ -977,6 +977,7 @@ void TKqpTasksGraph::BuildStreamLookupChannels(const TStageInfo& stageInfo, ui32
 
     settings->SetLookupStrategy(streamLookup.GetLookupStrategy());
     settings->SetKeepRowsOrder(streamLookup.GetKeepRowsOrder());
+    settings->SetCookieFormatVersion(streamLookup.GetCookieFormatVersion());
     settings->SetAllowNullKeysPrefixSize(streamLookup.GetAllowNullKeysPrefixSize());
     settings->SetIsolationLevel(GetMeta().RequestIsolationLevel);
 
diff --git a/ydb/core/kqp/query_compiler/kqp_mkql_compiler.cpp b/ydb/core/kqp/query_compiler/kqp_mkql_compiler.cpp
index 4f7276faf91..81b40f2a295 100644
--- a/ydb/core/kqp/query_compiler/kqp_mkql_compiler.cpp
+++ b/ydb/core/kqp/query_compiler/kqp_mkql_compiler.cpp
@@ -423,7 +423,8 @@ TIntrusivePtr<IMkqlCallableCompiler> CreateKqlCompiler(const TKqlCompileContext&
 
             auto input = MkqlBuildExpr(indexLookupJoin.Input().Ref(), buildCtx);
 
-            return ctx.PgmBuilder().KqpIndexLookupJoin(input, joinType, leftLabel, rightLabel);
+            return ctx.PgmBuilder().KqpIndexLookupJoin(input, joinType, leftLabel, rightLabel,
+                ctx.StreamLookupJoinCookieVersion());
         });
 
     compiler->AddCallable(TDqBlockHashJoinCore::CallableName(),
diff --git a/ydb/core/kqp/query_compiler/kqp_mkql_compiler.h b/ydb/core/kqp/query_compiler/kqp_mkql_compiler.h
index ebb7ccd718d..2692c73aacc 100644
--- a/ydb/core/kqp/query_compiler/kqp_mkql_compiler.h
+++ b/ydb/core/kqp/query_compiler/kqp_mkql_compiler.h
@@ -17,18 +17,24 @@ public:
     TKqlCompileContext(const TString& cluster,
         const TIntrusivePtr<NYql::TKikimrTablesData>& tablesData,
         const NMiniKQL::TTypeEnvironment& typeEnv,
-        const NMiniKQL::IFunctionRegistry& funcRegistry)
+        const NMiniKQL::IFunctionRegistry& funcRegistry,
+        ui32 streamLookupJoinCookieVersion = 0)
         : Cluster_(cluster)
         , TablesData_(tablesData)
-        , PgmBuilder_(MakeHolder<NMiniKQL::TKqpProgramBuilder>(typeEnv, funcRegistry)) {}
+        , PgmBuilder_(MakeHolder<NMiniKQL::TKqpProgramBuilder>(typeEnv, funcRegistry))
+        , StreamLookupJoinCookieVersion_(streamLookupJoinCookieVersion) {}
 
     NMiniKQL::TKqpProgramBuilder& PgmBuilder() const { return *PgmBuilder_; }
     const NYql::TKikimrTableMetadata& GetTableMeta(const NYql::NNodes::TKqpTable& table) const;
 
+    // Cookie wire format version to emit for the index lookup join consumer node.
+    ui32 StreamLookupJoinCookieVersion() const { return StreamLookupJoinCookieVersion_; }
+
 private:
     TString Cluster_;
     TIntrusivePtr<NYql::TKikimrTablesData> TablesData_;
     THolder<NMiniKQL::TKqpProgramBuilder> PgmBuilder_;
+    ui32 StreamLookupJoinCookieVersion_;
 };
 
 TIntrusivePtr<NYql::NCommon::IMkqlCallableCompiler> CreateKqlCompiler(
diff --git a/ydb/core/kqp/query_compiler/kqp_query_compiler.cpp b/ydb/core/kqp/query_compiler/kqp_query_compiler.cpp
index 321fff087a5..06aa69deb11 100644
--- a/ydb/core/kqp/query_compiler/kqp_query_compiler.cpp
+++ b/ydb/core/kqp/query_compiler/kqp_query_compiler.cpp
@@ -725,7 +725,10 @@ public:
         , FuncRegistry(funcRegistry)
         , Alloc(__LOCATION__, TAlignedPagePoolCounters(), funcRegistry.SupportsSizedAllocators())
         , TypeEnv(Alloc)
-        , KqlCtx(cluster, optimizeCtx.Tables, TypeEnv, FuncRegistry)
+        // 1 == StreamLookupJoinCookieVersionV1 (see kqp_stream_lookup_join_helpers.h);
+        // kept as a literal to avoid a runtime PEERDIR from the compiler.
+        , KqlCtx(cluster, optimizeCtx.Tables, TypeEnv, FuncRegistry,
+            config->GetEnableStreamLookupJoinCookieV2() ? 1u : 0u)
         , KqlCompiler(CreateKqlCompiler(KqlCtx, typesCtx))
         , TypesCtx(typesCtx)
         , OptimizeCtx(optimizeCtx)
@@ -2824,6 +2827,11 @@ private:
             auto settings = TKqpStreamLookupSettings::Parse(streamLookup);
             streamLookupProto.SetLookupStrategy(GetStreamLookupStrategy(settings.Strategy));
             streamLookupProto.SetKeepRowsOrder(Config->OrderPreservingLookupJoinEnabled());
+            if (Config->GetEnableStreamLookupJoinCookieV2()) {
+                // 1 == StreamLookupJoinCookieVersionV1 (see kqp_stream_lookup_join_helpers.h);
+                // kept as a literal to avoid a runtime PEERDIR from the compiler.
+                streamLookupProto.SetCookieFormatVersion(1);
+            }
             if (settings.AllowNullKeysPrefixSize) {
                 streamLookupProto.SetAllowNullKeysPrefixSize(*settings.AllowNullKeysPrefixSize);
             }
diff --git a/ydb/core/kqp/runtime/kqp_compute.cpp b/ydb/core/kqp/runtime/kqp_compute.cpp
index bee0cb0e491..c17fdbc3b38 100644
--- a/ydb/core/kqp/runtime/kqp_compute.cpp
+++ b/ydb/core/kqp/runtime/kqp_compute.cpp
@@ -152,7 +152,7 @@ public:
 
         bool OmitRowLeftJoin(TState& state, ui64 header, bool isNull) {
 
-            auto cookie = NKqp::TStreamLookupJoinRowCookie::Decode(header);
+            auto cookie = NKqp::TStreamLookupJoinRowCookie::Decode(header, Self->CookieFormatVersion);
 
             if (cookie.LastRow) {
                 // if row is the first and last row in the sequence at the same time
@@ -215,7 +215,7 @@ public:
             // Decode metadata from the row header
             // Contains information about the position of this right row in the sequence
             // of potential matches for the current left row
-            auto meta = NKqp::TStreamLookupJoinRowCookie::Decode(rowMeta);
+            auto meta = NKqp::TStreamLookupJoinRowCookie::Decode(rowMeta, Self->CookieFormatVersion);
 
             // Case 1: Right row is NULL (didn't pass filters) AND this isn't the last potential match
             // We need to wait for more potential matches before making a final decision
@@ -283,7 +283,7 @@ public:
             //
             // Even though we're processing a single logical right row, the lookup might return
             // multiple related rows that need to be considered as a group for semi-join semantics
-            auto meta = NKqp::TStreamLookupJoinRowCookie::Decode(rowMeta);
+            auto meta = NKqp::TStreamLookupJoinRowCookie::Decode(rowMeta, Self->CookieFormatVersion);
 
             // Optimization for single-row sequences from lookup:
             // If a row is both the first and last in its sequence, it's a complete unit
@@ -394,7 +394,8 @@ public:
 
 public:
     TKqpIndexLookupJoinWrapper(TComputationMutables& mutables, IComputationNode* inputNode,
-        EJoinKind joinType, TVector<ui32>&& leftColumnsIndices, TVector<ui32>&& rightColumnsIndices)
+        EJoinKind joinType, TVector<ui32>&& leftColumnsIndices, TVector<ui32>&& rightColumnsIndices,
+        ui32 cookieFormatVersion)
         : TMutableComputationNode<TKqpIndexLookupJoinWrapper>(mutables)
         , InputNode(inputNode)
         , JoinType(joinType)
@@ -402,6 +403,7 @@ public:
         , RightColumnsIndices(std::move(rightColumnsIndices))
         , ResultRowCache(mutables)
         , StateIndex(mutables.CurValueIndex++)
+        , CookieFormatVersion(cookieFormatVersion)
     {
     }
 
@@ -421,6 +423,7 @@ private:
     const TVector<ui32> RightColumnsIndices;
     const TContainerCacheOnContext ResultRowCache;
     const ui32 StateIndex;
+    const ui32 CookieFormatVersion;
 };
 
 } // namespace
@@ -440,7 +443,14 @@ IComputationNode* WrapKqpEnsure(TCallable& callable, const TComputationNodeFacto
 }
 
 IComputationNode* WrapKqpIndexLookupJoin(TCallable& callable, const TComputationNodeFactoryContext& ctx) {
-    MKQL_ENSURE(callable.GetInputsCount() == 4, "Expected 4 args");
+    // The 5th arg (cookie format version) is optional: legacy programs omit it and
+    // default to version 0. See NKqp::StreamLookupJoinCookieVersion* in the helpers.
+    MKQL_ENSURE(callable.GetInputsCount() == 4 || callable.GetInputsCount() == 5, "Expected 4 or 5 args");
+
+    ui32 cookieFormatVersion = 0;
+    if (callable.GetInputsCount() == 5) {
+        cookieFormatVersion = AS_VALUE(TDataLiteral, callable.GetInput(4))->AsValue().Get<ui32>();
+    }
 
     auto inputNode = LocateNode(ctx.NodeLocator, callable, 0);
     ui32 joinKind = AS_VALUE(TDataLiteral, callable.GetInput(1))->AsValue().Get<ui32>();
@@ -463,7 +473,7 @@ IComputationNode* WrapKqpIndexLookupJoin(TCallable& callable, const TComputation
         rightColumnsIndices[rightIndex] = resultIndex;
     }
 
-    return new TKqpIndexLookupJoinWrapper(ctx.Mutables, inputNode, GetJoinKind(joinKind), std::move(leftColumnsIndices), std::move(rightColumnsIndices));
+    return new TKqpIndexLookupJoinWrapper(ctx.Mutables, inputNode, GetJoinKind(joinKind), std::move(leftColumnsIndices), std::move(rightColumnsIndices), cookieFormatVersion);
 }
 
 } // namespace NMiniKQL
diff --git a/ydb/core/kqp/runtime/kqp_program_builder.cpp b/ydb/core/kqp/runtime/kqp_program_builder.cpp
index 487ac7d0383..291f00a3731 100644
--- a/ydb/core/kqp/runtime/kqp_program_builder.cpp
+++ b/ydb/core/kqp/runtime/kqp_program_builder.cpp
@@ -284,7 +284,7 @@ TRuntimeNode TKqpProgramBuilder::KqpEnsure(TRuntimeNode value, TRuntimeNode pred
 }
 
 TRuntimeNode TKqpProgramBuilder::KqpIndexLookupJoin(const TRuntimeNode& input, const TString& joinType,
-    const TString& leftLabel, const TString& rightLabel) {
+    const TString& leftLabel, const TString& rightLabel, ui32 cookieFormatVersion) {
 
     auto inputRowItems = AS_TYPE(TTupleType, AS_TYPE(TStreamType, input.GetStaticType())->GetItemType());
     MKQL_ENSURE(inputRowItems->GetElementsCount() == 3, "Expected 3 elements");
@@ -352,6 +352,11 @@ TRuntimeNode TKqpProgramBuilder::KqpIndexLookupJoin(const TRuntimeNode& input, c
     callableBuilder.Add(NewDataLiteral<ui32>((ui32)GetIndexLookupJoinKind(joinType)));
     callableBuilder.Add(TRuntimeNode(leftIndicesMap.Build(), true));
     callableBuilder.Add(TRuntimeNode(rightIndicesMap.Build(), true));
+    // Legacy (v0) programs stay 4-arg for compatibility with older binaries during
+    // a rolling upgrade; the cookie format version is only appended when non-legacy.
+    if (cookieFormatVersion != 0) {
+        callableBuilder.Add(NewDataLiteral<ui32>(cookieFormatVersion));
+    }
     return TRuntimeNode(callableBuilder.Build(), false);
 }
 
diff --git a/ydb/core/kqp/runtime/kqp_program_builder.h b/ydb/core/kqp/runtime/kqp_program_builder.h
index a8c8900ca32..4d4160e0830 100644
--- a/ydb/core/kqp/runtime/kqp_program_builder.h
+++ b/ydb/core/kqp/runtime/kqp_program_builder.h
@@ -66,7 +66,7 @@ public:
 
     TRuntimeNode KqpEnsure(TRuntimeNode value, TRuntimeNode predicate, TRuntimeNode issueCode, TRuntimeNode message);
 
-    TRuntimeNode KqpIndexLookupJoin(const TRuntimeNode& input, const TString& joinType, const TString& leftLabel, const TString& rightLabel);
+    TRuntimeNode KqpIndexLookupJoin(const TRuntimeNode& input, const TString& joinType, const TString& leftLabel, const TString& rightLabel, ui32 cookieFormatVersion = 0);
 
     TRuntimeNode FulltextAnalyze(TRuntimeNode text, TRuntimeNode settings, TRuntimeNode mode);
 
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.cpp b/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.cpp
index 2f8a4d86508..ce49db9b100 100644
--- a/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.cpp
+++ b/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.cpp
@@ -5,11 +5,17 @@ namespace {
 
 constexpr ui64 FirstRowMask = static_cast<ui64>(1) << 0;
 constexpr ui64 LastRowMask = static_cast<ui64>(1) << 1;
-constexpr ui64 ReservedSpace = 20;
+
+ui64 ReservedBits(ui32 version) {
+    return version == StreamLookupJoinCookieVersionLegacy
+        ? StreamLookupJoinCookieV0ReservedBits
+        : StreamLookupJoinCookieV1ReservedBits;
+}
+
 }
 
-ui64 TStreamLookupJoinRowCookie::Encode() const {
-    ui64 result = (RowSeqNo << ReservedSpace);
+ui64 TStreamLookupJoinRowCookie::Encode(ui32 version) const {
+    ui64 result = (RowSeqNo << ReservedBits(version));
 
     if (FirstRow) {
         result |= FirstRowMask;
@@ -22,13 +28,13 @@ ui64 TStreamLookupJoinRowCookie::Encode() const {
     return result;
 }
 
-TStreamLookupJoinRowCookie TStreamLookupJoinRowCookie::Decode(ui64 encoded) {
+TStreamLookupJoinRowCookie TStreamLookupJoinRowCookie::Decode(ui64 encoded, ui32 version) {
     return TStreamLookupJoinRowCookie{
-        .RowSeqNo = (encoded >> ReservedSpace),
+        .RowSeqNo = (encoded >> ReservedBits(version)),
         .LastRow = bool(encoded & LastRowMask),
         .FirstRow = bool(encoded & FirstRowMask)
     };
 }
 
 
-}
\ No newline at end of file
+}
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.h b/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.h
index 6c3d8af7d04..84b2d01d53c 100644
--- a/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.h
+++ b/ydb/core/kqp/runtime/kqp_stream_lookup_join_helpers.h
@@ -4,14 +4,34 @@
 
 namespace NKikimr::NKqp {
 
+// Cookie format versions for the stream lookup join per-row RowSeqNo wire value.
+//
+// v0 (legacy) packs RowSeqNo into the high 44 bits and reserves the low 20 bits
+// for flags. Because the seqno allocator builds rowSeqNo as (taskId << 40) | counter,
+// a full 64-bit value, the extra `<< 20` in Encode overflows and truncates the
+// taskId prefix for any taskId >= 16 — distinct rows from different distributed
+// tasks then collide on the same RowSeqNo. See StreamLookupJoin_RowSeqNoCollision_Repro.
+//
+// v1 fixes the bit budget: RowSeqNo occupies the high 60 bits (shifted by only 4),
+// so taskId (20 bits) + counter (40 bits) round-trip losslessly.
+//
+// The format version is chosen per query from a config flag and passed explicitly
+// to every producer and consumer of the cookie; it is never inferred from the
+// encoded value.
+inline constexpr ui32 StreamLookupJoinCookieVersionLegacy = 0;
+inline constexpr ui32 StreamLookupJoinCookieVersionV1 = 1;
+
+inline constexpr ui32 StreamLookupJoinCookieV0ReservedBits = 20;
+inline constexpr ui32 StreamLookupJoinCookieV1ReservedBits = 4;
+
 // maybe pack into a union?
 struct TStreamLookupJoinRowCookie {
     ui64 RowSeqNo = 0;
     bool LastRow = false;
     bool FirstRow = false;
 
-    ui64 Encode() const;
-    static TStreamLookupJoinRowCookie Decode(ui64 cookie);
+    ui64 Encode(ui32 version) const;
+    static TStreamLookupJoinRowCookie Decode(ui64 cookie, ui32 version);
 };
 
-} // namespace NKikimr::NKqp
\ No newline at end of file
+} // namespace NKikimr::NKqp
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp b/ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp
index 36c5ca0d037..b01effaa7b9 100644
--- a/ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp
+++ b/ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp
@@ -18,7 +18,20 @@ namespace NKikimr {
 namespace NKqp {
 
 constexpr ui64 SEQNO_SPACE = 40;
-constexpr ui64 MaxTaskId = (1ULL << (64 - SEQNO_SPACE));
+
+// Upper bound on the DQ task id whose rowSeqNo (taskId << SEQNO_SPACE) round-trips
+// losslessly through the cookie of the given format version. The cookie reserves
+// some low bits for flags, so taskId + counter must fit in 64 - reservedBits bits.
+//
+// v0 keeps the historical (intentionally too permissive) 2^24 bound so that with
+// the new-format feature flag disabled behavior is byte-identical to before; that
+// path still overflows for taskId >= 16, exactly as it does on current main.
+ui64 MaxTaskIdForCookieVersion(ui32 version) {
+    const ui64 reservedBits = (version == StreamLookupJoinCookieVersionLegacy)
+        ? 0
+        : StreamLookupJoinCookieV1ReservedBits;
+    return 1ULL << (64 - SEQNO_SPACE - reservedBits);
+}
 
 TStreamLookupShardReadResult::TStreamLookupShardReadResult(const ui64 shardId, THolder<TEventHandle<TEvDataShard::TEvReadResult>> readResult, NMiniKQL::TAllocState* alloc)
     : ShardId(shardId)
@@ -606,7 +619,7 @@ public:
         , InputRowSeqNo(taskId << SEQNO_SPACE)
         , InputRowSeqNoLast((taskId + 1) << SEQNO_SPACE)
     {
-        YQL_ENSURE(taskId < MaxTaskId);
+        YQL_ENSURE(taskId < MaxTaskIdForCookieVersion(Settings.CookieFormatVersion));
         // read columns should contain join key and result columns
         for (const auto& joinKey : Settings.InputColumns) {
             ReadColumns.emplace(joinKey.Name, joinKey);
@@ -646,7 +659,7 @@ public:
         bool lastRow = true;
         if (IsInputTriplet()) {
             auto value = inputRow.GetElement(2).Get<ui64>();
-            auto cookie = TStreamLookupJoinRowCookie::Decode(value);
+            auto cookie = TStreamLookupJoinRowCookie::Decode(value, Settings.CookieFormatVersion);
             rowSeqNo = cookie.RowSeqNo;
             firstRow = cookie.FirstRow;
             lastRow = cookie.LastRow;
@@ -671,7 +684,7 @@ public:
         ui64 joinKeyId = JoinKeySeqNo++;
         TOwnedCellVec cellVec(std::move(joinKeyCells));
         auto [resIt, _] = ResultRowsBySeqNo.emplace(
-            rowSeqNo, MakeIntrusive<TResultBatch>(KeepRowsOrder() ? nullptr : &FlushedResultRows, HolderFactory, rowSeqNo, std::move(row)));
+            rowSeqNo, MakeIntrusive<TResultBatch>(KeepRowsOrder() ? nullptr : &FlushedResultRows, HolderFactory, rowSeqNo, std::move(row), Settings.CookieFormatVersion));
         auto resultBatch = resIt->second;
         resultBatch->AcceptLeftRow(firstRow, lastRow);
         if (!IsKeyAllowed(cellVec)) {
@@ -1068,10 +1081,11 @@ private:
             TReadResultStats Stats;
         };
 
-        explicit TResultBatch(std::deque<TResultRow>* result, const NMiniKQL::THolderFactory& holderFactory, ui64 seqNo, TSizedUnboxedValue&& leftRow)
+        explicit TResultBatch(std::deque<TResultRow>* result, const NMiniKQL::THolderFactory& holderFactory, ui64 seqNo, TSizedUnboxedValue&& leftRow, ui32 cookieFormatVersion)
             : Result(result)
             , HolderFactory(holderFactory)
             , RowSeqNo(seqNo)
+            , CookieFormatVersion(cookieFormatVersion)
             , LeftRow(std::move(leftRow))
         {
         }
@@ -1085,6 +1099,7 @@ private:
         bool FirstRow = false;
         bool LastRow = false;
         ui64 RowSeqNo = 0;
+        ui32 CookieFormatVersion = 0;
 
         TSizedUnboxedValue LeftRow;
         TSizedUnboxedValue RightRow;
@@ -1134,7 +1149,7 @@ private:
                 resultRowItems[0] = LeftRow.Data;
                 resultRowItems[1] = std::move(RightRow.Data);
                 auto rowCookie = TStreamLookupJoinRowCookie{.RowSeqNo=RowSeqNo, .LastRow=ProcessedAllJoinKeys(), .FirstRow=FirstRowInBatch()};
-                resultRowItems[2] = NUdf::TUnboxedValuePod(rowCookie.Encode());
+                resultRowItems[2] = NUdf::TUnboxedValuePod(rowCookie.Encode(CookieFormatVersion));
 
                 rowStats.ReadRowsCount += (hasValue ? 1 : 0);
                 rowStats.ReadBytesCount += RightRow.StorageBytes;
@@ -1339,6 +1354,7 @@ std::unique_ptr<TKqpStreamLookupWorker> CreateStreamLookupWorker(NKikimrKqp::TKq
 
     preparedSettings.AllowNullKeysPrefixSize = settings.HasAllowNullKeysPrefixSize() ? settings.GetAllowNullKeysPrefixSize() : 0;
     preparedSettings.KeepRowsOrder = settings.HasKeepRowsOrder() && settings.GetKeepRowsOrder();
+    preparedSettings.CookieFormatVersion = settings.GetCookieFormatVersion();
     preparedSettings.LookupStrategy = settings.GetLookupStrategy();
 
     preparedSettings.KeyColumns.reserve(settings.GetKeyColumns().size());
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_worker.h b/ydb/core/kqp/runtime/kqp_stream_lookup_worker.h
index 348093c84f5..e582d7975ac 100644
--- a/ydb/core/kqp/runtime/kqp_stream_lookup_worker.h
+++ b/ydb/core/kqp/runtime/kqp_stream_lookup_worker.h
@@ -21,6 +21,7 @@ struct TLookupSettings {
 
     ui32 AllowNullKeysPrefixSize;
     bool KeepRowsOrder;
+    ui32 CookieFormatVersion = 0;
     NKqpProto::EStreamLookupStrategy LookupStrategy;
     std::unique_ptr<NKikimrKqp::TReadVectorTopK> VectorTopK;
 
diff --git a/ydb/core/kqp/ut/join/index_lookup/kqp_index_lookup_join_ut.cpp b/ydb/core/kqp/ut/join/index_lookup/kqp_index_lookup_join_ut.cpp
index 8b7456d2ad9..af98be5842f 100644
--- a/ydb/core/kqp/ut/join/index_lookup/kqp_index_lookup_join_ut.cpp
+++ b/ydb/core/kqp/ut/join/index_lookup/kqp_index_lookup_join_ut.cpp
@@ -1,4 +1,5 @@
 #include <ydb/core/kqp/ut/common/kqp_ut_common.h>
+#include <ydb/core/kqp/runtime/kqp_read_iterator_common.h>
 
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/proto/accessor.h>
 #include <library/cpp/json/json_reader.h>
@@ -1113,7 +1114,24 @@ Y_UNIT_TEST(StreamLookupJoin_RowSeqNoCollision_Repro) {
     // - stream index lookup join enabled (cookie-based sequencing)
     // - many DQ tasks (>= 17) producing overlapping RowSeqNo after Encode/Decode truncation
     // - LEFT join over a non-unique index producing multi-row sequences per left row
+    // Force the stream lookup actor to flush results in tiny fetches so that the
+    // First/middle/Last markers of different left-row sequences interleave heavily.
+    // This is what surfaces the RowSeqNo cookie collision as the crash in
+    // OmitRowLeftJoin (kqp_compute.cpp): Y_ENSURE(it != state.AllRowsAreNull.end()).
+    // The backoff settings are a process-global singleton, so restore the defaults
+    // when the test finishes to avoid leaking the tiny-fetch setting into other tests.
+    {
+        auto backoff = MakeIntrusive<TIteratorReadBackoffSettings>();
+        backoff->MaxRowsProcessingStreamLookup = 1;
+        SetReadIteratorBackoffSettings(backoff);
+    }
+    Y_DEFER {
+        SetReadIteratorBackoffSettings(MakeIntrusive<TIteratorReadBackoffSettings>());
+    };
+
     TKikimrSettings settings;
+    // Enable the fixed (v1) cookie format so RowSeqNo no longer overflows for taskId >= 16.
+    settings.AppConfig.MutableTableServiceConfig()->SetEnableStreamLookupJoinCookieV2(true);
     TKikimrRunner kikimr(settings);
     auto tableClient = kikimr.GetTableClient();
     auto session = tableClient.CreateSession().GetValueSync().GetSession();
diff --git a/ydb/core/protos/kqp.proto b/ydb/core/protos/kqp.proto
index 4e835249260..06833d84116 100644
--- a/ydb/core/protos/kqp.proto
+++ b/ydb/core/protos/kqp.proto
@@ -933,6 +933,10 @@ message TKqpStreamLookupSettings {
     optional string PoolId = 21;
     optional TReadVectorTopK VectorTopK = 19;
     optional uint64 QuerySpanId = 20;
+    // RowSeqNo cookie wire format version (0 = legacy, 1 = fixed bit budget).
+    // Gated by the EnableStreamLookupJoinCookieV2 feature flag so the new format
+    // is only emitted once every node in the cluster understands it.
+    optional uint32 CookieFormatVersion = 23 [default = 0];
 }
 
 message TKqpFullTextSourceSettings {
diff --git a/ydb/core/protos/kqp_physical.proto b/ydb/core/protos/kqp_physical.proto
index 00af8dc78f7..47a003467d7 100644
--- a/ydb/core/protos/kqp_physical.proto
+++ b/ydb/core/protos/kqp_physical.proto
@@ -329,6 +329,8 @@ message TKqpPhyCnStreamLookup {
     bool IsTableImmutable = 10;
     TKqpPhyVectorTopK VectorTopK = 11;
     repeated string InputColumns = 12;
+    // RowSeqNo cookie wire format version (0 = legacy, 1 = fixed bit budget).
+    uint32 CookieFormatVersion = 13;
 }
 
 message TKqpPhyCnVectorResolve {
diff --git a/ydb/core/protos/table_service_config.proto b/ydb/core/protos/table_service_config.proto
index 8b5402fe254..848671dba17 100644
--- a/ydb/core/protos/table_service_config.proto
+++ b/ydb/core/protos/table_service_config.proto
@@ -488,6 +488,12 @@ message TTableServiceConfig {
 
     optional bool EnableDqSourceStreamLookupJoin = 121 [default = false, (InvalidateCompileCache) = true];
 
+    // Emit the fixed (v1) stream lookup join RowSeqNo cookie format. Keep disabled
+    // until every node in the cluster is upgraded, then enable to fix RowSeqNo
+    // collisions across distributed tasks (taskId >= 16). InvalidateCompileCache
+    // ensures cached query plans are recompiled with the new format when toggled.
+    optional bool EnableStreamLookupJoinCookieV2 = 138 [default = false, (InvalidateCompileCache) = true];
+
     message TCompileCacheWarmupConfig {
         optional uint32 SoftDeadlineSeconds = 1 [default = 10];        // Soft deadline: stops new compilations, waits for in-flight ones to finish
         optional uint32 MaxConcurrentCompilations = 2 [default = 5];