[refactor](storage) extract MowKeyProbe and HistoricalRowFetcher; segment writers and row-binlog retriever delegate probe, row-cache invalidation and old-value reads

Part 2/6 of splitting #64674 (block transform chain) into independently
reviewable PRs. Part 1 was #65492 (RowKeyEncoder).

The merge-on-write primary key probe - look the key up in the load's rowset
snapshot, mark the delete bitmap, decide whether the old row has to be read for
the missing columns - existed in three near-verbatim copies:
SegmentWriter::probe_key_for_mow, VerticalSegmentWriter::_probe_key_for_mow and
PrimaryKeyModelRowRetriever. The key encoding around it and the row-cache
invalidation were duplicated the same way. The three copies differ in exactly
three rules, all of which were expressed by copy-editing the code.

Extract them into storage/mow/:

- MowKeyProbe: one probe behind a Policy (delete_bitmap_mode READ_ONLY /
  OLD_ROW / OLD_AND_NEW_ROW, skip_delete_sign, skip_seq_loses,
  skip_in_load_deleted), plus probe_previous_seq_value(),
  maybe_invalidate_row_cache() and encode_mow_key_invalidate_cache(), which
  encodes a key, invalidates the row cache and appends the sequence suffix in
  the one order that is correct.
- HistoricalRowFetcher: owns the rowset pins and the read plans behind
  FixedReadPlan / FlexibleReadPlan.

All three call sites delegate. Both writers keep a thin wrapper that turns a
ProbeOutcome back into the out-parameters their fill loops use and run their
fixed partial-update fill on a per-call HistoricalRowFetcher; the wrappers are
bridge code, deleted when the fills move into chain stages.
PrimaryKeyModelRowRetriever drops its own encode / probe / row-cache copies and
holds a RowKeyEncoder plus a HistoricalRowFetcher, still driven by the existing
RowBinlogSegmentWriter so the row-binlog regressions cover the rework.

RowKeyEncoder gains a second constructor for the probe side: the key the primary
key index is built on is always the schema key columns, while the writer's
constructor builds whatever the segment sorts by (cluster keys when it has them)
plus the separate primary-key view and the rowid suffix. The retriever is the
first caller that has to encode primary keys for any mow table, with or without
cluster keys.

Equivalence notes: the Policy reproduces all three old copies term for term
(delete-bitmap keys, marked rows and stat counters unchanged); SegmentWriter's
cast_set<uint32_t>(max_version) truncation is dropped in favour of the int64
behavior the other two callers already had; lookup_row_key is called with an
explicit with_rowid=false, which fixes the retriever on cluster-key mow tables
(the default strips ROW_ID_LENGTH bytes off a probe key that carries no rowid
suffix) and is inert for partial update, which FE forbids on those tables;
num_rows_new_added is bumped inside probe(), before the writers' handle_new_key
runs at the call site; fixed-path rowset pins move from writer-lifetime maps to
the per-call fetcher, so SegmentWriter::_rsid_to_rowset is gone while the
vertical writer keeps its map for the flexible fill and BlockAggregator; the
retriever's clear() now also drops the fetcher that holds the read plan.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: csun5285

be/src/storage/key/row_key_encoder.cpp

@@ -38,13 +38,23 @@ RowKeyEncoder::RowKeyEncoder(const TabletSchema& schema, bool mow)
     }
 }
 
-void RowKeyEncoder::_init_mow(const TabletSchema& schema) {
+RowKeyEncoder::RowKeyEncoder(const TabletSchema& schema)
+        : _num_short_key_columns(schema.num_short_key_columns()) {
+    _init_seq_coder(schema);
+    _add_default_sort_key_columns(schema);
+}
+
+void RowKeyEncoder::_init_seq_coder(const TabletSchema& schema) {
     // encode the sequence id into the primary key index
     if (schema.has_sequence_col()) {
         const auto& column = schema.column(schema.sequence_col_idx());
         _seq_coder = get_key_coder(column.type());
         _seq_col_length = column.length();
     }
+}
+
+void RowKeyEncoder::_init_mow(const TabletSchema& schema) {
+    _init_seq_coder(schema);
 
     if (schema.cluster_key_uids().empty()) {
         _add_default_sort_key_columns(schema);

be/src/storage/key/row_key_encoder.h

@@ -35,15 +35,24 @@ class TabletSchema;
 // is encoded as KEY_NULL_FIRST_MARKER without value bytes.
 class RowKeyEncoder {
 public:
+    // The segment writer's encoder: sort-key view +, for a mow table with
+    // cluster keys, the separate primary-key view and the rowid suffix.
     RowKeyEncoder(const TabletSchema& schema, bool mow);
 
+    // The probe side's encoder: the key the primary key index is built on and
+    // probed with, i.e. the schema key columns plus the sequence suffix. Its
+    // sort-key view IS that key, so full_encode() returns it for every mow
+    // table, with or without cluster keys.
+    explicit RowKeyEncoder(const TabletSchema& schema);
+
     // Encode the sort key columns at `pos` with full length.
     std::string full_encode(const std::vector<IOlapColumnDataAccessor*>& key_columns,
                             size_t pos) const;
 
     // For a mow table with cluster keys, encode the primary key columns at
     // `pos` with full length, producing the key stored in and probed against
-    // the primary key index.
+    // the primary key index. Without cluster keys the segment sorts by those
+    // same columns, so the single-argument constructor above covers that case.
     std::string full_encode_primary_keys(const std::vector<IOlapColumnDataAccessor*>& key_columns,
                                          size_t pos) const;
 
@@ -69,6 +78,7 @@ class RowKeyEncoder {
                                     const std::vector<IOlapColumnDataAccessor*>& key_columns,
                                     size_t pos);
 
+    void _init_seq_coder(const TabletSchema& schema);
     void _init_mow(const TabletSchema& schema);
     void _init_non_mow(const TabletSchema& schema);
     void _add_default_sort_key_columns(const TabletSchema& schema);

be/src/storage/mow/historical_row_fetcher.cpp

@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "storage/mow/historical_row_fetcher.h"
+
+#include "storage/rowset/rowset.h"
+#include "storage/tablet/tablet_schema.h"
+
+namespace doris {
+
+HistoricalRowFetcher::HistoricalRowFetcher(segment_v2::HistoricalRowRetrieverContext context)
+        : _context(std::move(context)),
+          _flexible_plan(_context.tablet_schema->has_row_store_for_all_columns()) {}
+
+void HistoricalRowFetcher::pin_rowset(const RowsetSharedPtr& rowset) {
+    _rsid_to_rowset.emplace(rowset->rowset_id(), rowset);
+}
+
+void HistoricalRowFetcher::plan_fixed_read(const RowLocation& loc, size_t dst_pos) {
+    _fixed_plan.prepare_to_read(loc, dst_pos);
+}
+
+void HistoricalRowFetcher::plan_flexible_read(const RowLocation& loc, size_t dst_pos,
+                                              const BitmapValue& skip_bitmap) {
+    _flexible_plan.prepare_to_read(loc, dst_pos, skip_bitmap);
+}
+
+Status HistoricalRowFetcher::fill_missing_columns(const TabletSchema& tablet_schema,
+                                                  Block& full_block,
+                                                  const std::vector<bool>& use_default_or_null_flag,
+                                                  bool has_default_or_nullable,
+                                                  uint32_t segment_start_pos,
+                                                  const Block* block) const {
+    return _fixed_plan.fill_missing_columns(_context, _rsid_to_rowset, tablet_schema, full_block,
+                                            use_default_or_null_flag, has_default_or_nullable,
+                                            segment_start_pos, block);
+}
+
+Status HistoricalRowFetcher::fill_non_primary_key_columns(
+        const TabletSchema& tablet_schema, Block& full_block,
+        const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
+        uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
+        std::vector<BitmapValue>* skip_bitmaps) const {
+    return _flexible_plan.fill_non_primary_key_columns(
+            _context, _rsid_to_rowset, tablet_schema, full_block, use_default_or_null_flag,
+            has_default_or_nullable, segment_start_pos, block_start_pos, block, skip_bitmaps);
+}
+
+Status HistoricalRowFetcher::read_columns(const TabletSchema& tablet_schema,
+                                          std::vector<uint32_t> cids_to_read, Block& dst_block,
+                                          std::map<uint32_t, uint32_t>* read_index,
+                                          bool force_read_old_delete_signs,
+                                          const signed char* __restrict cur_delete_signs) const {
+    return _fixed_plan.read_columns_by_plan(tablet_schema, std::move(cids_to_read), _rsid_to_rowset,
+                                            dst_block, read_index, force_read_old_delete_signs,
+                                            cur_delete_signs);
+}
+
+} // namespace doris

be/src/storage/mow/historical_row_fetcher.h

@@ -0,0 +1,90 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <vector>
+
+#include "common/status.h"
+#include "storage/partial_update_info.h"
+#include "storage/rowset/rowset_fwd.h"
+#include "storage/segment/historical_row_retriever.h"
+
+namespace doris {
+class Block;
+class TabletSchema;
+
+// Reads old-row column values for merge-on-write loads. One instance for each
+// segment-writer flush; it owns the rowset pins and the read plans that the
+// MowKeyProbe outcomes feed.
+class HistoricalRowFetcher {
+public:
+    explicit HistoricalRowFetcher(segment_v2::HistoricalRowRetrieverContext context);
+
+    // Keep `rowset` alive until fill/read is done.
+    void pin_rowset(const RowsetSharedPtr& rowset);
+
+    // ---- plan building; the destination position (segment_pos / delta_pos /
+    //      block_pos) is not interpreted by the fetcher ----
+    void plan_fixed_read(const RowLocation& loc, size_t dst_pos);
+    void plan_flexible_read(const RowLocation& loc, size_t dst_pos, const BitmapValue& skip_bitmap);
+
+    // ---- fixed partial update / binlog AFTER fill ----
+    // Same behavior as FixedReadPlan::fill_missing_columns: forces reading
+    // old delete signs, fills in this order: default -> null -> auto-inc ->
+    // any value, keeps the sequence column non-decreasing over delete-signed
+    // old rows.
+    Status fill_missing_columns(const TabletSchema& tablet_schema, Block& full_block,
+                                const std::vector<bool>& use_default_or_null_flag,
+                                bool has_default_or_nullable, uint32_t segment_start_pos,
+                                const Block* block) const;
+
+    // ---- flexible partial update fill ----
+    // Same behavior as FlexibleReadPlan::fill_non_primary_key_columns:
+    // driven by the skip bitmap, re-checks the delete sign with the seq map
+    // column in mind, on_update_current_timestamp forces default, auto-inc
+    // copied in place from the current block.
+    Status fill_non_primary_key_columns(const TabletSchema& tablet_schema, Block& full_block,
+                                        const std::vector<bool>& use_default_or_null_flag,
+                                        bool has_default_or_nullable, uint32_t segment_start_pos,
+                                        uint32_t block_start_pos, const Block* block,
+                                        std::vector<BitmapValue>* skip_bitmaps) const;
+
+    // ---- raw column read on the fixed plan ----
+    // No default/null fill steps, no auto-inc, no seq handling. Used by the
+    // binlog BEFORE image (cids = visible value columns; rows not in
+    // *read_index are left as NULL) and BlockAggregator::fill_sequence_column.
+    Status read_columns(const TabletSchema& tablet_schema, std::vector<uint32_t> cids_to_read,
+                        Block& dst_block, std::map<uint32_t, uint32_t>* read_index,
+                        bool force_read_old_delete_signs,
+                        const signed char* __restrict cur_delete_signs = nullptr) const;
+
+    const std::map<RowsetId, RowsetSharedPtr>& pinned_rowsets() const { return _rsid_to_rowset; }
+
+private:
+    // NOTE: _context must stay declared before _flexible_plan; the constructor
+    // init list reads _context.tablet_schema (FlexibleReadPlan has no default ctor).
+    segment_v2::HistoricalRowRetrieverContext _context;
+    FixedReadPlan _fixed_plan;
+    FlexibleReadPlan _flexible_plan;
+    std::map<RowsetId, RowsetSharedPtr> _rsid_to_rowset;
+};
+
+} // namespace doris

be/src/storage/mow/key_probe.cpp

@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "storage/mow/key_probe.h"
+
+#include "common/cast_set.h"
+#include "common/config.h"
+#include "common/logging.h"
+#include "service/point_query_executor.h"
+#include "storage/key/row_key_encoder.h"
+#include "storage/partial_update_info.h"
+#include "storage/tablet/base_tablet.h"
+#include "storage/tablet/tablet_meta.h"
+#include "storage/tablet/tablet_schema.h"
+
+namespace doris::segment_v2 {
+
+using namespace ErrorCode;
+
+MowKeyProbe::MowKeyProbe(BaseTablet* tablet, TabletSchema* lookup_schema, bool has_sequence_col,
+                         std::shared_ptr<MowContext> mow_context, const RowsetId& writing_rowset_id,
+                         uint32_t writing_segment_id, Policy policy)
+        : _tablet(tablet),
+          _lookup_schema(lookup_schema),
+          _has_sequence_col(has_sequence_col),
+          _mow_context(std::move(mow_context)),
+          _writing_rowset_id(writing_rowset_id),
+          _writing_segment_id(writing_segment_id),
+          _policy(policy) {}
+
+Result<ProbeOutcome> MowKeyProbe::probe(
+        const std::string& key, size_t segment_pos, bool key_has_seq_suffix, bool have_delete_sign,
+        const std::vector<RowsetSharedPtr>& specified_rowsets,
+        std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+        PartialUpdateStats& stats) const {
+    RowLocation loc;
+    // save rowset shared ptr so this rowset wouldn't delete
+    RowsetSharedPtr rowset;
+    auto st = _tablet->lookup_row_key(key, _lookup_schema, key_has_seq_suffix, specified_rowsets,
+                                      &loc, _mow_context->max_version, segment_caches, &rowset,
+                                      /*with_rowid=*/false);
+    if (st.is<KEY_NOT_FOUND>()) {
+        ++stats.num_rows_new_added;
+        return ProbeOutcome {KeyProbeResult::NOT_FOUND, {}, nullptr, /*use_default_or_null=*/true};
+    }
+    if (!st.ok() && !st.is<KEY_ALREADY_EXISTS>()) {
+        LOG(WARNING) << "failed to lookup row key, error: " << st;
+        return ResultError(std::move(st));
+    }
+
+    // Stored row's seq is larger, so the incoming row loses.
+    bool seq_loses = st.is<KEY_ALREADY_EXISTS>();
+    // A delete-signed row needs no old values -- but only without a seq col,
+    // since the seq must still be read for merge-on-read compaction.
+    bool delete_sign_skip = have_delete_sign && !_has_sequence_col && _policy.skip_delete_sign;
+    // Flexible PU insert-after-delete: the old row was already deleted in this
+    // load, so the insert counts as a brand-new row.
+    bool in_load_deleted =
+            _policy.skip_in_load_deleted &&
+            _mow_context->delete_bitmap->contains(
+                    {loc.rowset_id, loc.segment_id, DeleteBitmap::TEMP_VERSION_COMMON}, loc.row_id);
+    // Skip reading the old row (fill defaults) in any of these cases.
+    bool use_default = (seq_loses && _policy.skip_seq_loses) || delete_sign_skip || in_load_deleted;
+    ProbeOutcome outcome {seq_loses ? KeyProbeResult::FOUND_NEWER : KeyProbeResult::FOUND, loc,
+                          std::move(rowset), use_default};
+
+    // Apply the delete-bitmap marks right away -- see class comment (segcompaction).
+    if (seq_loses) {
+        if (_policy.delete_bitmap_mode == DeleteBitmapMode::OLD_AND_NEW_ROW) {
+            // although we need to mark delete current row, we still need to read missing
+            // columns for this row, we need to ensure that each column is aligned
+            _mow_context->delete_bitmap->add(
+                    {_writing_rowset_id, _writing_segment_id, DeleteBitmap::TEMP_VERSION_COMMON},
+                    cast_set<uint32_t>(segment_pos));
+            ++stats.num_rows_deleted;
+        }
+    } else if (_policy.delete_bitmap_mode != DeleteBitmapMode::READ_ONLY) {
+        _mow_context->delete_bitmap->add(
+                {loc.rowset_id, loc.segment_id, DeleteBitmap::TEMP_VERSION_COMMON}, loc.row_id);
+        ++stats.num_rows_updated;
+    }
+    return outcome;
+}
+
+Result<PrevSeqProbe> MowKeyProbe::probe_previous_seq_value(
+        const std::string& key, const std::vector<RowsetSharedPtr>& specified_rowsets,
+        std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) const {
+    RowLocation loc;
+    RowsetSharedPtr rowset;
+    PrevSeqProbe result;
+    auto st =
+            _tablet->lookup_row_key(key, _lookup_schema, /*with_seq_col=*/false, specified_rowsets,
+                                    &loc, _mow_context->max_version, segment_caches, &rowset,
+                                    /*with_rowid=*/true, &result.encoded_seq_value);
+    if (st.is<KEY_NOT_FOUND>()) {
+        result.outcome = ProbeOutcome {KeyProbeResult::NOT_FOUND,
+                                       {},
+                                       nullptr,
+                                       /*use_default_or_null=*/true};
+        return result;
+    }
+    if (!st.ok()) {
+        return ResultError(std::move(st));
+    }
+    result.outcome.result = KeyProbeResult::FOUND;
+    result.outcome.loc = loc;
+    result.outcome.rowset = std::move(rowset);
+    result.outcome.use_default_or_null = false;
+    return result;
+}
+
+void MowKeyProbe::maybe_invalidate_row_cache(int64_t tablet_id, const TabletSchema& schema,
+                                             DataWriteType write_type, const std::string& key) {
+    // Just invalid row cache for simplicity, since the rowset is not visible at present.
+    // If we update/insert cache, if load failed rowset will not be visible but cached data
+    // will be visible, and lead to inconsistency.
+    if (!config::disable_storage_row_cache && schema.has_row_store_for_all_columns() &&
+        write_type == DataWriteType::TYPE_DIRECT) {
+        // invalidate cache
+        RowCache::instance()->erase({tablet_id, key});
+    }
+}
+
+std::string encode_mow_key_invalidate_cache(
+        const RowKeyEncoder& key_encoder, const std::vector<IOlapColumnDataAccessor*>& key_columns,
+        const IOlapColumnDataAccessor* seq_column, size_t pos, bool row_has_seq, int64_t tablet_id,
+        const TabletSchema& schema, DataWriteType write_type) {
+    std::string key = key_encoder.full_encode(key_columns, pos);
+    // the row cache uses the key without the seq as its key, so invalidate before the suffix
+    MowKeyProbe::maybe_invalidate_row_cache(tablet_id, schema, write_type, key);
+    if (row_has_seq) {
+        key_encoder.append_seq_suffix(&key, seq_column, pos);
+    }
+    return key;
+}
+
+} // namespace doris::segment_v2