[refactor](storage) move flexible partial-update fill into the block transform

be/src/common/config.cpp

@@ -2360,6 +2360,8 @@ Status set_fuzzy_configs() {
             ((distribution(*generator) % 2) == 0) ? "true" : "false";
     fuzzy_field_and_value["enable_packed_file"] =
             ((distribution(*generator) % 2) == 0) ? "true" : "false";
+    fuzzy_field_and_value["enable_vertical_segment_writer"] =
+            ((distribution(*generator) % 2) == 0) ? "true" : "false";
     fuzzy_field_and_value["max_segment_partial_column_cache_size"] =
             ((distribution(*generator) % 2) == 0) ? "5" : "10";
 

be/src/storage/partial_update_info.cpp

@@ -30,18 +30,18 @@
 #include "core/value/bitmap_value.h"
 #include "storage/iterator/olap_data_convertor.h"
 #include "storage/key/row_key_encoder.h"
+#include "storage/mow/historical_row_fetcher.h"
+#include "storage/mow/key_probe.h"
 #include "storage/olap_common.h"
 #include "storage/rowset/rowset.h"
 #include "storage/rowset/rowset_writer_context.h"
 #include "storage/segment/historical_row_retriever.h"
-#include "storage/segment/vertical_segment_writer.h"
 #include "storage/tablet/base_tablet.h"
 #include "storage/tablet/tablet_meta.h"
 #include "storage/tablet/tablet_schema.h"
 #include "storage/utils.h"
 
 namespace doris {
-namespace {
 
 ColumnBitmap* get_mutable_skip_bitmap_column(Block* block, size_t skip_bitmap_col_idx) {
     auto skip_bitmap_column =
@@ -51,8 +51,6 @@ ColumnBitmap* get_mutable_skip_bitmap_column(Block* block, size_t skip_bitmap_co
     return skip_bitmap_column_ptr;
 }
 
-} // namespace
-
 Status PartialUpdateInfo::init(int64_t tablet_id, int64_t txn_id, const TabletSchema& tablet_schema,
                                UniqueKeyUpdateModePB unique_key_update_mode,
                                PartialUpdateNewRowPolicyPB policy,
@@ -840,8 +838,31 @@ Status FlexibleReadPlan::fill_non_primary_key_columns_for_row_store(
     return Status::OK();
 }
 
-BlockAggregator::BlockAggregator(segment_v2::VerticalSegmentWriter& vertical_segment_writer)
-        : _writer(vertical_segment_writer), _tablet_schema(*_writer._tablet_schema) {}
+BlockAggregator::BlockAggregator(TabletSchema& tablet_schema, BaseTabletSPtr tablet,
+                                 std::shared_ptr<MowContext> mow_context,
+                                 const PartialUpdateInfo& partial_update_info,
+                                 const RowKeyEncoder& key_encoder,
+                                 const segment_v2::MowKeyProbe& probe,
+                                 HistoricalRowFetcher& fetcher)
+        : _tablet_schema(tablet_schema),
+          _tablet(std::move(tablet)),
+          _mow_context(std::move(mow_context)),
+          _partial_update_info(partial_update_info),
+          _key_encoder(key_encoder),
+          _convertor(std::make_unique<OlapBlockDataConvertor>()),
+          _probe(probe),
+          _fetcher(fetcher) {
+    _convertor->resize(tablet_schema.num_columns());
+    for (uint32_t cid = 0; cid < tablet_schema.num_key_columns(); ++cid) {
+        _convertor->add_column_data_convertor_at(tablet_schema.column(cid), cid);
+    }
+    if (tablet_schema.has_sequence_col()) {
+        auto cid = cast_set<uint32_t>(tablet_schema.sequence_col_idx());
+        _convertor->add_column_data_convertor_at(tablet_schema.column(cid), cid);
+    }
+}
+
+BlockAggregator::~BlockAggregator() = default;
 
 void BlockAggregator::merge_one_row(MutableBlock& dst_block, Block* src_block, int rid,
                                     BitmapValue& skip_bitmap) {
@@ -919,30 +940,26 @@ Status BlockAggregator::aggregate_rows(
 
     _state.reset();
 
-    RowLocation loc;
-    RowsetSharedPtr rowset;
-    std::string previous_encoded_seq_value {};
-    Status st = _writer._tablet->lookup_row_key(
-            key, &_tablet_schema, false, specified_rowsets, &loc, _writer._mow_context->max_version,
-            segment_caches, &rowset, true, &previous_encoded_seq_value);
+    auto prev_seq_res = _probe.probe_previous_seq_value(key, specified_rowsets, segment_caches);
     int pos = start;
-    bool is_expected_st = (st.is<ErrorCode::KEY_NOT_FOUND>() || st.ok());
-    DCHECK(is_expected_st || st.is<ErrorCode::MEM_LIMIT_EXCEEDED>())
+    DCHECK(prev_seq_res.has_value() || prev_seq_res.error().is<ErrorCode::MEM_LIMIT_EXCEEDED>())
             << "[BlockAggregator::aggregate_rows] unexpected error status while lookup_row_key:"
-            << st;
-    if (!is_expected_st) {
-        return st;
+            << prev_seq_res.error();
+    if (!prev_seq_res.has_value()) {
+        return prev_seq_res.error();
     }
+    std::string previous_encoded_seq_value = std::move(prev_seq_res.value().encoded_seq_value);
+    segment_v2::ProbeOutcome probe_out = std::move(prev_seq_res.value().outcome);
 
     std::string cur_seq_val;
-    if (st.ok()) {
+    if (probe_out.result == segment_v2::KeyProbeResult::FOUND) {
         for (pos = start; pos < end; pos++) {
             auto& skip_bitmap = skip_bitmaps->at(pos);
             bool row_has_sequence_col = (!skip_bitmap.contains(seq_col_unique_id));
             // Discard all the rows whose seq value is smaller than previous_encoded_seq_value.
             if (row_has_sequence_col) {
                 std::string seq_val {};
-                _writer._key_encoder.append_seq_suffix(&seq_val, seq_column, pos);
+                _key_encoder.append_seq_suffix(&seq_val, seq_column, pos);
                 if (Slice {seq_val}.compare(Slice {previous_encoded_seq_value}) < 0) {
                     continue;
                 }
@@ -959,12 +976,11 @@ Status BlockAggregator::aggregate_rows(
         if (row_has_sequence_col) {
             std::string seq_val {};
             // for rows that don't specify seqeunce col, seq_val will be encoded to minial value
-            _writer._key_encoder.append_seq_suffix(&seq_val, seq_column, pos);
+            _key_encoder.append_seq_suffix(&seq_val, seq_column, pos);
             cur_seq_val = std::move(seq_val);
         } else {
             cur_seq_val.clear();
-            RETURN_IF_ERROR(_writer._generate_encoded_default_seq_value(
-                    _tablet_schema, *_writer._opts.rowset_ctx->partial_update_info, &cur_seq_val));
+            RETURN_IF_ERROR(_generate_encoded_default_seq_value(&cur_seq_val));
         }
     }
 
@@ -977,7 +993,7 @@ Status BlockAggregator::aggregate_rows(
             append_or_merge_row(output_block, block, rid, skip_bitmap, have_delete_sign);
         } else {
             std::string seq_val {};
-            _writer._key_encoder.append_seq_suffix(&seq_val, seq_column, rid);
+            _key_encoder.append_seq_suffix(&seq_val, seq_column, rid);
             if (Slice {seq_val}.compare(Slice {cur_seq_val}) >= 0) {
                 append_or_merge_row(output_block, block, rid, skip_bitmap, have_delete_sign);
                 cur_seq_val = std::move(seq_val);
@@ -990,6 +1006,33 @@ Status BlockAggregator::aggregate_rows(
     return Status::OK();
 };
 
+Status BlockAggregator::_generate_encoded_default_seq_value(std::string* encoded_value) {
+    const auto& seq_column = _tablet_schema.column(_tablet_schema.sequence_col_idx());
+    auto block = _tablet_schema.create_block_by_cids(
+            {cast_set<uint32_t>(_tablet_schema.sequence_col_idx())});
+    if (seq_column.has_default_value()) {
+        auto idx = _tablet_schema.sequence_col_idx() - _tablet_schema.num_key_columns();
+        const auto& default_value = _partial_update_info.default_values[idx];
+        StringRef str {default_value};
+        RETURN_IF_ERROR(block.get_by_position(0).type->get_serde()->default_from_string(
+                str, *block.get_by_position(0).column->assert_mutable().get()));
+
+    } else {
+        block.get_by_position(0).column->assert_mutable()->insert_default();
+    }
+    DCHECK_EQ(block.rows(), 1);
+    auto olap_data_convertor = std::make_unique<OlapBlockDataConvertor>();
+    olap_data_convertor->add_column_data_convertor(seq_column);
+    olap_data_convertor->set_source_content(&block, 0, 1);
+    auto [status, column] = olap_data_convertor->convert_column_data(0);
+    if (!status.ok()) {
+        return status;
+    }
+    // include marker
+    _key_encoder.append_seq_suffix(encoded_value, column, 0);
+    return Status::OK();
+}
+
 Status BlockAggregator::aggregate_for_sequence_column(
         Block* block, int num_rows, const std::vector<IOlapColumnDataAccessor*>& key_columns,
         IOlapColumnDataAccessor* seq_column, const std::vector<RowsetSharedPtr>& specified_rowsets,
@@ -1008,7 +1051,7 @@ Status BlockAggregator::aggregate_for_sequence_column(
     int same_key_rows {0};
     std::string previous_key {};
     for (int block_pos {0}; block_pos < num_rows; block_pos++) {
-        std::string key = _writer._key_encoder.full_encode(key_columns, block_pos);
+        std::string key = _key_encoder.full_encode(key_columns, block_pos);
         if (block_pos > 0 && previous_key == key) {
             same_key_rows++;
         } else {
@@ -1042,7 +1085,7 @@ Status BlockAggregator::fill_sequence_column(Block* block, size_t num_rows,
     auto seq_col_block = _tablet_schema.create_block_by_cids(cids);
     auto tmp_block = _tablet_schema.create_block_by_cids(cids);
     std::map<uint32_t, uint32_t> read_index;
-    RETURN_IF_ERROR(read_plan.read_columns_by_plan(_tablet_schema, cids, _writer._rsid_to_rowset,
+    RETURN_IF_ERROR(read_plan.read_columns_by_plan(_tablet_schema, cids, _fetcher.pinned_rowsets(),
                                                    seq_col_block, &read_index, false));
 
     auto new_seq_col_ptr = tmp_block.get_by_position(0).column->assert_mutable();
@@ -1085,10 +1128,20 @@ Status BlockAggregator::aggregate_for_insert_after_delete(
                     ? _tablet_schema.column(_tablet_schema.sequence_col_idx()).unique_id()
                     : -1;
     FixedReadPlan read_plan;
+    // This insert-after-delete pass doesn't report partial-update counts; discard them.
+    PartialUpdateStats discarded;
+    // the losing delete-sign row is removed from the block below instead of
+    // marking itself, so the probe only marks the old row
+    segment_v2::MowKeyProbe probe(
+            _tablet.get(), &_tablet_schema, _tablet_schema.has_sequence_col(), _mow_context,
+            RowsetId {}, 0,
+            segment_v2::MowKeyProbe::Policy {
+                    .mark_deleted = segment_v2::MowKeyProbe::MarkDeleted::OLD_ROW,
+            });
     for (size_t block_pos {0}; block_pos < num_rows; block_pos++) {
         size_t delta_pos = block_pos;
         auto& skip_bitmap = skip_bitmaps->at(block_pos);
-        std::string key = _writer._key_encoder.full_encode(key_columns, delta_pos);
+        std::string key = _key_encoder.full_encode(key_columns, delta_pos);
         bool have_delete_sign =
                 (!skip_bitmap.contains(delete_sign_col_unique_id) && delete_signs[block_pos] != 0);
         if (delta_pos > 0 && previous_key == key) {
@@ -1099,37 +1152,31 @@ Status BlockAggregator::aggregate_for_insert_after_delete(
             DCHECK(previous_has_delete_sign);
             DCHECK(!have_delete_sign);
             ++duplicate_rows;
-            RowLocation loc;
-            RowsetSharedPtr rowset;
-            Status st = _writer._tablet->lookup_row_key(
-                    key, &_tablet_schema, false, specified_rowsets, &loc,
-                    _writer._mow_context->max_version, segment_caches, &rowset, true);
-            bool is_expected_st = (st.is<ErrorCode::KEY_NOT_FOUND>() || st.ok());
-            DCHECK(is_expected_st || st.is<ErrorCode::MEM_LIMIT_EXCEEDED>())
+            auto probe_res = probe.probe(key, /*segment_pos=*/0, /*key_has_seq_suffix=*/false,
+                                         /*have_delete_sign=*/false, specified_rowsets,
+                                         segment_caches, discarded);
+            DCHECK(probe_res.has_value() || probe_res.error().is<ErrorCode::MEM_LIMIT_EXCEEDED>())
                     << "[BlockAggregator::aggregate_for_insert_after_delete] unexpected error "
                        "status while lookup_row_key:"
-                    << st;
-            if (!is_expected_st) {
-                return st;
+                    << probe_res.error();
+            if (!probe_res.has_value()) {
+                return probe_res.error();
             }
+            segment_v2::ProbeOutcome out = std::move(probe_res.value());
 
-            Slice previous_seq_slice {};
-            if (st.ok()) {
+            if (out.result == segment_v2::KeyProbeResult::FOUND) {
                 if (_tablet_schema.has_sequence_col()) {
                     // if the insert row doesn't specify the sequence column, we need to
                     // read the historical's sequence column value so that we don't need
                     // to handle seqeunce column in append_block_with_flexible_content()
                     // for this row
                     bool row_has_sequence_col = (!skip_bitmap.contains(seq_col_unique_id));
                     if (!row_has_sequence_col) {
-                        read_plan.prepare_to_read(loc, block_pos);
-                        _writer._rsid_to_rowset.emplace(rowset->rowset_id(), rowset);
+                        read_plan.prepare_to_read(out.loc, block_pos);
+                        _fetcher.pin_rowset(out.rowset);
                     }
                 }
-                // delete the existing row
-                _writer._mow_context->delete_bitmap->add(
-                        {loc.rowset_id, loc.segment_id, DeleteBitmap::TEMP_VERSION_COMMON},
-                        loc.row_id);
+                // the old-row delete mark is set inside probe()
             }
             // and remove the row with delete sign from the current block
             filter_map[block_pos - 1] = 0;
@@ -1170,9 +1217,9 @@ Status BlockAggregator::convert_pk_columns(Block* block, size_t row_pos, size_t
                                            std::vector<IOlapColumnDataAccessor*>& key_columns) {
     key_columns.clear();
     for (uint32_t cid {0}; cid < _tablet_schema.num_key_columns(); cid++) {
-        RETURN_IF_ERROR(_writer._olap_data_convertor->set_source_content_with_specifid_column(
+        RETURN_IF_ERROR(_convertor->set_source_content_with_specifid_column(
                 block->get_by_position(cid), row_pos, num_rows, cid));
-        auto [status, column] = _writer._olap_data_convertor->convert_column_data(cid);
+        auto [status, column] = _convertor->convert_column_data(cid);
         if (!status.ok()) {
             return status;
         }
@@ -1186,9 +1233,9 @@ Status BlockAggregator::convert_seq_column(Block* block, size_t row_pos, size_t
     seq_column = nullptr;
     if (_tablet_schema.has_sequence_col()) {
         auto seq_col_idx = _tablet_schema.sequence_col_idx();
-        RETURN_IF_ERROR(_writer._olap_data_convertor->set_source_content_with_specifid_column(
+        RETURN_IF_ERROR(_convertor->set_source_content_with_specifid_column(
                 block->get_by_position(seq_col_idx), row_pos, num_rows, seq_col_idx));
-        auto [status, column] = _writer._olap_data_convertor->convert_column_data(seq_col_idx);
+        auto [status, column] = _convertor->convert_column_data(seq_col_idx);
         if (!status.ok()) {
             return status;
         }
@@ -1220,7 +1267,7 @@ Status BlockAggregator::aggregate_for_flexible_partial_update(
     if (block->rows() != num_rows) {
         num_rows = block->rows();
         // data in block has changed, should re-encode key columns, sequence column
-        _writer._olap_data_convertor->clear_source_content();
+        _convertor->clear_source_content();
         RETURN_IF_ERROR(convert_pk_columns(block, 0, num_rows, key_columns));
         RETURN_IF_ERROR(convert_seq_column(block, 0, num_rows, seq_column));
     }

be/src/storage/partial_update_info.h

@@ -21,12 +21,14 @@
 #include <cstdint>
 #include <functional>
 #include <map>
+#include <memory>
 #include <set>
 #include <string>
 #include <vector>
 
 #include "common/status.h"
 #include "core/column/column.h"
+#include "core/data_type/primitive_type.h"
 #include "storage/rowset/rowset_fwd.h"
 #include "storage/tablet/tablet_fwd.h"
 
@@ -45,8 +47,12 @@ struct HistoricalRowRetrieverContext;
 struct RowsetWriterContext;
 struct RowsetId;
 class BitmapValue;
+class HistoricalRowFetcher;
+class OlapBlockDataConvertor;
+class RowKeyEncoder;
+struct MowContext;
 namespace segment_v2 {
-class VerticalSegmentWriter;
+class MowKeyProbe;
 }
 
 class SegmentCacheHandle;
@@ -194,10 +200,18 @@ class FlexibleReadPlan {
     std::map<RowsetId, std::map<uint32_t /* segment_id */, std::vector<RidAndPos>>> row_store_plan;
 };
 
+ColumnBitmap* get_mutable_skip_bitmap_column(Block* block, size_t skip_bitmap_col_idx);
+
 class BlockAggregator {
 public:
-    ~BlockAggregator() = default;
-    BlockAggregator(segment_v2::VerticalSegmentWriter& vertical_segment_writer);
+    ~BlockAggregator();
+    // All references must live longer than the aggregator; the flexible fill
+    // stage builds everything as locals in one apply() scope. The aggregator
+    // owns its block convertor (key + sequence column slots).
+    BlockAggregator(TabletSchema& tablet_schema, BaseTabletSPtr tablet,
+                    std::shared_ptr<MowContext> mow_context,
+                    const PartialUpdateInfo& partial_update_info, const RowKeyEncoder& key_encoder,
+                    const segment_v2::MowKeyProbe& probe, HistoricalRowFetcher& fetcher);
 
     Status convert_pk_columns(Block* block, size_t row_pos, size_t num_rows,
                               std::vector<IOlapColumnDataAccessor*>& key_columns);
@@ -237,8 +251,16 @@ class BlockAggregator {
                           const std::vector<RowsetSharedPtr>& specified_rowsets,
                           std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
 
-    segment_v2::VerticalSegmentWriter& _writer;
+    Status _generate_encoded_default_seq_value(std::string* encoded_value);
+
     TabletSchema& _tablet_schema;
+    BaseTabletSPtr _tablet;
+    std::shared_ptr<MowContext> _mow_context;
+    const PartialUpdateInfo& _partial_update_info;
+    const RowKeyEncoder& _key_encoder;
+    std::unique_ptr<OlapBlockDataConvertor> _convertor;
+    const segment_v2::MowKeyProbe& _probe;
+    HistoricalRowFetcher& _fetcher;
 
     // used to store state when aggregating rows in block
     struct AggregateState {

be/src/storage/rowset/beta_rowset_writer.cpp

@@ -1209,7 +1209,6 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction(
     writer_options.write_type = _context.write_type;
     writer_options.write_type = DataWriteType::TYPE_COMPACTION;
     writer_options.max_rows_per_segment = _context.max_rows_per_segment;
-    writer_options.mow_ctx = _context.mow_context;
 
     *writer = std::make_unique<segment_v2::SegmentWriter>(
             file_writer.get(), _num_segcompacted, _context.tablet_schema, _context.tablet,