Intermediate changes

commit_hash:8171ad7327107ff77cc8bbe11c24d3e4f2fd4c06

13 files changed, 196 insertions, 23 deletions

diff --git a/yt/yql/providers/yt/fmr/coordinator/impl/ut/ya.make b/yt/yql/providers/yt/fmr/coordinator/impl/ut/ya.make
index 8ca30f57288..6deb56f58d7 100644
--- a/yt/yql/providers/yt/fmr/coordinator/impl/ut/ya.make
+++ b/yt/yql/providers/yt/fmr/coordinator/impl/ut/ya.make
@@ -23,7 +23,6 @@ PEERDIR(
     yt/yql/providers/yt/fmr/test_tools/mock_time_provider
     yt/yql/providers/yt/fmr/test_tools/sorted_partitioner
     yt/yql/providers/yt/fmr/test_tools/table_data_service
-    yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl
     yt/yql/providers/yt/gateway/fmr
     yt/yql/providers/yt/gateway/file
     yt/yql/providers/yt/codec/codegen/llvm16
diff --git a/yt/yql/providers/yt/fmr/coordinator/operation_manager/impl/map_reduce/yql_yt_map_reduce_stage_operation_manager.cpp b/yt/yql/providers/yt/fmr/coordinator/operation_manager/impl/map_reduce/yql_yt_map_reduce_stage_operation_manager.cpp
index b919e13dea8..bb0c896898a 100644
--- a/yt/yql/providers/yt/fmr/coordinator/operation_manager/impl/map_reduce/yql_yt_map_reduce_stage_operation_manager.cpp
+++ b/yt/yql/providers/yt/fmr/coordinator/operation_manager/impl/map_reduce/yql_yt_map_reduce_stage_operation_manager.cpp
@@ -255,6 +255,9 @@ private:
             reduceTaskParams.Input = taskInput;
             reduceTaskParams.SerializedReduceJobState = operationParams.SerializedReduceJobState;
             reduceTaskParams.ReduceOperationSpec = operationParams.ReduceOperationSpec;
+            // The map stage above always shuffles via a synthetic _yql_key_hash prefix (see
+            // GenerateIntermediateTable), unlike a plain Reduce operation's real YT-sorted input.
+            reduceTaskParams.SortByHasKeyHashPrefix = true;
 
             std::vector<TFmrTableOutputRef> outputRefs;
             std::transform(
diff --git a/yt/yql/providers/yt/fmr/coordinator/partitioner/yql_yt_reduce_partitioner.cpp b/yt/yql/providers/yt/fmr/coordinator/partitioner/yql_yt_reduce_partitioner.cpp
index 8fd368852d8..69b9a40dc92 100644
--- a/yt/yql/providers/yt/fmr/coordinator/partitioner/yql_yt_reduce_partitioner.cpp
+++ b/yt/yql/providers/yt/fmr/coordinator/partitioner/yql_yt_reduce_partitioner.cpp
@@ -3,9 +3,19 @@
 #include "yql_yt_sorted_partitioner_base.h"
 
 #include <yt/yql/providers/yt/fmr/utils/comparator/yql_yt_binary_yson_compare_impl.h>
+#include <library/cpp/yson/node/node_io.h>
 
 namespace NYql::NFmr {
 
+namespace {
+
+// Binary YSON key rows may contain arbitrary non-UTF8 bytes; text YSON escapes them, keeping error messages valid UTF-8.
+TString FormatKeyRow(const TString& binaryYsonRow) {
+    return NYT::NodeToYsonString(NYT::NodeFromYsonString(binaryYsonRow));
+}
+
+} // namespace
+
 TReducePartitioner::TReducePartitioner(
     const std::unordered_map<TFmrTableId, std::vector<TString>>& partIdsForTables,
     const std::unordered_map<TString, std::vector<TChunkStats>>& partIdStats,
@@ -47,7 +57,7 @@ TFmrTableKeysRange TReducePartitioner::GetReadRangeFromSlices(const std::vector<
     YQL_ENSURE(taskRangeLeftBorderInclusive, " task range left border should always be true for reduce paritioner");
 
     if (taskRangeLeftBorder == taskRangeRightBorder && !taskRangeRightBorderInclusive) {
-        ythrow yexception() << "Key " << taskRangeLeftBorder.Row << " takes all slice and is too large, cannot partition";
+        ythrow yexception() << "Key " << FormatKeyRow(taskRangeLeftBorder.Row) << " takes all slice and is too large, cannot partition";
     }
 
     taskRange.SetFirstKeysBound(taskRangeLeftBorder, taskRangeLeftBorderInclusive);
@@ -93,7 +103,7 @@ void TReducePartitioner::CheckMaxKeySizePerSlices(const std::vector<TSlice>& sli
 
         if (*curLeftBorder != curRightBorder) {
             if (currentFilledSlicesWeight > maxKeySizePerPart) {
-                ythrow yexception() << "Key " << curLeftBorder->Row << " weighs at least " << currentFilledSlicesWeight << " bytes which is larget than maxKeySizePerPart - " << maxKeySizePerPart << " - cannot partition";
+                ythrow yexception() << "Key " << FormatKeyRow(curLeftBorder->Row) << " weighs at least " << currentFilledSlicesWeight << " bytes which is larget than maxKeySizePerPart - " << maxKeySizePerPart << " - cannot partition";
             }
             curLeftBorder = Nothing();
             currentFilledSlicesWeight = 0;
@@ -102,7 +112,7 @@ void TReducePartitioner::CheckMaxKeySizePerSlices(const std::vector<TSlice>& sli
         }
     }
     if (currentFilledSlicesWeight > maxKeySizePerPart) {
-        ythrow yexception() << "Key " << curLeftBorder->Row << " weighs at least " << currentFilledSlicesWeight << " bytes which is larget than maxKeySizePerPart - " << maxKeySizePerPart;
+        ythrow yexception() << "Key " << FormatKeyRow(curLeftBorder->Row) << " weighs at least " << currentFilledSlicesWeight << " bytes which is larget than maxKeySizePerPart - " << maxKeySizePerPart;
     }
     CurrentLastKeyWeight_ = currentFilledSlicesWeight;
 }
diff --git a/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.cpp b/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.cpp
index 6f4bc3e7d50..7d9c7a57b7f 100644
--- a/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.cpp
+++ b/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.cpp
@@ -113,6 +113,7 @@ void TFmrUserJob::Save(IOutputStream& s) const {
         ReduceOperationSpec_,
         IsMapReduceReducer_,
         IsMapReduceMap_,
+        SortByHasKeyHashPrefix_,
         ForceTableIndexMarking_
     );
 }
@@ -132,6 +133,7 @@ void TFmrUserJob::Load(IInputStream& s) {
         ReduceOperationSpec_,
         IsMapReduceReducer_,
         IsMapReduceMap_,
+        SortByHasKeyHashPrefix_,
         ForceTableIndexMarking_
     );
 }
@@ -283,6 +285,15 @@ void TFmrUserJob::FillQueueFromReduceInput() {
     YQL_ENSURE(ReduceOperationSpec_.Defined());
     auto reduceBy = ReduceOperationSpec_->ReduceBy;
     auto sortBy = ReduceOperationSpec_->SortBy;
+    // sortBy is [...reduceBy, ...tiebreaker] (e.g. a join's "_yql_sort"), with a leading
+    // _yql_key_hash only for MapReduce's hash-shuffled intermediate tables (SortByHasKeyHashPrefix_,
+    // set by the coordinator - see TReduceTaskParams::SortByHasKeyHashPrefix) - a plain Reduce
+    // reads its input pre-sorted by the real reduceBy columns, with no hash involved. Either
+    // way, task boundaries must only constrain on the reduce-key prefix (hash-if-present +
+    // reduceBy) - see the comment on CompareRowToBoundaryPrefix for why comparing the tiebreaker
+    // too would silently split a reduce group across tasks whenever a task cut lands between
+    // rows that share a reduce key but differ in tiebreaker.
+    const size_t numBoundaryKeyColumns = (SortByHasKeyHashPrefix_ ? 1 : 0) + reduceBy.Columns.size();
     for (const auto& inputTableRef : InputTables_.Inputs) {
         if (auto fmrInput = std::get_if<TFmrTableInputRef>(&inputTableRef)) {
             blockIterators.push_back(MakeIntrusive<TTableDataServiceBlockIterator>(
@@ -296,7 +307,9 @@ void TFmrUserJob::FillQueueFromReduceInput() {
                 fmrInput->IsFirstRowInclusive,
                 fmrInput->IsLastRowInclusive,
                 fmrInput->FirstRowKeys,
-                fmrInput->LastRowKeys
+                fmrInput->LastRowKeys,
+                /*readAheadChunks*/ ui64{4},
+                numBoundaryKeyColumns
             ));
         } else {
             ythrow TFmrNonRetryableJobException() << "YtTables unsupported inside Reduce task for now";
@@ -486,7 +499,8 @@ TStatistics TFmrUserJob::DoFmrJob(const TFmrUserJobOptions& options) {
             sortedWriter->NotifyRowEnd();
         }
         sortedWriter->Flush();
-        TStatistics result({{OutputTables_[0], sortedWriter->GetStats()}});
+        auto sortedWriterStats = sortedWriter->GetStats();
+        TStatistics result({{OutputTables_[0], sortedWriterStats}});
 
         // Direct (map-bypass) outputs at indices 1..K were already written to the real TDS via
         // their normal writers (see InitializeFmrUserJob) and flushed generically by
diff --git a/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.h b/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.h
index 54a8ff4ffde..a0bf4c70c07 100644
--- a/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.h
+++ b/yt/yql/providers/yt/fmr/process/yql_yt_job_fmr.h
@@ -116,6 +116,14 @@ public:
         IsMapReduceMap_ = value;
     }
 
+    // Whether this Reduce job's input is sorted by [_yql_key_hash, ...ReduceOperationSpec's
+    // SortBy] (a MapReduce operation's reduce stage) or by SortBy alone with no hash column at
+    // all (a plain Reduce operation reading real YT-sorted input) - see
+    // TReduceTaskParams::SortByHasKeyHashPrefix, which this is set from.
+    void SetSortByHasKeyHashPrefix(bool value) {
+        SortByHasKeyHashPrefix_ = value;
+    }
+
     // Set by the gateway (from execCtx->InputTables_, before this job's state is serialized to
     // workers) whenever the ORIGINAL OPERATION has more than one distinct input position (e.g. a
     // JOIN's mapper reading the same or different tables via multiple sections). This must be a
@@ -180,6 +188,7 @@ private:
     TMaybe<TReduceOperationSpec> ReduceOperationSpec_;
     bool IsMapReduceReducer_ = false;
     bool IsMapReduceMap_ = false;
+    bool SortByHasKeyHashPrefix_ = false;
     bool ForceTableIndexMarking_ = false;
     // End of serializable part
 
diff --git a/yt/yql/providers/yt/fmr/proto/request_options.proto b/yt/yql/providers/yt/fmr/proto/request_options.proto
index aacb3822b15..aafb3f2a32a 100644
--- a/yt/yql/providers/yt/fmr/proto/request_options.proto
+++ b/yt/yql/providers/yt/fmr/proto/request_options.proto
@@ -300,6 +300,11 @@ message TReduceTaskParams {
     repeated TFmrTableOutputRef Output = 2;
     bytes SerializedReduceJobState = 3;
     TReduceOperationSpec ReduceOperationSpec = 4;
+    // Whether Input's rows are sorted by [_yql_key_hash, ...ReduceOperationSpec.SortBy] (true for
+    // a MapReduce operation's reduce stage, which shuffles via a synthetic hash column) or by
+    // ReduceOperationSpec.SortBy alone (a plain Reduce operation, which reads real YT-sorted
+    // input with no hash column at all).
+    bool SortByHasKeyHashPrefix = 5;
 }
 
 message TSortOperationParams {
diff --git a/yt/yql/providers/yt/fmr/request_options/proto_helpers/yql_yt_request_proto_helpers.cpp b/yt/yql/providers/yt/fmr/request_options/proto_helpers/yql_yt_request_proto_helpers.cpp
index 855b572b486..c8f992255b6 100644
--- a/yt/yql/providers/yt/fmr/request_options/proto_helpers/yql_yt_request_proto_helpers.cpp
+++ b/yt/yql/providers/yt/fmr/request_options/proto_helpers/yql_yt_request_proto_helpers.cpp
@@ -717,6 +717,7 @@ NProto::TReduceTaskParams ReduceTaskParamsToProto(const TReduceTaskParams& reduc
     protoReduceTaskParams.SetSerializedReduceJobState(reduceTaskParams.SerializedReduceJobState);
     auto protoReduceOperationSpec = ReduceOperationSpecToProto(reduceTaskParams.ReduceOperationSpec);
     protoReduceTaskParams.MutableReduceOperationSpec()->Swap(&protoReduceOperationSpec);
+    protoReduceTaskParams.SetSortByHasKeyHashPrefix(reduceTaskParams.SortByHasKeyHashPrefix);
     return protoReduceTaskParams;
 }
 
@@ -730,6 +731,7 @@ TReduceTaskParams ReduceTaskParamsFromProto(const NProto::TReduceTaskParams& pro
     reduceTaskParams.Output = outputTables;
     reduceTaskParams.SerializedReduceJobState = protoReduceTaskParams.GetSerializedReduceJobState();
     reduceTaskParams.ReduceOperationSpec = ReduceOperationSpecFromProto(protoReduceTaskParams.GetReduceOperationSpec());
+    reduceTaskParams.SortByHasKeyHashPrefix = protoReduceTaskParams.GetSortByHasKeyHashPrefix();
     return reduceTaskParams;
 }
 
diff --git a/yt/yql/providers/yt/fmr/request_options/yql_yt_request_options.h b/yt/yql/providers/yt/fmr/request_options/yql_yt_request_options.h
index a32e521ba36..a2694941ae3 100644
--- a/yt/yql/providers/yt/fmr/request_options/yql_yt_request_options.h
+++ b/yt/yql/providers/yt/fmr/request_options/yql_yt_request_options.h
@@ -538,6 +538,11 @@ struct TReduceTaskParams {
     std::vector<TFmrTableOutputRef> Output;
     TString SerializedReduceJobState;
     TReduceOperationSpec ReduceOperationSpec;
+    // Whether Input's rows are sorted by [_yql_key_hash, ...ReduceOperationSpec.SortBy] (true for
+    // a MapReduce operation's reduce stage, which shuffles via a synthetic hash column) or by
+    // ReduceOperationSpec.SortBy alone (a plain Reduce operation, which reads real YT-sorted
+    // input with no hash column at all).
+    bool SortByHasKeyHashPrefix = false;
 };
 
 // Service column added by map stage to hash-route rows to the correct reducer.
diff --git a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ut/ya.make b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ut/ya.make
new file mode 100644
index 00000000000..61a5d2ff9f6
--- /dev/null
+++ b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ut/ya.make
@@ -0,0 +1,17 @@
+UNITTEST()
+
+SRCS(
+    yql_yt_yson_tds_block_iterator_ut.cpp
+)
+
+PEERDIR(
+    yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl
+    yt/yql/providers/yt/fmr/request_options
+    yt/yql/providers/yt/fmr/table_data_service/local/impl
+    yt/yql/providers/yt/fmr/test_tools/yson
+    yt/yql/providers/yt/fmr/utils
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()
diff --git a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ut/yql_yt_yson_tds_block_iterator_ut.cpp b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ut/yql_yt_yson_tds_block_iterator_ut.cpp
new file mode 100644
index 00000000000..e2f92d30156
--- /dev/null
+++ b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ut/yql_yt_yson_tds_block_iterator_ut.cpp
@@ -0,0 +1,88 @@
+#include <library/cpp/testing/unittest/registar.h>
+
+#include <yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.h>
+#include <yt/yql/providers/yt/fmr/test_tools/yson/yql_yt_yson_helpers.h>
+#include <yt/yql/providers/yt/fmr/table_data_service/local/impl/yql_yt_table_data_service_local.h>
+#include <yt/yql/providers/yt/fmr/utils/yql_yt_table_data_service_key.h>
+
+namespace NYql::NFmr {
+
+namespace {
+
+// Two rows sharing the same reduce key (_yql_key_hash=1, k=5) but differing only in the
+// tiebreaker column _yql_sort - e.g. a join's build-side row (sort=0) and one of its probe-side
+// rows (sort=1). Mirrors the real production row shape from yt/yql/providers/yt/fmr/job/impl/
+// yql_yt_reduce_reader.cpp's dumps.
+const TVector<TString> KeyColumns = {"_yql_key_hash", "k", "_yql_sort"};
+
+std::vector<TIndexedBlock> ReadAllBlocks(TTableDataServiceBlockIterator::TPtr iter) {
+    std::vector<TIndexedBlock> blocks;
+    TIndexedBlock block;
+    while (iter->NextBlock(block)) {
+        blocks.push_back(block);
+    }
+    return blocks;
+}
+
+ui64 CountRows(const std::vector<TIndexedBlock>& blocks) {
+    ui64 total = 0;
+    for (const auto& b : blocks) {
+        total += b.Rows.size();
+    }
+    return total;
+}
+
+} // namespace
+
+Y_UNIT_TEST_SUITE(TableDataServiceBlockIteratorBoundaryTests) {
+    Y_UNIT_TEST(TieBreakerColumnMustNotSplitReduceGroupAcrossBoundary) {
+        auto tds = MakeLocalTableDataService();
+        const TString tableId = "t";
+        const TString partId = "p0";
+        const TString group = GetTableDataServiceGroup(tableId, partId);
+        const TString chunkId = GetTableDataServiceChunkId(0, TString());
+
+        const TString chunkText =
+            "{\"_yql_key_hash\"=1u;\"k\"=5u;\"_yql_sort\"=0u};\n"
+            "{\"_yql_key_hash\"=1u;\"k\"=5u;\"_yql_sort\"=1u};\n";
+        tds->Put(group, chunkId, GetBinaryYson(chunkText)).GetValueSync();
+
+        // The task boundary itself is built from the sort=0 row's real key - exactly what
+        // happens in production when that row is the last physical row before a task cut.
+        const TString boundaryRow = GetBinaryYson(
+            "{\"_yql_key_hash\"=1u;\"k\"=5u;\"_yql_sort\"=0u}",
+            NYson::EYsonType::Node
+        );
+
+        std::vector<TTableRange> tableRanges = {TTableRange{.PartId = partId, .MinChunk = 0, .MaxChunk = 1}};
+        std::vector<ESortOrder> sortOrders(KeyColumns.size(), ESortOrder::Ascending);
+
+        // Without a cap (numBoundaryKeyColumns defaults to comparing every key column, including
+        // the tiebreaker): reproduces the bug - the sort=1 row looks like it's past the boundary
+        // and gets dropped, even though it shares the exact same reduce key.
+        {
+            auto iter = MakeIntrusive<TTableDataServiceBlockIterator>(
+                tableId, tableRanges, tds, KeyColumns, sortOrders,
+                /*neededColumns*/ TVector<TString>{}, /*serializedColumnGroupsSpec*/ TString(),
+                /*isFirstRowKeysInclusive*/ Nothing(), /*isLastRowKeysInclusive*/ true,
+                /*firstRowKeys*/ Nothing(), /*lastRowKeys*/ boundaryRow
+            );
+            UNIT_ASSERT_VALUES_EQUAL(CountRows(ReadAllBlocks(iter)), 1);
+        }
+
+        // With the fix - numBoundaryKeyColumns=2 caps comparison to [_yql_key_hash, k], excluding
+        // _yql_sort: both rows of the reduce group survive the same boundary.
+        {
+            auto iter = MakeIntrusive<TTableDataServiceBlockIterator>(
+                tableId, tableRanges, tds, KeyColumns, sortOrders,
+                /*neededColumns*/ TVector<TString>{}, /*serializedColumnGroupsSpec*/ TString(),
+                /*isFirstRowKeysInclusive*/ Nothing(), /*isLastRowKeysInclusive*/ true,
+                /*firstRowKeys*/ Nothing(), /*lastRowKeys*/ boundaryRow,
+                /*readAheadChunks*/ ui64{4}, /*numBoundaryKeyColumns*/ size_t{2}
+            );
+            UNIT_ASSERT_VALUES_EQUAL(CountRows(ReadAllBlocks(iter)), 2);
+        }
+    }
+}
+
+} // namespace NYql::NFmr
diff --git a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ya.make b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ya.make
index afb3020f720..7f68fea4899 100644
--- a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ya.make
+++ b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/ya.make
@@ -21,3 +21,7 @@ PEERDIR(
 YQL_LAST_ABI_VERSION()
 
 END()
+
+RECURSE_FOR_TESTS(
+    ut
+)
diff --git a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.cpp b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.cpp
index 584614844d9..e87a9791cbb 100644
--- a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.cpp
+++ b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.cpp
@@ -21,7 +21,8 @@ TTableDataServiceBlockIterator::TTableDataServiceBlockIterator(
     TMaybe<bool> isLastRowKeysInclusive,
     TMaybe<TString> firstRowKeys,
     TMaybe<TString> lastRowKeys,
-    ui64 readAheadChunks
+    ui64 readAheadChunks,
+    TMaybe<size_t> numBoundaryKeyColumns
 )
     : TableId_(std::move(tableId))
     , TableRanges_(std::move(tableRanges))
@@ -31,6 +32,7 @@ TTableDataServiceBlockIterator::TTableDataServiceBlockIterator(
     , NeededColumns_(std::move(neededColumns))
     , SerializedColumnGroupsSpec_(std::move(serializedColumnGroupsSpec))
     , ReadAheadChunks_(readAheadChunks)
+    , NumBoundaryKeyColumns_(numBoundaryKeyColumns.GetOrElse(KeyColumns_.size()))
 {
     if (SortOrders_.empty()) {
         SortOrders_.assign(KeyColumns_.size(), ESortOrder::Ascending);
@@ -38,6 +40,7 @@ TTableDataServiceBlockIterator::TTableDataServiceBlockIterator(
     if (SortOrders_.size() != KeyColumns_.size()) {
         ythrow yexception() << "SortOrders and KeyColumns sizes are different";
     }
+    Y_ENSURE(NumBoundaryKeyColumns_ <= KeyColumns_.size(), "numBoundaryKeyColumns must not exceed keyColumns size");
 
     if (firstRowKeys) {
         FirstBoundary_ = TFmrTableKeysBoundary(*firstRowKeys, KeyColumns_, SortOrders_);
@@ -144,24 +147,29 @@ bool IsNullColumnValue(TStringBuf blob, const TColumnOffsetRange& range) {
     return SliceRange(blob, range) == TStringBuf(&EntitySymbol, 1);
 }
 
-// FirstRowKeys/LastRowKeys boundaries are built by the coordinator from a task's chunk-stats
-// key columns (e.g. [_yql_key_hash, ...ReduceBy] for MapReduce reduce tasks), which can be a
-// strict PREFIX of the fuller key column set (e.g. [_yql_key_hash, ...SortBy]) this iterator
-// parses actual rows with. TFmrTableKeysBoundary parses its own Row bytes with the SAME
-// (fuller) key columns as the row markup, so any column beyond the boundary's real prefix comes
-// out with an IsValid()==false range - not because the value is legitimately null, but because
-// the boundary blob simply never encoded it. Comparing that as "boundary < row" (the usual
-// null-handling rule) would incorrectly exclude real rows that land exactly on the boundary, so
-// this comparison stops at the boundary's own valid prefix instead of delegating to the
-// generic null-aware CompareKeyRowsAcrossYsonBlocks.
+// FirstRowKeys/LastRowKeys boundaries are built by the coordinator from a task's chunk-stats,
+// i.e. from an actual physical row's real key - for MapReduce reduce tasks over joins that key
+// is [_yql_key_hash, ...ReduceBy, ...SortBy tiebreaker] in full, tiebreaker included, because
+// that's what TReducePartitioner sorts intermediate data by (see
+// yql_yt_map_reduce_stage_operation_manager.cpp). It is NOT reliably a strict prefix that omits
+// the tiebreaker: whichever physical row happens to land on a task cut contributes its own real
+// tiebreaker value to the boundary blob. So this can't stop at "wherever the boundary blob runs
+// out of valid columns" (that already-invalid range doesn't reliably appear here) - it must be
+// told explicitly, via numBoundaryKeyColumns, how many leading columns actually define a reduce
+// group, and ignore everything past that. Comparing the tiebreaker as a real bound column would
+// silently split a reduce group in half whenever a task cut lands between two physical rows that
+// share a reduce key but differ in tiebreaker (e.g. a join's build-side row vs. its probe-side
+// rows) - each side would look like it's on the "wrong side" of the other's boundary.
 int CompareRowToBoundaryPrefix(
     TStringBuf rowBlob,
     const TRowIndexMarkup& row,
     TStringBuf boundaryBlob,
     const TRowIndexMarkup& boundaryMarkup,
-    const std::vector<ESortOrder>& sortOrders
+    const std::vector<ESortOrder>& sortOrders,
+    size_t numBoundaryKeyColumns
 ) {
-    for (ui64 colIdx = 0; colIdx + 1 < boundaryMarkup.size(); ++colIdx) {
+    const size_t limit = std::min<size_t>(numBoundaryKeyColumns, boundaryMarkup.empty() ? 0 : boundaryMarkup.size() - 1);
+    for (ui64 colIdx = 0; colIdx < limit; ++colIdx) {
         const auto& boundaryRange = boundaryMarkup[colIdx];
         if (!boundaryRange.IsValid()) {
             // Boundary doesn't encode this column (or any further one, since it's a prefix) -
@@ -204,7 +212,8 @@ bool TTableDataServiceBlockIterator::RowInKeyBounds(const TString& blob, const T
             row,
             FirstBoundary_->Row,
             FirstBoundary_->Markup,
-            SortOrders_
+            SortOrders_,
+            NumBoundaryKeyColumns_
         );
         if (c < 0) { // if row < first boundary
             return false;
@@ -218,7 +227,8 @@ bool TTableDataServiceBlockIterator::RowInKeyBounds(const TString& blob, const T
             row,
             LastBoundary_->Row,
             LastBoundary_->Markup,
-            SortOrders_
+            SortOrders_,
+            NumBoundaryKeyColumns_
         );
         if (c > 0) { // if row > last boundary
             return false;
diff --git a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.h b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.h
index 4ab202869fc..8f661e5ec01 100644
--- a/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.h
+++ b/yt/yql/providers/yt/fmr/utils/yson_block_iterator/impl/yql_yt_yson_tds_block_iterator.h
@@ -27,8 +27,14 @@ public:
         TMaybe<bool> isLastRowKeysInclusive = Nothing(),
         TMaybe<TString> firstRowKeys = Nothing(),
         TMaybe<TString> lastRowKeys = Nothing(),
-        ui64 readAheadChunks = 4
-
+        ui64 readAheadChunks = 4,
+        // How many leading keyColumns actually define a reduce group (e.g. _yql_key_hash +
+        // reduceBy, excluding a join's _yql_sort tiebreaker). Boundary comparisons stop after
+        // this many columns instead of running to the end of keyColumns - see the comment on
+        // CompareRowToBoundaryPrefix for why relying on the boundary blob happening to omit
+        // trailing columns is not safe. Defaults to comparing every column in keyColumns, which
+        // is correct whenever keyColumns has no trailing tiebreaker (e.g. non-reduce contexts).
+        TMaybe<size_t> numBoundaryKeyColumns = Nothing()
     );
 
     ~TTableDataServiceBlockIterator() final;
@@ -73,6 +79,7 @@ private:
     std::vector<TString> GroupNamesToRead_;
 
     ui64 ReadAheadChunks_ = 4;
+    const size_t NumBoundaryKeyColumns_;
     std::deque<TPrefetchEntry> PrefetchQueue_;
 };