[refactor](changelog) Use explicit column mappings for reads and writes

be/src/agent/task_worker_pool.cpp

@@ -2186,8 +2186,13 @@ void PublishVersionWorkerPool::publish_version_callback(const TAgentTaskRequest&
     }
 
     for (auto& item : discontinuous_version_tablets) {
-        _engine.add_async_publish_task(item.partition_id, item.tablet_id, item.publish_version,
-                                       publish_version_req.transaction_id, false, item.commit_tso);
+        auto enqueue_status = _engine.add_async_publish_task(
+                item.partition_id, item.tablet_id, item.publish_version,
+                publish_version_req.transaction_id, false, item.commit_tso, item.column_mappings);
+        if (!enqueue_status.ok()) {
+            status = std::move(enqueue_status);
+            error_tablet_ids.insert(item.tablet_id);
+        }
     }
     TFinishTaskRequest finish_task_request;
     if (!status.ok()) [[unlikely]] {

be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp

@@ -28,6 +28,7 @@
 #include "cloud/cloud_tablet.h"
 #include "common/status.h"
 #include "runtime/memory/mem_tracker_limiter.h"
+#include "storage/changelog.h"
 #include "storage/olap_common.h"
 #include "storage/rowset/beta_rowset.h"
 #include "storage/rowset/rowset.h"
@@ -351,6 +352,18 @@ Status CloudTabletCalcDeleteBitmapTask::_handle_rowset(
         return status;
     }
 
+    std::shared_ptr<const changelog::ResolvedChangelogLayout> changelog_layout;
+    if (attach_changelog.rowset != nullptr) {
+        if (attach_changelog.column_mappings == nullptr) {
+            return Status::InvalidArgument("cloud txn {} tablet {} misses changelog mappings",
+                                           transaction_id, _tablet_id);
+        }
+        RETURN_IF_ERROR(changelog::resolve_layout(
+                rowset->tablet_schema()->copy_without_variant_extracted_columns(),
+                attach_changelog.rowset->tablet_schema()->copy_without_variant_extracted_columns(),
+                *attach_changelog.column_mappings, &changelog_layout));
+    }
+
     rowset->set_version(Version(version, version));
     if (rowset->tablet_schema()->is_tso_enabled() && rowset->commit_tso().start_tso() == -1) {
         DORIS_CHECK_EQ(rowset->commit_tso().end_tso(), -1);
@@ -406,14 +419,15 @@ Status CloudTabletCalcDeleteBitmapTask::_handle_rowset(
 
         if (invisible_rowsets == nullptr) {
             status = CloudTablet::update_delete_bitmap(tablet, &txn_info, transaction_id,
-                                                       txn_expiration);
+                                                       txn_expiration, nullptr, changelog_layout);
         } else {
             txn_info.is_txn_load = true;
             txn_info.invisible_rowsets = *invisible_rowsets;
             txn_info.lock_id = _transaction_id;
             txn_info.next_visible_version = _version;
             status = CloudTablet::update_delete_bitmap(tablet, &txn_info, transaction_id,
-                                                       txn_expiration, tablet_delete_bitmap);
+                                                       txn_expiration, tablet_delete_bitmap,
+                                                       changelog_layout);
         }
     }
     if (status != Status::OK()) {

be/src/cloud/cloud_rowset_builder.cpp

@@ -150,6 +150,13 @@ Status CloudGroupRowsetBuilder::init() {
         auto& changelog_ctx =
                 const_cast<RowsetWriterContext&>(_changelog_builder->rowset_writer()->context());
         auto& cfg = changelog_ctx.write_changelog_opt().write_changelog_config();
+        const auto& mapping = _data_builder->changelog_column_mappings();
+        if (mapping == nullptr) {
+            return Status::InvalidArgument("source index misses changelog column mappings");
+        }
+        RETURN_IF_ERROR(changelog::resolve_layout(
+                data_ctx.tablet_schema, changelog_ctx.tablet_schema, *mapping, &cfg.layout));
+        cfg.write_before = cfg.layout->need_historical_value;
         cfg.source.tablet_schema = data_ctx.tablet_schema;
         cfg.source.partial_update_info = data_ctx.partial_update_info;
         cfg.source.mow_context = data_ctx.mow_context;
@@ -194,6 +201,7 @@ Status CloudGroupRowsetBuilder::commit_rowset(const std::string& job_id, int64_t
 
 Status CloudGroupRowsetBuilder::set_txn_related_info() {
     ChangelogTxnInfo attach_changelog;
+    attach_changelog.column_mappings = _data_builder->changelog_column_mappings();
     attach_changelog.rowset = _changelog_builder->rowset();
     attach_changelog.tablet = _changelog_builder->tablet_sptr();
     if (_data_builder->tablet()->enable_unique_key_merge_on_write()) {

be/src/cloud/cloud_schema_change_job.cpp

@@ -245,8 +245,7 @@ Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& reque
 
     ReadSchemaSPtr read_schema = std::make_shared<ReadSchema>(_base_tablet_schema->columns());
     RETURN_IF_ERROR(read_schema->init_from_tablet_schema(*_base_tablet_schema,
-                                                         /*merge_by_sequence_mapping=*/false,
-                                                         /*map_changelog_columns=*/false));
+                                                         /*merge_by_sequence_mapping=*/false));
 
     // delete handlers to filter out deleted rows
     DeleteHandler delete_handler;

be/src/cloud/cloud_tablet.cpp

@@ -850,9 +850,6 @@ Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_rowset_writer(
     context.partition_id = partition_id();
     context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();
     context.encrypt_algorithm = tablet_meta()->encryption_algorithm();
-    if (context.write_changelog_opt().enable) {
-        context.write_changelog_opt().set_need_before(binlog_config().need_historical_value());
-    }
     context.inverted_index_storage_format = tablet_meta()->inverted_index_storage_format();
     context.persist_inverted_index_storage_format =
             tablet_meta()->has_inverted_index_storage_format();
@@ -895,7 +892,6 @@ Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_transient_rowset_write
     context.is_transient_rowset_writer = true;
     if (rowset.rowset_meta() != nullptr && rowset.rowset_meta()->is_changelog()) {
         context.write_changelog_opt().enable = true;
-        context.write_changelog_opt().set_need_before(binlog_config().need_historical_value());
     }
     context.rowset_id = rowset.rowset_id();
     context.tablet_id = tablet_id();

be/src/exec/scan/olap_scanner.cpp

@@ -28,6 +28,7 @@
 #include <atomic>
 #include <ostream>
 #include <set>
+#include <unordered_map>
 
 #include "cloud/cloud_storage_engine.h"
 #include "cloud/cloud_tablet_hotspot.h"
@@ -656,7 +657,81 @@ Status OlapScanner::_init_read_schema() {
     // every storage key, sequence, TSO and changelog column required below.
     _tablet_reader_params.read_schema =
             std::make_shared<ReadSchema>(std::move(read_columns), std::move(expected_types));
+    const auto& scan_node = assert_cast<OlapScanLocalState*>(_local_state)->olap_scan_node();
+    return _init_changelog_read_mappings(
+            scan_node, _output_tuple_desc->slots(), *_tablet_reader_params.tablet_schema,
+            _tablet_reader_params.changelog_scan_type, *_tablet_reader_params.read_schema);
+}
 
+Status OlapScanner::_init_changelog_read_mappings(const TOlapScanNode& scan_node,
+                                                  const std::vector<SlotDescriptor*>& slots,
+                                                  const TabletSchema& tablet_schema,
+                                                  TChangelogScanType::type scan_type,
+                                                  ReadSchema& read_schema) {
+    const bool merge_scan =
+            scan_type == TChangelogScanType::DETAIL || scan_type == TChangelogScanType::MIN_DELTA;
+    // Missing mappings are a malformed merge plan. Explicit empty pairs are valid for
+    // key-only reads and DETAIL on a table that does not record historical values.
+    if (merge_scan && (!scan_node.__isset.changelog_column_mappings ||
+                       !scan_node.changelog_column_mappings.__isset.pairs)) {
+        return Status::InvalidArgument("Changelog merge scan requires explicit read mappings");
+    }
+    DORIS_CHECK_EQ(slots.size(), read_schema.num_block_columns());
+    std::unordered_map<int32_t, ColumnId> slot_ordinals;
+    for (ColumnId ordinal = 0; ordinal < slots.size(); ++ordinal) {
+        if (!slot_ordinals.emplace(slots[ordinal]->id(), ordinal).second) {
+            return Status::InvalidArgument("Duplicate changelog scan slot {}",
+                                           slots[ordinal]->id());
+        }
+    }
+    ReadSchema::ChangelogValueColumnPairs pairs;
+    if (scan_node.__isset.changelog_column_mappings) {
+        for (const auto& pair : scan_node.changelog_column_mappings.pairs) {
+            if (!pair.__isset.current_slot_id || !pair.__isset.before_slot_id) {
+                return Status::InvalidArgument("Changelog read pair requires both slot IDs");
+            }
+            const auto current = slot_ordinals.find(pair.current_slot_id);
+            const auto before = slot_ordinals.find(pair.before_slot_id);
+            if (current == slot_ordinals.end() || before == slot_ordinals.end()) {
+                return Status::InvalidArgument("Unknown changelog read slot pair ({}, {})",
+                                               pair.current_slot_id, pair.before_slot_id);
+            }
+            pairs.emplace_back(current->second, before->second);
+        }
+    }
+    RETURN_IF_ERROR(read_schema.init_changelog_column_mappings(tablet_schema, pairs));
+    if (scan_type != TChangelogScanType::NONE &&
+        (read_schema.tso_ordinal() < 0 || read_schema.op_ordinal() < 0)) {
+        return Status::InvalidArgument("Changelog scan requires TSO and OP columns");
+    }
+    if (scan_type == TChangelogScanType::MIN_DELTA) {
+        std::unordered_set<ColumnId> paired;
+        for (const auto& [current, before] : pairs) {
+            paired.insert(current);
+            paired.insert(before);
+        }
+        for (ColumnId cid = 0; cid < tablet_schema.num_columns(); ++cid) {
+            if (static_cast<int32_t>(cid) == tablet_schema.changelog_tso_col_idx() ||
+                static_cast<int32_t>(cid) == tablet_schema.changelog_lsn_col_idx() ||
+                static_cast<int32_t>(cid) == tablet_schema.changelog_op_col_idx()) {
+                continue;
+            }
+            const auto& column = tablet_schema.column(cid);
+            const int32_t ordinal = read_schema.ordinal_by_uid(column.unique_id());
+            // Even an unprojected value may change: comparing only v1 would incorrectly
+            // suppress an UPDATE that changed v2. Require the entire original value shape.
+            if (ordinal < 0 || static_cast<size_t>(ordinal) >= read_schema.num_block_columns() ||
+                (!column.is_key() && !paired.contains(cast_set<ColumnId>(ordinal))) ||
+                !remove_nullable(read_schema.data_type(ordinal))
+                         ->equals(*remove_nullable(column.get_vec_type())) ||
+                remove_nullable(read_schema.data_type(ordinal))->get_name() !=
+                        remove_nullable(column.get_vec_type())->get_name()) {
+                return Status::InvalidArgument(
+                        "MIN_DELTA requires complete current and before values: column uid {}",
+                        column.unique_id());
+            }
+        }
+    }
     return Status::OK();
 }
 

be/src/exec/scan/olap_scanner.h

@@ -47,6 +47,9 @@ struct OlapScanRange;
 class RuntimeProfile;
 class RuntimeState;
 class TPaloScanRange;
+class TOlapScanNode;
+class ReadSchema;
+class SlotDescriptor;
 class ScanLocalStateBase;
 struct FilterPredicates;
 #ifndef NDEBUG
@@ -118,6 +121,11 @@ class OlapScanner : public Scanner {
 
     [[nodiscard]] Status _init_tso_predicates();
     [[nodiscard]] Status _init_read_schema();
+    static Status _init_changelog_read_mappings(const TOlapScanNode& scan_node,
+                                                const std::vector<SlotDescriptor*>& slots,
+                                                const TabletSchema& tablet_schema,
+                                                TChangelogScanType::type scan_type,
+                                                ReadSchema& read_schema);
     [[nodiscard]] Status _init_variant_columns();
 #ifndef NDEBUG
     Status _check_ann_cache_hit_debug_points(const OlapReaderStatistics& stats);

be/src/storage/changelog.h

@@ -40,11 +40,36 @@
 namespace doris {
 
 class AutoIncIDBuffer;
+class PChangelogWriteColumnMappings;
+class TChangelogWriteColumnMappings;
 struct PartialUpdateInfo;
 struct MowContext;
 
 namespace changelog {
 
+// A layout belongs to exactly these two pinned schemas. Writers share it for
+// their operation; no CID is stored in rowset metadata or reused after recovery.
+struct ResolvedChangelogLayout {
+    TabletSchemaSPtr source_schema;
+    TabletSchemaSPtr target_schema;
+    std::vector<uint32_t> current_source_cids;
+    std::vector<uint32_t> current_target_cids;
+    std::vector<bool> current_nullable_widening;
+    std::vector<uint32_t> before_source_cids;
+    std::vector<uint32_t> before_target_cids;
+    uint32_t tso_cid;
+    uint32_t lsn_cid;
+    uint32_t op_cid;
+    bool need_historical_value;
+};
+
+Status validate_mapping(const PChangelogWriteColumnMappings& mapping);
+Status mapping_from_thrift(const TChangelogWriteColumnMappings& input,
+                           PChangelogWriteColumnMappings* output);
+Status resolve_layout(const TabletSchemaSPtr& source, const TabletSchemaSPtr& target,
+                      const PChangelogWriteColumnMappings& mapping,
+                      std::shared_ptr<const ResolvedChangelogLayout>* result);
+
 // ATTN(changelog-rename): formerly "__BEFORE__<col>__"
 inline std::string build_before_column_name(std::string_view name) {
     std::string before_name = "__DORIS_CHANGELOG_BEFORE_";
@@ -144,6 +169,7 @@ class SegmentAllocatedLsnMap {
 struct SegmentWriteChangelogOptions {
 public:
     bool write_before = false;
+    std::shared_ptr<const changelog::ResolvedChangelogLayout> layout;
 
     // source context, used for retrieving historical row and building changelog block
     struct SourceWriteDataOptions {

be/src/storage/data_dir.cpp

@@ -495,9 +495,15 @@ Status DataDir::load() {
             LOG(WARNING) << "parse pending publish info failed, tablet_id: " << tablet_id
                          << " publish_version: " << publish_version;
         }
-        engine.add_async_publish_task(pending_publish_info_pb.partition_id(), tablet_id,
-                                      publish_version, pending_publish_info_pb.transaction_id(),
-                                      true, pending_publish_info_pb.commit_tso());
+        CHECK(engine.add_async_publish_task(
+                            pending_publish_info_pb.partition_id(), tablet_id, publish_version,
+                            pending_publish_info_pb.transaction_id(), true,
+                            pending_publish_info_pb.commit_tso(),
+                            pending_publish_info_pb.has_changelog_column_mappings()
+                                    ? std::make_shared<const PChangelogWriteColumnMappings>(
+                                              pending_publish_info_pb.changelog_column_mappings())
+                                    : nullptr)
+                      .ok());
         return true;
     };
     MonotonicStopWatch pending_publish_timer;

be/src/storage/iterator/block_reader.cpp

@@ -555,19 +555,18 @@ Status BlockReader::init(const ReaderParams& read_params) {
     SCOPED_RAW_TIMER(&_stats.tablet_reader_init_timer_ns);
     RETURN_IF_ERROR(TabletReader::init(read_params));
 
-    // A changelog scan maps the before-image columns; every other read of this reader, which is
-    // the one that merges rows across rowsets, builds the sequence mapping instead.
-    const bool map_changelog_columns =
+    // Scanner already resolved changelog pairs. Other merge reads may need a sequence mapping.
+    const bool is_changelog_merge_scan =
             read_params.changelog_scan_type == TChangelogScanType::MIN_DELTA ||
             read_params.changelog_scan_type == TChangelogScanType::DETAIL;
     const bool merge_by_sequence_mapping =
-            !map_changelog_columns && _tablet_schema->has_seq_map() &&
+            !is_changelog_merge_scan && _tablet_schema->has_seq_map() &&
             _tablet_schema->keys_type() == UNIQUE_KEYS && !_direct_mode &&
             !(read_params.reader_type == ReaderType::READER_QUERY &&
               _tablet->enable_unique_key_merge_on_write());
-    RETURN_IF_ERROR(_read_schema->init_from_tablet_schema(
-            *_tablet_schema, merge_by_sequence_mapping, map_changelog_columns));
-    if (map_changelog_columns) {
+    RETURN_IF_ERROR(
+            _read_schema->init_from_tablet_schema(*_tablet_schema, merge_by_sequence_mapping));
+    if (is_changelog_merge_scan) {
         _min_delta_value_compare_unsupported = false;
     }
 

be/src/storage/iterator/vertical_block_reader.cpp

@@ -270,8 +270,7 @@ Status VerticalBlockReader::init(const ReaderParams& read_params,
     RETURN_IF_ERROR(TabletReader::init(read_params));
 
     RETURN_IF_ERROR(_read_schema->init_from_tablet_schema(*_tablet_schema,
-                                                          /*merge_by_sequence_mapping=*/false,
-                                                          /*map_changelog_columns=*/false));
+                                                          /*merge_by_sequence_mapping=*/false));
 
     auto status = _init_collect_iter(read_params, sample_info);
     if (!status.ok()) [[unlikely]] {