[fix](wal) Fix WAL replay loading 0 rows due to columns being marked as missing

Author: mymeiyi

be/src/load/group_commit/wal/wal_reader.cpp

@@ -41,16 +41,16 @@ Status WalReader::init_reader(const TupleDescriptor* tuple_descriptor) {
 
 // ---- Unified init_reader(ReaderInitContext*) overrides ----
 
-Status WalReader::_open_file_reader(ReaderInitContext* /*ctx*/) {
+Status WalReader::_open_file_reader(ReaderInitContext* ctx) {
+    auto* wal_ctx = checked_context_cast<WalInitContext>(ctx);
+    _tuple_descriptor = wal_ctx->output_tuple_descriptor;
     RETURN_IF_ERROR(_state->exec_env()->wal_mgr()->get_wal_path(_wal_id, _wal_path));
     _wal_reader = std::make_shared<doris::WalFileReader>(_wal_path);
     RETURN_IF_ERROR(_wal_reader->init());
     return Status::OK();
 }
 
-Status WalReader::_do_init_reader(ReaderInitContext* base_ctx) {
-    auto* ctx = checked_context_cast<WalInitContext>(base_ctx);
-    _tuple_descriptor = ctx->output_tuple_descriptor;
+Status WalReader::_do_init_reader(ReaderInitContext* /*base_ctx*/) {
     return Status::OK();
 }
 
@@ -131,6 +131,15 @@ Status WalReader::_get_columns_impl(std::unordered_map<std::string, DataTypePtr>
     } catch (const std::invalid_argument& e) {
         return Status::InvalidArgument("Invalid format, {}", e.what());
     }
+    // Report WAL columns so on_before_init_reader does not mark them as missing.
+    if (_tuple_descriptor) {
+        for (auto* slot_desc : _tuple_descriptor->slots()) {
+            if (_column_pos_map.contains(slot_desc->col_unique_id())) {
+                name_to_type->emplace(slot_desc->col_name(),
+                                      slot_desc->get_data_type_ptr());
+            }
+        }
+    }
     return Status::OK();
 }