[fix](binlog) Fix BE crash when APPEND_ONLY row-binlog scan omits key columns

be/src/exec/scan/olap_scanner.cpp

@@ -476,13 +476,14 @@ Status OlapScanner::_init_tablet_reader_params(
         }
     };
 
-    // For row-binlog scans that emit BEFORE/AFTER pairs (MIN_DELTA / DETAIL), we must read
-    // every key column, every requested value column, the binlog meta columns (tso / op)
-    // and their __BEFORE__ mirrors, so the BlockReader can reconstruct change rows.
-    const bool need_before_columns =
+    // MIN_DELTA / DETAIL row-binlog scans reconstruct change rows in BlockReader through a
+    // key-ordered merge. They must read every key column, every requested value column, the
+    // binlog meta columns (tso / op) and their __BEFORE__ mirrors. APPEND_ONLY streams rows
+    // as-is and stays on the plain projection paths below.
+    const bool is_binlog_merge_scan =
             _tablet_reader_params.binlog_scan_type == TBinlogScanType::MIN_DELTA ||
             _tablet_reader_params.binlog_scan_type == TBinlogScanType::DETAIL;
-    if (need_before_columns) {
+    if (is_binlog_merge_scan) {
         for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) {
             add_return_column_if_absent(static_cast<uint32_t>(i));
         }
@@ -565,16 +566,26 @@ Status OlapScanner::_init_tablet_reader_params(
 
     RETURN_IF_ERROR(_init_tso_pushdown());
 
-    // For any row-binlog scan, force the storage layer to deliver rows strictly in primary-key
-    // order so the BlockReader can group consecutive same-key changes (MIN_DELTA) or emit
-    // BEFORE/AFTER pairs in deterministic order (DETAIL). Disable ORDER BY / TopN pushdowns
-    // and reset their related params, since they would otherwise re-order the stream.
+    // Row-binlog scans must not be re-ordered or truncated by ORDER BY / TopN pushdowns,
+    // so reset every reorder-related param for all binlog scan types.
+    //
+    // Only MIN_DELTA / DETAIL additionally force the storage layer to deliver rows strictly
+    // in primary-key order, so the BlockReader can group consecutive same-key changes
+    // (MIN_DELTA) or emit BEFORE/AFTER pairs in deterministic order (DETAIL). Their storage
+    // projection is widened above with the full key prefix, which the key-ordered merge
+    // comparator relies on: with read_orderby_key_num_prefix_columns == 0 the comparator
+    // falls back to comparing the first num_key_columns block positions.
+    //
+    // APPEND_ONLY does no key grouping and keeps the raw SQL projection, which may omit
+    // some or even all key columns. Forcing a key-ordered merge would make the fallback
+    // comparator read key positions that do not exist in the projected blocks and crash
+    // the BE (issue #66390), so it reads unordered like a plain scan.
     if (_tablet_reader_params.binlog_scan_type != TBinlogScanType::NONE) {
-        _tablet_reader_params.read_orderby_key = true;
+        _tablet_reader_params.read_orderby_key = is_binlog_merge_scan;
+        _tablet_reader_params.force_key_ordered_read = is_binlog_merge_scan;
         _tablet_reader_params.read_orderby_key_reverse = false;
         _tablet_reader_params.read_orderby_key_num_prefix_columns = 0;
         _tablet_reader_params.read_orderby_key_limit = 0;
-        _tablet_reader_params.force_key_ordered_read = true;
         _tablet_reader_params.topn_filter_source_node_ids.clear();
     }
 

be/src/storage/iterator/vcollect_iterator.cpp

@@ -709,6 +709,8 @@ Status VCollectIterator::Level1Iterator::init(bool get_data_by_ref) {
             }
         }
 
+        RETURN_IF_ERROR(_validate_merge_compare_contract(sequence_loc));
+
         _heap = std::make_unique<MergeHeap>(LevelIteratorComparator(
                 sequence_loc, _is_reverse, _reader->_reader_context.use_insert_order_when_same,
                 tso_col_id >= 0));
@@ -766,6 +768,61 @@ void VCollectIterator::Level1Iterator::init_level0_iterators_for_union() {
     }
 }
 
+// LevelIteratorComparator reads block positions that carry no bounds checks in release
+// builds: either the explicit compare-column positions, or positions
+// [0, tablet_schema().num_key_columns()) plus the sequence tie-break position. A read
+// projection that omits or reorders the leading key columns would turn the first heap
+// comparison into an out-of-bounds or semantically wrong positional access (issue #66390).
+// Validate the contract against every child's first block before anything enters _heap.
+Status VCollectIterator::Level1Iterator::_validate_merge_compare_contract(int sequence_loc) const {
+    const auto& return_columns = _reader->_return_columns;
+    const size_t num_key_columns = _schema.num_key_columns();
+    for (const auto& child : _children) {
+        const IteratorRowRef* ref = child->current_row_ref();
+        if (ref->block == nullptr) {
+            continue;
+        }
+        const size_t block_columns = ref->block->columns();
+        auto contract_error = [&](const std::string& detail) {
+            std::string projected_ids;
+            for (auto cid : return_columns) {
+                if (!projected_ids.empty()) {
+                    projected_ids += ',';
+                }
+                projected_ids += std::to_string(cid);
+            }
+            return Status::InternalError(
+                    "merge heap compare contract violated: {}, tablet_id={}, block_columns={}, "
+                    "num_key_columns={}, sequence_loc={}, return_columns=[{}]",
+                    detail, _reader->_tablet->tablet_id(), block_columns, num_key_columns,
+                    sequence_loc, projected_ids);
+        };
+        if (_compare_columns != nullptr) {
+            for (uint32_t pos : *_compare_columns) {
+                if (pos >= block_columns) {
+                    return contract_error(
+                            fmt::format("compare column position {} out of range", pos));
+                }
+            }
+        } else {
+            if (num_key_columns > return_columns.size() || num_key_columns > block_columns) {
+                return contract_error("projection has fewer columns than the key prefix");
+            }
+            for (size_t i = 0; i < num_key_columns; ++i) {
+                if (return_columns[i] != i) {
+                    return contract_error(
+                            fmt::format("position {} holds column id {} instead of key column {}",
+                                        i, return_columns[i], i));
+                }
+            }
+        }
+        if (sequence_loc != -1 && static_cast<size_t>(sequence_loc) >= block_columns) {
+            return contract_error("sequence column position out of range");
+        }
+    }
+    return Status::OK();
+}
+
 Status VCollectIterator::Level1Iterator::_merge_next(IteratorRowRef* ref) {
     auto res = _cur_child->next(ref);
     if (LIKELY(res.ok())) {

be/src/storage/iterator/vcollect_iterator.h

@@ -324,6 +324,13 @@ class VCollectIterator {
         bool collected_enough_rows(const MutableColumns& columns, int rows_to_merge) const;
 
     private:
+        // Validate that every block position LevelIteratorComparator may touch exists in
+        // each child's current block and, for the default key-prefix comparison, that the
+        // read projection really starts with the full ordered key prefix. Called before
+        // any child is pushed into _heap, so a broken projection surfaces as an error
+        // instead of an out-of-bounds positional access (issue #66390).
+        Status _validate_merge_compare_contract(int sequence_loc) const;
+
         Status _merge_next(IteratorRowRef* ref);
 
         Status _normal_next(IteratorRowRef* ref);

be/src/storage/iterator/vgeneric_iterators.cpp

@@ -296,13 +296,66 @@ Status VMergeIteratorContext::init(const StorageReadOptions& opts) {
     _record_rowids = opts.record_rowids;
     RETURN_IF_ERROR(_load_next_block());
     if (valid()) {
+        RETURN_IF_ERROR(_validate_compare_contract(opts));
         RETURN_IF_ERROR(advance());
     }
     _pre_ctx_same_bit.reserve(_block_row_max);
     _pre_ctx_same_bit.assign(_block_row_max, false);
     return Status::OK();
 }
 
+// compare() reads block positions that are only DCHECK-bounds-checked in Block::compare_at(),
+// so in a release build a projection violating the merge contract turns into an out-of-bounds
+// read inside std::push_heap and kills the BE (issue #66390). Verify the contract once the
+// first block is loaded and surface a diagnosable error instead:
+//   - explicit compare columns (_compare_columns) must all point inside the block;
+//   - otherwise the default comparison touches positions [0, _num_key_columns), where
+//     _num_key_columns counts the key columns of the WHOLE tablet schema. Key columns always
+//     occupy column ids [0, num_key_columns) of the tablet schema, so the projection must
+//     start with exactly those ids, in order, for the positional comparison to be key order;
+//   - the sequence tie-break column, when present, must point inside the block as well.
+Status VMergeIteratorContext::_validate_compare_contract(const StorageReadOptions& opts) const {
+    const size_t block_columns = _block->columns();
+    auto contract_error = [&](const std::string& detail) {
+        std::string projected_ids;
+        for (auto cid : _output_schema->column_ids()) {
+            if (!projected_ids.empty()) {
+                projected_ids += ',';
+            }
+            projected_ids += std::to_string(cid);
+        }
+        return Status::InternalError(
+                "merge iterator compare contract violated: {}, tablet_id={}, rowset_id={}, "
+                "version={}, block_columns={}, num_key_columns={}, sequence_id_idx={}, "
+                "projected_column_ids=[{}]",
+                detail, opts.tablet_id, opts.rowset_id.to_string(), opts.version.to_string(),
+                block_columns, _num_key_columns, _sequence_id_idx, projected_ids);
+    };
+    if (_compare_columns != nullptr) {
+        for (uint32_t pos : *_compare_columns) {
+            if (pos >= block_columns) {
+                return contract_error(fmt::format("compare column position {} out of range", pos));
+            }
+        }
+    } else {
+        const auto num_key_columns = static_cast<size_t>(_num_key_columns);
+        if (num_key_columns > _output_schema->num_column_ids() || num_key_columns > block_columns) {
+            return contract_error("projection has fewer columns than the key prefix");
+        }
+        for (size_t i = 0; i < num_key_columns; ++i) {
+            if (_output_schema->column_ids()[i] != static_cast<ColumnId>(i)) {
+                return contract_error(
+                        fmt::format("position {} holds column id {} instead of key column {}", i,
+                                    _output_schema->column_ids()[i], i));
+            }
+        }
+    }
+    if (_sequence_id_idx != -1 && static_cast<size_t>(_sequence_id_idx) >= block_columns) {
+        return contract_error("sequence column position out of range");
+    }
+    return Status::OK();
+}
+
 Status VMergeIteratorContext::advance() {
     _skip = false;
     _same = false;

be/src/storage/iterator/vgeneric_iterators.h

@@ -185,6 +185,12 @@ class VMergeIteratorContext {
     // Load next block into _block
     Status _load_next_block();
 
+    // Validate that every block position compare() may touch actually exists in _block
+    // and, for the default key-prefix comparison, that the projection really starts with
+    // the full ordered key prefix. Returns an error instead of letting compare() perform
+    // an out-of-bounds or semantically wrong positional access (issue #66390).
+    Status _validate_compare_contract(const StorageReadOptions& opts) const;
+
     RowwiseIteratorUPtr _iter;
 
     int _sequence_id_idx = -1;

be/test/exec/scan/vgeneric_iterators_test.cpp

@@ -42,7 +42,7 @@ class VGenericIteratorsTest : public testing::Test {
     virtual ~VGenericIteratorsTest() {}
 };
 
-static Schema create_schema() {
+static std::vector<TabletColumnPtr> create_col_schemas() {
     std::vector<TabletColumnPtr> col_schemas;
     auto c1 = std::make_shared<TabletColumn>(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
                                              FieldType::OLAP_FIELD_TYPE_SMALLINT, true);
@@ -57,6 +57,11 @@ static Schema create_schema() {
     col_schemas.emplace_back(
             std::make_shared<TabletColumn>(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM,
                                            FieldType::OLAP_FIELD_TYPE_BIGINT, true));
+    return col_schemas;
+}
+
+static Schema create_schema() {
+    std::vector<TabletColumnPtr> col_schemas = create_col_schemas();
 
     std::vector<ColumnId> column_ids(col_schemas.size());
     for (uint32_t cid = 0; cid < column_ids.size(); ++cid) {
@@ -402,4 +407,100 @@ TEST(VGenericIteratorsTest, MergeWithSeqColumnSmallSeqFirst) {
     EXPECT_EQ(0, actual_value);
 }
 
+// Emits num_rows rows for a schema whose projection (col_ids) may be narrower than the
+// full tablet schema. The filled block layout follows the projection, matching how
+// VMergeIteratorContext::block_reset builds its block from the output schema.
+class ProjectedColumnsUtIterator : public RowwiseIterator {
+public:
+    ProjectedColumnsUtIterator(Schema schema, size_t num_rows)
+            : _schema(std::move(schema)), _num_rows(num_rows) {}
+    ~ProjectedColumnsUtIterator() override = default;
+
+    Status init(const StorageReadOptions& opts) override { return Status::OK(); }
+
+    Status next_batch(Block* block) override {
+        if (_rows_returned >= _num_rows) {
+            return Status::EndOfFile("End of ProjectedColumnsUtIterator");
+        }
+        while (_rows_returned < _num_rows) {
+            for (size_t j = 0; j < _schema.num_column_ids(); ++j) {
+                ColumnWithTypeAndName& vc = block->get_by_position(j);
+                IColumn& vi = (IColumn&)(*vc.column);
+
+                char data[16] = {};
+                size_t data_len = 0;
+                const auto* col_schema = _schema.column(_schema.column_id(j));
+                switch (col_schema->type()) {
+                case FieldType::OLAP_FIELD_TYPE_SMALLINT:
+                    *(int16_t*)data = static_cast<int16_t>(_rows_returned);
+                    data_len = sizeof(int16_t);
+                    break;
+                case FieldType::OLAP_FIELD_TYPE_INT:
+                    *(int32_t*)data = static_cast<int32_t>(_rows_returned);
+                    data_len = sizeof(int32_t);
+                    break;
+                case FieldType::OLAP_FIELD_TYPE_BIGINT:
+                    *(int64_t*)data = static_cast<int64_t>(_rows_returned);
+                    data_len = sizeof(int64_t);
+                    break;
+                default:
+                    break;
+                }
+
+                vi.insert_data(data, data_len);
+            }
+            ++_rows_returned;
+        }
+        return Status::OK();
+    }
+
+    const Schema& schema() const override { return _schema; }
+
+private:
+    Schema _schema;
+    size_t _num_rows;
+    size_t _rows_returned = 0;
+};
+
+// The merge-heap comparator compares the first num_key_columns block positions when no
+// explicit compare columns are given, and num_key_columns counts the key columns of the
+// WHOLE tablet schema. A projection narrower than the key prefix must be rejected at
+// init time with an error instead of crashing inside std::push_heap (issue #66390: a
+// ROW-binlog APPEND_ONLY scan projected value columns only).
+TEST(VGenericIteratorsTest, MergeRejectsProjectionMissingKeyPrefix) {
+    // Full tablet schema: k0(smallint), k1(int), v2(bigint); project only v2.
+    Schema projected(create_col_schemas(), std::vector<ColumnId> {2});
+    auto output_schema = std::make_shared<Schema>(projected);
+
+    std::vector<RowwiseIteratorUPtr> inputs;
+    inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected, 10));
+    inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected, 10));
+
+    auto iter = new_merge_iterator(std::move(inputs), -1, false, false, nullptr, output_schema);
+    StorageReadOptions opts;
+    auto st = iter->init(opts);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("compare contract violated") != std::string::npos)
+            << st.to_string();
+}
+
+// Same as above, but the projection has enough columns while not starting with the full
+// ordered key prefix (k0 is missing): position 0 would be compared as if it were k0.
+TEST(VGenericIteratorsTest, MergeRejectsProjectionWithoutLeadingKey) {
+    // Full tablet schema: k0(smallint), k1(int), v2(bigint); project {k1, v2}.
+    Schema projected(create_col_schemas(), std::vector<ColumnId> {1, 2});
+    auto output_schema = std::make_shared<Schema>(projected);
+
+    std::vector<RowwiseIteratorUPtr> inputs;
+    inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected, 10));
+    inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected, 10));
+
+    auto iter = new_merge_iterator(std::move(inputs), -1, false, false, nullptr, output_schema);
+    StorageReadOptions opts;
+    auto st = iter->init(opts);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("compare contract violated") != std::string::npos)
+            << st.to_string();
+}
+
 } // namespace doris