[refactor](storage) introduce the block transform chain

be/src/storage/rowset/segment_creator.cpp

@@ -49,6 +49,7 @@
 #include "storage/segment/segment_writer.h"
 #include "storage/segment/vertical_segment_writer.h"
 #include "storage/tablet/tablet_schema.h"
+#include "storage/transform/block_transform.h"
 #include "storage/utils.h"
 #include "util/debug_points.h"
 #include "util/json/json_parser.h"
@@ -58,6 +59,23 @@
 namespace doris {
 using namespace ErrorCode;
 
+namespace {
+
+segment_v2::TransformExecContext make_transform_exec_context(RowsetWriterContext& context,
+                                                             int32_t segment_id) {
+    return {.tablet_schema = context.tablet_schema,
+            .write_type = context.write_type,
+            .tablet = context.tablet,
+            .mow_context = context.mow_context,
+            .partial_update_info = context.partial_update_info,
+            .rowset_ctx = &context,
+            .rowset_id = context.rowset_id,
+            .segment_id = segment_id,
+            .derived_column = {}};
+}
+
+} // namespace
+
 SegmentFlusher::SegmentFlusher(RowsetWriterContext& context, SegmentFileCollection& seg_files,
                                InvertedIndexFileCollection& idx_files)
         : _context(context), _seg_files(seg_files), _idx_files(idx_files) {}
@@ -72,14 +90,21 @@ Status SegmentFlusher::flush_single_block(const Block* block, int32_t segment_id
     }
     Block flush_block(*block);
     bool no_compression = flush_block.bytes() <= config::segment_compression_threshold_kb * 1024;
+    segment_v2::DerivedColumn derived_column;
+    RETURN_IF_ERROR(transform_block(&flush_block, segment_id, &derived_column));
     bool use_vertical_segment_writer =
             config::enable_vertical_segment_writer && !_context.write_binlog_opt().enable;
     if (use_vertical_segment_writer) {
         std::unique_ptr<segment_v2::VerticalSegmentWriter> writer;
         RETURN_IF_ERROR(_create_segment_writer(writer, segment_id, no_compression));
+        // the vertical writer feeds the derived column in small fixed-size batches
+        writer->set_derived_column(std::move(derived_column));
         RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_add_rows(writer, &flush_block, 0, flush_block.rows()));
         RETURN_IF_ERROR(_flush_segment_writer(writer, flush_size));
     } else {
+        // the horizontal writer has no streaming feed, build it all up front
+        RETURN_IF_ERROR_OR_CATCH_EXCEPTION(
+                segment_v2::materialize_derived_columns(derived_column, &flush_block));
         std::unique_ptr<segment_v2::SegmentWriter> writer;
         RETURN_IF_ERROR(_create_segment_writer(writer, segment_id, no_compression));
         RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_add_rows(writer, &flush_block, 0, flush_block.rows()));
@@ -88,6 +113,15 @@ Status SegmentFlusher::flush_single_block(const Block* block, int32_t segment_id
     return Status::OK();
 }
 
+Status SegmentFlusher::transform_block(Block* block, int32_t segment_id,
+                                       segment_v2::DerivedColumn* derived_column) {
+    auto transform_ctx = make_transform_exec_context(_context, segment_id);
+    RETURN_IF_ERROR_OR_CATCH_EXCEPTION(
+            segment_v2::build_transform_chain(_context).apply(transform_ctx, block));
+    *derived_column = std::move(transform_ctx.derived_column);
+    return Status::OK();
+}
+
 Status SegmentFlusher::close() {
     RETURN_IF_ERROR(_seg_files.close());
     RETURN_IF_ERROR(_preload_segment_indexes_to_file_cache());
@@ -406,6 +440,15 @@ Status SegmentCreator::add_block(const Block* block) {
     size_t block_row_num = block->rows();
     size_t row_avg_size_in_bytes = std::max((size_t)1, block_size_in_bytes / block_row_num);
     size_t row_offset = 0;
+    // This seam always feeds the horizontal writer, so the derived column is
+    // materialized up front, like flush_single_block's horizontal branch.
+    Block* shared_block = const_cast<Block*>(block);
+    auto transform_block = [&]() -> Status {
+        segment_v2::DerivedColumn derived_column;
+        RETURN_IF_ERROR(
+                _segment_flusher.transform_block(shared_block, /*segment_id=*/-1, &derived_column));
+        return segment_v2::materialize_derived_columns(derived_column, shared_block);
+    };
 
     if (_flush_writer == nullptr) {
         RETURN_IF_ERROR(_segment_flusher.create_writer(_flush_writer, allocate_segment_id()));
@@ -421,6 +464,7 @@ Status SegmentCreator::add_block(const Block* block) {
             DCHECK(max_row_add > 0);
         }
         size_t input_row_num = std::min(block_row_num - row_offset, size_t(max_row_add));
+        RETURN_IF_ERROR(transform_block());
         RETURN_IF_ERROR(_flush_writer->add_rows(block, row_offset, input_row_num));
         row_offset += input_row_num;
     } while (row_offset < block_row_num);

be/src/storage/rowset/segment_creator.h

@@ -20,7 +20,9 @@
 #include <gen_cpp/internal_service.pb.h>
 #include <gen_cpp/olap_file.pb.h>
 
+#include <memory>
 #include <mutex>
+#include <utility>
 #include <vector>
 
 #include "common/status.h"
@@ -37,6 +39,10 @@ class Block;
 namespace segment_v2 {
 class SegmentWriter;
 class VerticalSegmentWriter;
+class DerivedColumnGenerator;
+// Matches block_transform.h: at most one derived column (the row-store column)
+// for each flush, held as a {cid, generator} pair; null generator means none.
+using DerivedColumn = std::pair<uint32_t, std::shared_ptr<const DerivedColumnGenerator>>;
 } // namespace segment_v2
 
 struct SegmentStatistics;
@@ -101,6 +107,11 @@ class SegmentFlusher {
 
     ~SegmentFlusher();
 
+    // Runs the block transform chain on `block` and hands back the derived (row-store)
+    // column for the caller to feed into its writer.
+    Status transform_block(Block* block, int32_t segment_id,
+                           segment_v2::DerivedColumn* derived_column);
+
     // Return the file size flushed to disk in "flush_size"
     // This method is thread-safe.
     Status flush_single_block(const Block* block, int32_t segment_id,

be/src/storage/segment/segment_writer.cpp

@@ -634,19 +634,8 @@ Status SegmentWriter::append_block(const Block* block, size_t row_pos, size_t nu
             << ", block->columns()=" << block->columns()
             << ", _column_writers.size()=" << _column_writers.size()
             << ", _tablet_schema->dump_structure()=" << _tablet_schema->dump_structure();
-    // Row column should be filled here when it's a directly write from memtable
-    // or it's schema change write(since column data type maybe changed, so we should reubild)
-    if (_opts.write_type == DataWriteType::TYPE_DIRECT ||
-        _opts.write_type == DataWriteType::TYPE_SCHEMA_CHANGE) {
-        _serialize_block_to_row_column(*const_cast<Block*>(block));
-    }
-
-    if (_opts.rowset_ctx->write_type != DataWriteType::TYPE_COMPACTION &&
-        _tablet_schema->num_variant_columns() > 0) {
-        RETURN_IF_ERROR(variant_util::parse_and_materialize_variant_columns(
-                const_cast<Block&>(*block), *_tablet_schema, _column_ids));
-    }
-
+    // Blocks from the seams arrive already transformed (variants parsed, row-store
+    // column materialized); compaction-family callers bring rows that are already final.
     _olap_data_convertor->set_source_content(block, row_pos, num_rows);
 
     // convert column data from engine format to storage layer format

be/src/storage/segment/vertical_segment_writer.cpp

@@ -78,6 +78,7 @@
 #include "storage/segment/variant/variant_ext_meta_writer.h"
 #include "storage/tablet/base_tablet.h"
 #include "storage/tablet/tablet_schema.h"
+#include "storage/transform/block_transform.h"
 #include "storage/utils.h"
 #include "util/coding.h"
 #include "util/debug_points.h"
@@ -360,6 +361,39 @@ Status VerticalSegmentWriter::_append_row_store_column(const Block& block, size_
     return Status::OK();
 }
 
+Status VerticalSegmentWriter::_append_generated_column(const DerivedColumnGenerator& generator,
+                                                       const Block& block, size_t row_pos,
+                                                       size_t num_rows, uint32_t cid) {
+    if (num_rows == 0) {
+        return Status::OK();
+    }
+    DCHECK_LE(row_pos + num_rows, block.rows());
+
+    size_t end_pos = row_pos + num_rows;
+    size_t batch_rows = _opts.num_rows_per_block;
+    static constexpr size_t kDerivedColumnBatchBytes = 4 * 1024 * 1024;
+    DCHECK_GT(batch_rows, 0);
+    for (size_t pos = row_pos; pos < end_pos;) {
+        size_t max_rows = std::min(batch_rows, end_pos - pos);
+        auto generated_column = block.get_by_position(cid).column->clone_empty();
+        size_t rows = generator.generate(block, pos, max_rows, kDerivedColumnBatchBytes,
+                                         generated_column.get());
+        DCHECK_GT(rows, 0);
+
+        auto typed_column = block.get_by_position(cid);
+        typed_column.column = std::move(generated_column);
+        RETURN_IF_ERROR(_olap_data_convertor->set_source_content_with_specifid_column(
+                typed_column, 0, rows, cid));
+        auto [status, column] = _olap_data_convertor->convert_column_data(cid);
+        RETURN_IF_ERROR(status);
+        RETURN_IF_ERROR(
+                _column_writers[cid]->append(column->get_nullmap(), column->get_data(), rows));
+        _olap_data_convertor->clear_source_content(cid);
+        pos += rows;
+    }
+    return Status::OK();
+}
+
 Status VerticalSegmentWriter::_probe_key_for_mow(
         const MowKeyProbe& probe, std::string key, std::size_t segment_pos,
         bool have_input_seq_column, bool have_delete_sign,
@@ -966,44 +1000,25 @@ Status VerticalSegmentWriter::write_batch() {
         }
         return Status::OK();
     }
-    // Row column should be filled here when it's a directly write from memtable
-    // or it's schema change write(since column data type maybe changed, so we should reubild)
-    bool should_write_row_store_column = _opts.write_type == DataWriteType::TYPE_DIRECT ||
-                                         _opts.write_type == DataWriteType::TYPE_SCHEMA_CHANGE;
-    if (should_write_row_store_column) {
-        for (uint32_t cid = 0; cid < _tablet_schema->num_columns(); ++cid) {
-            if (!_tablet_schema->column(cid).is_row_store_column()) {
-                continue;
-            }
-            RETURN_IF_ERROR(
-                    _create_column_writer(cid, _tablet_schema->column(cid), _tablet_schema));
-            for (auto& data : _batched_blocks) {
-                RETURN_IF_ERROR(
-                        _append_row_store_column(*data.block, data.row_pos, data.num_rows, cid));
-            }
-            RETURN_IF_ERROR(_check_column_writer_disk_capacity(cid));
-            RETURN_IF_ERROR(_finalize_column_writer_and_update_meta(cid));
-        }
-    }
-
-    std::vector<uint32_t> column_ids;
-    for (uint32_t i = 0; i < _tablet_schema->num_columns(); ++i) {
-        column_ids.emplace_back(i);
-    }
-    if (_opts.rowset_ctx->write_type != DataWriteType::TYPE_COMPACTION &&
-        _tablet_schema->num_variant_columns() > 0) {
+    // The transform chain already validated, parsed variants and decided the derived
+    // (row-store) column; this writer only pumps the generator in bounded batches.
+    if (_derived_column.second) {
+        const auto& [cid, generator] = _derived_column;
+        RETURN_IF_ERROR(_create_column_writer(cid, _tablet_schema->column(cid), _tablet_schema));
         for (auto& data : _batched_blocks) {
-            RETURN_IF_ERROR(variant_util::parse_and_materialize_variant_columns(
-                    const_cast<Block&>(*data.block), *_tablet_schema, column_ids));
+            RETURN_IF_ERROR(_append_generated_column(*generator, *data.block, data.row_pos,
+                                                     data.num_rows, cid));
         }
+        RETURN_IF_ERROR(_check_column_writer_disk_capacity(cid));
+        RETURN_IF_ERROR(_finalize_column_writer_and_update_meta(cid));
     }
 
     std::vector<IOlapColumnDataAccessor*> key_columns;
     IOlapColumnDataAccessor* seq_column = nullptr;
     // the key is cluster key column unique id
     std::map<uint32_t, IOlapColumnDataAccessor*> cid_to_column;
     for (uint32_t cid = 0; cid < _tablet_schema->num_columns(); ++cid) {
-        if (should_write_row_store_column && _tablet_schema->column(cid).is_row_store_column()) {
+        if (_derived_column.second && _derived_column.first == cid) {
             continue;
         }
         RETURN_IF_ERROR(_create_column_writer(cid, _tablet_schema->column(cid), _tablet_schema));
@@ -1045,6 +1060,8 @@ Status VerticalSegmentWriter::write_batch() {
     }
 
     _batched_blocks.clear();
+    // The generator snapshots the batched blocks' rows; it must not survive them.
+    _derived_column = {};
     return Status::OK();
 }
 

be/src/storage/segment/vertical_segment_writer.h

@@ -71,6 +71,11 @@ struct VerticalSegmentWriterOptions {
     std::shared_ptr<MowContext> mow_ctx;
 };
 
+class DerivedColumnGenerator;
+// Matches block_transform.h: at most one derived column (the row-store column)
+// for each flush, held as a {cid, generator} pair; null generator means none.
+using DerivedColumn = std::pair<uint32_t, std::shared_ptr<const DerivedColumnGenerator>>;
+
 struct RowsInBlock {
     const Block* block;
     size_t row_pos;
@@ -96,6 +101,10 @@ class VerticalSegmentWriter {
     Status batch_block(const Block* block, size_t row_pos, size_t num_rows);
     Status write_batch();
 
+    void set_derived_column(DerivedColumn derived_column) {
+        _derived_column = std::move(derived_column);
+    }
+
     [[nodiscard]] std::string data_dir_path() const {
         return _data_dir == nullptr ? "" : _data_dir->path();
     }
@@ -153,6 +162,8 @@ class VerticalSegmentWriter {
     void _set_max_key(const Slice& key);
     Status _append_row_store_column(const Block& block, size_t row_pos, size_t num_rows,
                                     uint32_t cid);
+    Status _append_generated_column(const DerivedColumnGenerator& generator, const Block& block,
+                                    size_t row_pos, size_t num_rows, uint32_t cid);
     // Thin wrapper over MowKeyProbe that translates a ProbeOutcome back into the out-parameters the
     // partial update fill loops use. `found_cb` receives the rowset that holds `loc`: the fixed
     // path pins it in its HistoricalRowFetcher, the flexible path in `_rsid_to_rowset`, which its
@@ -255,6 +266,9 @@ class VerticalSegmentWriter {
 
     std::vector<RowsInBlock> _batched_blocks;
 
+    // the derived column the transform chain hands off to this writer's bounded pump
+    DerivedColumn _derived_column;
+
     BlockAggregator _block_aggregator;
 };