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

Problem
-------
In cloud mode, when a snapshot of a table created with
`light_schema_change = false` is restored, the schema KV persisted under
`meta_schema_key({instance_id, index_id, schema_version})` ends up with
every column carrying `unique_id = -1`, because the schema is first
written by `create_tablet` and at that point the unique_ids are not yet
assigned. `commit_restore_job` then receives the correct schema (with
valid `unique_id >= 0`) in each rowset meta from the backup, but the
existing `put_schema_kv` is a no-op when the key already exists, so the
broken schema leaks through and subsequent reads fail with errors such
as `column reader is nullptr` or `different type between schema and
column reader`.

Fix
---
Introduce `put_schema_kv_on_restore()` in the cloud MetaService. It
reads the existing schema value, detects the broken-schema signature
(`column(0).unique_id() == -1` or an unparseable value), range-removes
all chunks of the stale schema, and writes the correct one. To avoid
replacing a bad schema with another bad one, it also refuses to write
a schema that is itself broken (empty columns or `column(0).unique_id
== -1`) and only logs a warning in that case.

In `MetaServiceImpl::commit_restore_job`, replace both existing
`put_schema_kv` call sites with `put_schema_kv_on_restore`, guarded by
an in-RPC `std::set<std::string>` so the same
`(index_id, schema_version)` pair does not issue redundant FDB
reads/writes when the restore spans many rowsets. Four counters
(`rs_meta_schema_put_cnt/skip_cnt`,
`tablet_meta_schema_put_cnt/skip_cnt`) are logged at the end of the
RPC to make the behaviour observable in production.

`put_versioned_schema_kv()` is intentionally NOT wrapped by the dedup
set: it targets a different key space (`versioned::meta_schema_key`)
and is already skip-if-exists inside the function.

Tests
-----
`cloud/test/meta_service_test.cpp` adds 5 unit tests covering every
branch of `put_schema_kv_on_restore`:
  - `PutWhenKeyNotExist`                     — first-write path
  - `NoopWhenExistingSchemaIsGood`           — skip-if-healthy path
  - `OverwriteWhenExistingIsBroken`          — the fix itself
  - `DefensiveSkipWhenIncomingHasEmptyColumns`  — defensive guard
  - `DefensiveSkipWhenIncomingHasUidNegativeOne` — defensive guard

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,15 @@ 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 for which `put_schema_kv_on_restore` has already been
+    // invoked within this RPC, so the same (index_id, schema_version) does not
+    // issue redundant FDB reads/writes across rowsets. Only covers the
+    // `meta_schema_key` path; `put_versioned_schema_kv` remains independent.
+    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 +1801,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 +2069,14 @@ 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 +2184,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

@@ -34,6 +34,7 @@
 #include "cpp/sync_point.h"
 #include "meta-service/meta_service_helper.h"
 #include "meta-store/blob_message.h"
+#include "meta-store/codec.h"
 #include "meta-store/document_message.h"
 #include "meta-store/keys.h"
 #include "meta-store/txn_kv.h"
@@ -133,6 +134,57 @@ void put_schema_kv(MetaServiceCode& code, std::string& msg, Transaction* txn,
     }
 }
 
+void put_schema_kv_on_restore(MetaServiceCode& code, std::string& msg, Transaction* txn,
+                              std::string_view schema_key,
+                              const doris::TabletSchemaCloudPB& schema) {
+    // Decide whether we need to (re)write the schema at this key.
+    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 the existing schema value cannot be parsed, or if its
+        // first column has unique_id == -1 which is the signature of a broken
+        // schema written by create_tablet for a light_schema_change=false
+        // table during restore.
+        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) {
+        return;
+    }
+    // Defensive check: refuse to overwrite with a schema that is itself
+    // broken (empty columns, or column(0).unique_id == -1). This guarantees
+    // we never replace a bad schema with another bad one. On rejection we
+    // only log, so callers can continue committing other work.
+    if (schema.column_size() == 0 || schema.column(0).unique_id() == -1) {
+        LOG_WARNING("skip put schema during restore, incoming schema is broken")
+                .tag("key", hex(schema_key))
+                .tag("column_size", schema.column_size());
+        return;
+    }
+    // `put_schema_kv` stores the schema as either a single KV (when
+    // meta_schema_value_version == 0) or multiple blob chunks keyed by
+    // `schema_key + encode_int64(ver << 56 + i)`. To clean up all possible
+    // existing chunks we do a range remove over `[schema_key, schema_key + INT64_MAX)`.
+    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").tag("key", hex(schema_key));
+}
+
 void put_versioned_schema_kv(MetaServiceCode& code, std::string& msg, Transaction* txn,
                              std::string_view schema_key,
                              const doris::TabletSchemaCloudPB& schema) {

cloud/src/meta-service/meta_service_schema.h

@@ -27,6 +27,15 @@ struct ValueBuf;
 void put_schema_kv(MetaServiceCode& code, std::string& msg, Transaction* txn,
                    std::string_view schema_key, const doris::TabletSchemaCloudPB& schema);
 
+// Put schema during a restore job. Unlike put_schema_kv, this function will
+// overwrite an existing schema key whose contents are broken (cannot be
+// parsed, or column(0).unique_id == -1 produced by create_tablet for tables
+// with light_schema_change=false). It also refuses to write a new schema
+// that is itself broken, to avoid replacing a bad schema with another bad one.
+void put_schema_kv_on_restore(MetaServiceCode& code, std::string& msg, Transaction* txn,
+                              std::string_view schema_key,
+                              const doris::TabletSchemaCloudPB& schema);
+
 void put_versioned_schema_kv(MetaServiceCode& code, std::string& msg, Transaction* txn,
                              std::string_view schema_key, const doris::TabletSchemaCloudPB& schema);