[fix](cloud)(restore) fix broken schema during restore of lsc=false tables

When restoring a `light_schema_change=false` table from a shared-nothing
cluster to a shared-storage cluster, the tablet schema created by
`create_tablet` has all column `unique_id` set to -1. Later,
`commit_restore_job` writes rowset meta that contains the correct schema
(with valid `unique_id >= 0`). However, `put_schema_kv` skips writing
when the schema key already exists, leaving the broken schema
(`unique_id = -1`) in place and causing subsequent reads to fail.

Fix:
1. Introduce a new helper `put_schema_kv_on_restore` in
   `meta_service_schema.{h,cpp}`. It checks the existing schema value;
   if the key does not exist, or the existing schema is unparsable, or
   the first column's `unique_id` is -1 (i.e. broken), it removes the
   old key (range-remove to cover blob format) and writes the correct
   schema. The original `put_schema_kv` interface is kept untouched.
2. In `commit_restore_job`, replace `put_schema_kv` with
   `put_schema_kv_on_restore` for both rowset-meta and tablet-meta
   branches. A `std::set<std::string> restored_schema_keys` is added to
   avoid redundant FDB get/put for the same `(index_id, schema_version)`
   within a single restore job.
3. Log `rs_meta_schema_put_cnt / skip_cnt` and
   `tablet_meta_schema_put_cnt / skip_cnt` tags for observability.

Author: carolinchen-docker

cloud/src/meta-service/meta_service.cpp

@@ -41,6 +41,7 @@
 #include <memory>
 #include <numeric>
 #include <ostream>
+#include <set>
 #include <sstream>
 #include <string>
 #include <string_view>
@@ -1764,6 +1765,13 @@ void MetaServiceImpl::commit_restore_job(::google::protobuf::RpcController* cont
     TabletStats tablet_stat;
     int64_t converted_rowset_num = 0;
     int32_t max_batch_size = config::max_restore_job_rowsets_per_batch;
+    // Track schema keys that have already been restored to avoid redundant
+    // FDB get/put operations for the same (index_id, schema_version).
+    std::set<std::string> restored_schema_keys;
+    int64_t rs_meta_schema_put_cnt = 0;
+    int64_t rs_meta_schema_skip_cnt = 0;
+    int64_t tablet_meta_schema_put_cnt = 0;
+    int64_t tablet_meta_schema_skip_cnt = 0;
     for (size_t i = 0; i < restore_job_rs_metas.size(); i += max_batch_size) {
         size_t end = (i + max_batch_size) > restore_job_rs_metas.size()
                              ? restore_job_rs_metas.size()
@@ -1791,9 +1799,16 @@ void MetaServiceImpl::commit_restore_job(::google::protobuf::RpcController* cont
                         return;
                     }
                 }
-                put_schema_kv(code, msg, txn.get(), schema_key, rowset_meta.tablet_schema());
-                if (code != MetaServiceCode::OK) {
-                    return;
+                if (restored_schema_keys.count(schema_key) == 0) {
+                    put_schema_kv_on_restore(code, msg, txn.get(), schema_key,
+                                             rowset_meta.tablet_schema());
+                    if (code != MetaServiceCode::OK) {
+                        return;
+                    }
+                    restored_schema_keys.insert(schema_key);
+                    ++rs_meta_schema_put_cnt;
+                } else {
+                    ++rs_meta_schema_skip_cnt;
                 }
                 if (is_versioned_write) {
                     std::string versioned_schema_key = versioned::meta_schema_key(
@@ -2052,8 +2067,15 @@ void MetaServiceImpl::commit_restore_job(::google::protobuf::RpcController* cont
             fix_column_type(tablet_meta->mutable_schema());
             auto schema_key = meta_schema_key(
                     {instance_id, tablet_meta->index_id(), tablet_meta->schema_version()});
-            put_schema_kv(code, msg, txn0.get(), schema_key, tablet_meta->schema());
-            if (code != MetaServiceCode::OK) return;
+            if (restored_schema_keys.count(schema_key) == 0) {
+                put_schema_kv_on_restore(code, msg, txn0.get(), schema_key,
+                                         tablet_meta->schema());
+                if (code != MetaServiceCode::OK) return;
+                restored_schema_keys.insert(schema_key);
+                ++tablet_meta_schema_put_cnt;
+            } else {
+                ++tablet_meta_schema_skip_cnt;
+            }
 
             bool is_versioned_write = is_version_write_enabled(instance_id);
             if (is_versioned_write) {
@@ -2161,7 +2183,11 @@ void MetaServiceImpl::commit_restore_job(::google::protobuf::RpcController* cont
             .tag("tablet_id", tablet_idx.tablet_id())
             .tag("state", restore_job_pb.state())
             .tag("mtime_s", restore_job_pb.mtime_s())
-            .tag("committed_rowset_num", converted_rowset_num);
+            .tag("committed_rowset_num", converted_rowset_num)
+            .tag("rs_meta_schema_put_cnt", rs_meta_schema_put_cnt)
+            .tag("rs_meta_schema_skip_cnt", rs_meta_schema_skip_cnt)
+            .tag("tablet_meta_schema_put_cnt", tablet_meta_schema_put_cnt)
+            .tag("tablet_meta_schema_skip_cnt", tablet_meta_schema_skip_cnt);
     err = txn0->commit();
     if (err != TxnErrorCode::TXN_OK) {
         code = cast_as<ErrCategory::COMMIT>(err);

cloud/src/meta-service/meta_service_schema.cpp

@@ -17,6 +17,8 @@
 
 #include "meta-service/meta_service_schema.h"
 
+#include "meta-store/codec.h"
+
 #include <fmt/format.h>
 #include <gen_cpp/cloud.pb.h>
 #include <gen_cpp/olap_file.pb.h>
@@ -156,6 +158,39 @@ void put_versioned_schema_kv(MetaServiceCode& code, std::string& msg, Transactio
     }
 }
 
+void put_schema_kv_on_restore(MetaServiceCode& code, std::string& msg, Transaction* txn,
+                              std::string_view schema_key,
+                              const doris::TabletSchemaCloudPB& schema) {
+    bool need_put = false;
+    ValueBuf val_buf;
+    TxnErrorCode err = cloud::blob_get(txn, schema_key, &val_buf);
+    if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+        need_put = true;
+    } else if (err == TxnErrorCode::TXN_OK) {
+        // Overwrite if existing schema is corrupted or has bad unique_id from create_tablet
+        doris::TabletSchemaCloudPB saved_schema;
+        need_put = !parse_schema_value(val_buf, &saved_schema) ||
+                   (saved_schema.column_size() > 0 &&
+                    saved_schema.column(0).unique_id() == -1);
+    } else {
+        code = cast_as<ErrCategory::READ>(err);
+        msg = fmt::format("failed to get schema during restore, err={}", err);
+        return;
+    }
+    if (need_put) {
+        std::string schema_key_end(schema_key);
+        encode_int64(INT64_MAX, &schema_key_end);
+        txn->remove(schema_key, schema_key_end);
+        uint8_t ver = config::meta_schema_value_version;
+        if (ver > 0) {
+            cloud::blob_put(txn, schema_key, schema, ver);
+        } else {
+            txn->put(schema_key, schema.SerializeAsString());
+        }
+        LOG(INFO) << "put schema during restore, key=" << hex(schema_key);
+    }
+}
+
 bool parse_schema_value(const ValueBuf& buf, doris::TabletSchemaCloudPB* schema) {
     // TODO(plat1ko): Apply decompression based on value version
     return buf.to_pb(schema);

cloud/src/meta-service/meta_service_schema.h

@@ -30,6 +30,12 @@ void put_schema_kv(MetaServiceCode& code, std::string& msg, Transaction* txn,
 void put_versioned_schema_kv(MetaServiceCode& code, std::string& msg, Transaction* txn,
                              std::string_view schema_key, const doris::TabletSchemaCloudPB& schema);
 
+// Put schema during restore. Only overwrite if existing schema has broken
+// unique_id=-1 (from create_tablet) or doesn't exist yet.
+void put_schema_kv_on_restore(MetaServiceCode& code, std::string& msg, Transaction* txn,
+                              std::string_view schema_key,
+                              const doris::TabletSchemaCloudPB& schema);
+
 // Return true if parse success
 [[nodiscard]] bool parse_schema_value(const ValueBuf& buf, doris::TabletSchemaCloudPB* schema);