[improvement](cloud) Add blob key codec helpers (#67013)

cloud/src/meta-store/blob_message.cpp

@@ -21,11 +21,13 @@
 #include <butil/iobuf_inl.h>
 #include <google/protobuf/message.h>
 
+#include <algorithm>
 #include <cstdint>
 
 #include "common/logging.h"
 #include "common/util.h"
 #include "meta-store/codec.h"
+#include "meta-store/keys.h"
 #include "meta-store/txn_kv.h"
 
 namespace doris::cloud {
@@ -40,6 +42,15 @@ static std::vector<std::string_view> split_string(const std::string_view& str, i
     return substrings;
 }
 
+std::string encode_blob_key(std::string_view origin_key, uint8_t version, size_t sequence) {
+    std::string split_key(origin_key);
+    int64_t suffix = version;
+    suffix <<= 56;
+    suffix += sequence;
+    encode_int64(suffix, &split_key);
+    return split_key;
+}
+
 bool ValueBuf::to_pb(google::protobuf::Message* pb) const {
     butil::IOBuf merge;
     for (auto&& it : iters) {
@@ -150,13 +161,10 @@ void blob_put(Transaction* txn, std::string_view key, const google::protobuf::Me
 
 void blob_put(Transaction* txn, std::string_view key, std::string_view value, uint8_t ver,
               size_t split_size) {
+    split_size = std::max(split_size, MIN_BLOB_SPLIT_SIZE);
     auto split_vec = split_string(value, split_size);
-    int64_t suffix_base = ver;
-    suffix_base <<= 56;
     for (size_t i = 0; i < split_vec.size(); ++i) {
-        std::string k(key);
-        encode_int64(suffix_base + i, &k);
-        txn->put(k, split_vec[i]);
+        txn->put(encode_blob_key(key, ver, i), split_vec[i]);
     }
 }
 
@@ -240,31 +248,47 @@ void BlobIterator::load_current_blob() {
     }
 }
 
-bool BlobIterator::extract_origin_key(std::string_view raw_key, std::string* output,
-                                      uint8_t* version, uint16_t* sequence) {
-    // The suffix is 8 bytes: |version(1)|dummy(5)|sequence(2)|
+bool decode_blob_key(
+        std::string_view raw_key, std::string* origin_key, uint8_t* version, uint16_t* sequence,
+        std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>>* fields) {
     if (raw_key.size() < 9) {
-        LOG_WARNING("failed to extract origin key").tag("key", hex(raw_key));
-        error_code_ = TxnErrorCode::TXN_INVALID_DATA;
         return false;
     }
 
     const size_t origin_key_size = raw_key.size() - 9;
-    std::string_view origin_key = raw_key.substr(0, origin_key_size);
+    std::string_view decoded_origin_key = raw_key.substr(0, origin_key_size);
     raw_key.remove_prefix(origin_key_size);
     int64_t suffix = 0;
     if (decode_int64(&raw_key, &suffix) != 0) {
-        LOG_WARNING("failed to decode int64")
-                .tag("key", hex(raw_key))
-                .tag("origin_key", hex(origin_key));
-        error_code_ = TxnErrorCode::TXN_INVALID_DATA;
         return false;
     }
 
+    if (fields != nullptr) {
+        if (decoded_origin_key.size() <= 1) {
+            return false;
+        }
+        auto encoded_origin_key = decoded_origin_key;
+        encoded_origin_key.remove_prefix(1);
+        if (decode_key(&encoded_origin_key, fields) != 0) {
+            return false;
+        }
+    }
+
     *version = (suffix >> 56) & 0xff;
     *sequence = suffix & 0xffff;
-    *output = std::string(origin_key);
+    if (origin_key != nullptr) {
+        *origin_key = std::string(decoded_origin_key);
+    }
+    return true;
+}
 
+bool BlobIterator::extract_origin_key(std::string_view raw_key, std::string* output,
+                                      uint8_t* version, uint16_t* sequence) {
+    if (!decode_blob_key(raw_key, output, version, sequence)) {
+        LOG_WARNING("failed to extract origin key").tag("key", hex(raw_key));
+        error_code_ = TxnErrorCode::TXN_INVALID_DATA;
+        return false;
+    }
     return true;
 }
 
@@ -296,17 +320,14 @@ namespace versioned {
 
 void blob_put(Transaction* txn, std::string_view key, std::string_view value, uint8_t ver,
               size_t split_size) {
+    split_size = std::max(split_size, MIN_BLOB_SPLIT_SIZE);
     std::string encoded_key(key);
     uint32_t offset = encode_versionstamp(Versionstamp::min(), &encoded_key);
     encode_versionstamp_end(&encoded_key);
 
     auto split_vec = split_string(value, split_size);
-    int64_t suffix_base = ver;
-    suffix_base <<= 56;
     for (size_t i = 0; i < split_vec.size(); ++i) {
-        std::string k(encoded_key);
-        encode_int64(suffix_base + i, &k);
-        txn->atomic_set_ver_key(k, offset, split_vec[i]);
+        txn->atomic_set_ver_key(encode_blob_key(encoded_key, ver, i), offset, split_vec[i]);
     }
 }
 

cloud/src/meta-store/blob_message.h

@@ -17,8 +17,12 @@
 
 #pragma once
 
+#include <cstddef>
 #include <string>
 #include <string_view>
+#include <tuple>
+#include <variant>
+#include <vector>
 
 #include "meta-store/txn_kv.h"
 #include "meta-store/txn_kv_error.h"
@@ -39,6 +43,8 @@ namespace doris::cloud {
  *    |Bytes 0      |Bytes 1-5    |Bytes 6-7    |
  *    |-------------|-------------|-------------|
  *    |version      |dummy        |sequence     |
+ * version is uint8_t restricted to 0-127; sequence is uint16_t (0-65535).
+ * Reserved dummy bytes can extend sequence if needed.
  */
 struct ValueBuf {
     // TODO(plat1ko): Support decompression
@@ -59,9 +65,35 @@ struct ValueBuf {
     std::vector<std::string> keys() const;
 
     std::vector<std::unique_ptr<RangeGetIterator>> iters;
-    int8_t ver {-1};
+    int16_t ver {-1};
 };
 
+inline constexpr size_t DEFAULT_BLOB_SPLIT_SIZE = 90 * 1000;
+// A 10 MB FDB transaction produces at most 500 KVs with 20 KB splits, within uint16_t.
+inline constexpr size_t MIN_BLOB_SPLIT_SIZE = 20 * 1000;
+
+/**
+ * Encode a split KV key.
+ * @param origin_key logical key
+ * @param version value encoding version; valid range is [0, 127]
+ * @param sequence zero-based split KV index; valid range is [0, 65535]
+ * @return encoded split KV key
+ */
+std::string encode_blob_key(std::string_view origin_key, uint8_t version, size_t sequence);
+
+/**
+ * Decode a split KV key.
+ * @param raw_key encoded split KV key
+ * @param origin_key optional logical key output
+ * @param version value encoding version output, must not be nullptr
+ * @param sequence zero-based split KV index output, must not be nullptr
+ * @param fields optional decoded origin-key fields; non-null enables validation
+ * @return true on success
+ */
+bool decode_blob_key(
+        std::string_view raw_key, std::string* origin_key, uint8_t* version, uint16_t* sequence,
+        std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>>* fields = nullptr);
+
 /**
  * Get a key, return key's value, value length may be bigger than 100k
  * @param txn fdb txn handler
@@ -78,22 +110,22 @@ TxnErrorCode blob_get(Transaction* txn, std::string_view key, ValueBuf* val, boo
  * @param txn fdb txn handler
  * @param key encode key
  * @param pb value to save
- * @param ver value version
- * @param split_size how many byte sized fragments are the value split into
+ * @param ver value version; valid range is [0, 127]
+ * @param split_size fragment size, clamped to MIN_BLOB_SPLIT_SIZE
  */
 void blob_put(Transaction* txn, std::string_view key, const google::protobuf::Message& pb,
-              uint8_t ver, size_t split_size = 90 * 1000);
+              uint8_t ver, size_t split_size = DEFAULT_BLOB_SPLIT_SIZE);
 
 /**
  * Put a KV, it's value may be bigger than 100k
  * @param txn fdb txn handler
  * @param key encode key
  * @param value value to save
- * @param ver value version
- * @param split_size how many byte sized fragments are the value split into
+ * @param ver value version; valid range is [0, 127]
+ * @param split_size fragment size, clamped to MIN_BLOB_SPLIT_SIZE
  */
 void blob_put(Transaction* txn, std::string_view key, std::string_view value, uint8_t ver,
-              size_t split_size = 90 * 1000);
+              size_t split_size = DEFAULT_BLOB_SPLIT_SIZE);
 
 // Iterator for blob key-value pairs.
 //
@@ -189,19 +221,20 @@ namespace versioned {
 
 // Put a blob message with a auto generated versionstamp.
 void blob_put(Transaction* txn, std::string_view key, const google::protobuf::Message& pb,
-              uint8_t ver = 0, size_t split_size = 90 * 1000);
+              uint8_t ver = 0, size_t split_size = DEFAULT_BLOB_SPLIT_SIZE);
 
 // Put a blob message with a auto generated versionstamp.
 void blob_put(Transaction* txn, std::string_view key, std::string_view value, uint8_t ver = 0,
-              size_t split_size = 90 * 1000);
+              size_t split_size = DEFAULT_BLOB_SPLIT_SIZE);
 
 // Put a blob message with a specified versionstamp.
 void blob_put(Transaction* txn, std::string_view key, Versionstamp v,
-              const google::protobuf::Message& pb, uint8_t ver = 0, size_t split_size = 90 * 1000);
+              const google::protobuf::Message& pb, uint8_t ver = 0,
+              size_t split_size = DEFAULT_BLOB_SPLIT_SIZE);
 
 // Put a blob message with a specified versionstamp.
 void blob_put(Transaction* txn, std::string_view key, Versionstamp v, std::string_view value,
-              uint8_t ver = 0, size_t split_size = 90 * 1000);
+              uint8_t ver = 0, size_t split_size = DEFAULT_BLOB_SPLIT_SIZE);
 
 } // namespace versioned
 

cloud/test/blob_message_test.cpp

@@ -25,6 +25,8 @@
 #include <memory>
 #include <string>
 
+#include "meta-store/codec.h"
+#include "meta-store/keys.h"
 #include "meta-store/mem_txn_kv.h"
 #include "meta-store/txn_kv.h"
 #include "meta-store/txn_kv_error.h"
@@ -64,6 +66,102 @@ static std::string dump_range(TxnKv* txn_kv, std::string_view begin = "",
     return buffer;
 }
 
+TEST(BlobMessageTest, BlobKeyCodec) {
+    const std::vector<std::tuple<std::string, uint8_t, uint16_t>> test_cases {
+            {"test_blob_key", 0, 0},
+            {std::string("\0binary\xff", 8), 127, std::numeric_limits<uint16_t>::max()},
+    };
+    for (const auto& [expected_origin_key, expected_version, expected_sequence] : test_cases) {
+        std::string origin_key;
+        uint8_t version = 0;
+        uint16_t sequence = 0;
+        auto raw_key = encode_blob_key(expected_origin_key, expected_version, expected_sequence);
+        ASSERT_TRUE(decode_blob_key(raw_key, &origin_key, &version, &sequence));
+        EXPECT_EQ(origin_key, expected_origin_key);
+        EXPECT_EQ(version, expected_version);
+        EXPECT_EQ(sequence, expected_sequence);
+        EXPECT_TRUE(decode_blob_key(raw_key, nullptr, &version, &sequence));
+    }
+
+    EXPECT_EQ(encode_blob_key("key", 0x56, 0x1234),
+              std::string("key\x12\x56\0\0\0\0\0\x12\x34", 12));
+}
+
+TEST(BlobMessageTest, BlobKeySequenceDoesNotWrap) {
+    const size_t sequence = static_cast<size_t>(std::numeric_limits<uint16_t>::max()) + 1;
+    EXPECT_NE(encode_blob_key("key", 0, 0), encode_blob_key("key", 0, sequence));
+}
+
+TEST(BlobMessageTest, ValueBufVersionAboveInt8Max) {
+    auto txn_kv = std::make_shared<MemTxnKv>();
+    ASSERT_EQ(txn_kv->init(), 0);
+    std::unique_ptr<Transaction> txn;
+    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+
+    std::string key = "version_above_int8_max";
+    std::string raw_key = key;
+    encode_int64(-(int64_t {1} << 56), &raw_key); // version 255, sequence 0
+    txn->put(raw_key, "value");
+    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+    ValueBuf value;
+    ASSERT_EQ(blob_get(txn.get(), key, &value), TxnErrorCode::TXN_OK);
+    EXPECT_EQ(value.ver, std::numeric_limits<uint8_t>::max());
+    EXPECT_EQ(value.value(), "value");
+}
+
+TEST(BlobMessageTest, ClampSmallSplitSize) {
+    auto txn_kv = std::make_shared<MemTxnKv>();
+    ASSERT_EQ(txn_kv->init(), 0);
+    std::unique_ptr<Transaction> txn;
+    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+
+    std::string value(MIN_BLOB_SPLIT_SIZE + 1, 'a');
+    blob_put(txn.get(), "key", value, 0, 1);
+    EXPECT_EQ(txn->num_put_keys(), 2);
+    versioned::blob_put(txn.get(), "versioned_key", value, 0, 1);
+    EXPECT_EQ(txn->num_put_keys(), 4);
+}
+
+TEST(BlobMessageTest, DecodeBlobKeyFields) {
+    auto encoded_origin_key = meta_delete_bitmap_key({"instance_id", 1, "rowset_id", 2, 3});
+    std::string origin_key;
+    uint8_t version = 0;
+    uint16_t sequence = 0;
+    std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> fields;
+    ASSERT_TRUE(decode_blob_key(encode_blob_key(encoded_origin_key, 2, 123), &origin_key, &version,
+                                &sequence, &fields));
+    EXPECT_EQ(origin_key, encoded_origin_key);
+    EXPECT_EQ(version, 2);
+    EXPECT_EQ(sequence, 123);
+    ASSERT_EQ(fields.size(), 7);
+    EXPECT_EQ(std::get<std::string>(std::get<0>(fields[0])), "meta");
+    EXPECT_EQ(std::get<std::string>(std::get<0>(fields[1])), "instance_id");
+    EXPECT_EQ(std::get<std::string>(std::get<0>(fields[2])), "delete_bitmap");
+    EXPECT_EQ(std::get<int64_t>(std::get<0>(fields[3])), 1);
+    EXPECT_EQ(std::get<std::string>(std::get<0>(fields[4])), "rowset_id");
+    EXPECT_EQ(std::get<int64_t>(std::get<0>(fields[5])), 2);
+    EXPECT_EQ(std::get<int64_t>(std::get<0>(fields[6])), 3);
+}
+
+TEST(BlobMessageTest, RejectInvalidBlobKey) {
+    std::string origin_key;
+    uint8_t version = 0;
+    uint16_t sequence = 0;
+    EXPECT_FALSE(decode_blob_key("invalid", &origin_key, &version, &sequence));
+
+    std::string invalid_suffix = "origin";
+    invalid_suffix.append(9, '\0');
+    EXPECT_FALSE(decode_blob_key(invalid_suffix, &origin_key, &version, &sequence));
+
+    std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> fields;
+    EXPECT_FALSE(
+            decode_blob_key(encode_blob_key("", 2, 123), nullptr, &version, &sequence, &fields));
+    EXPECT_FALSE(decode_blob_key(encode_blob_key("invalid", 2, 123), nullptr, &version, &sequence,
+                                 &fields));
+}
+
 // Test blob_put and blob_get with small message (single KV)
 TEST(BlobMessageTest, PutGetSmallMessage) {
     auto txn_kv = std::make_shared<MemTxnKv>();
@@ -259,6 +357,32 @@ TEST(BlobMessageTest, GetRangeWithTxnKv) {
     }
 }
 
+TEST(BlobMessageTest, BlobIteratorRejectsMalformedChunks) {
+    auto expect_invalid = [](std::string_view origin_key,
+                             const std::vector<std::string>& raw_keys) {
+        auto txn_kv = std::make_shared<MemTxnKv>();
+        ASSERT_EQ(txn_kv->init(), 0);
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+        for (const auto& raw_key : raw_keys) {
+            txn->put(raw_key, "value");
+        }
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+        auto iter = blob_get_range(txn_kv, origin_key, std::string(origin_key) + "\xff");
+        EXPECT_FALSE(iter->valid());
+        EXPECT_EQ(iter->error_code(), TxnErrorCode::TXN_INVALID_DATA);
+    };
+
+    expect_invalid("sequence_gap",
+                   {encode_blob_key("sequence_gap", 0, 0), encode_blob_key("sequence_gap", 0, 2)});
+    expect_invalid("version_mismatch", {encode_blob_key("version_mismatch", 1, 0),
+                                        encode_blob_key("version_mismatch", 2, 1)});
+    std::string invalid_suffix = "invalid_suffix";
+    invalid_suffix.append(9, '\0');
+    expect_invalid("invalid_suffix", {invalid_suffix});
+}
+
 // Test blob_get_range with empty range
 TEST(BlobMessageTest, GetRangeEmpty) {
     auto txn_kv = std::make_shared<MemTxnKv>();

cloud/test/recycler_test.cpp

@@ -817,8 +817,8 @@ static void create_delete_bitmaps(Transaction* txn, int64_t tablet_id, std::stri
                 std::string val {"test_data"};
                 txn->put(key, val);
             } else {
-                std::string val(1000, 'A');
-                cloud::blob_put(txn, key, val, 0, 300);
+                std::string val(MIN_BLOB_SPLIT_SIZE + 1, 'A');
+                cloud::blob_put(txn, key, val, 0, MIN_BLOB_SPLIT_SIZE);
             }
         }
     }
@@ -6380,8 +6380,9 @@ TEST(CheckerTest, delete_bitmap_inverted_check_normal) {
                 // delete bitmaps may be spilitted into mulitiple KVs if too large
                 auto delete_bitmap_key =
                         meta_delete_bitmap_key({instance_id, tablet_id, rowset_id, ver, 0});
-                std::string delete_bitmap_val(1000, 'A');
-                cloud::blob_put(txn.get(), delete_bitmap_key, delete_bitmap_val, 0, 300);
+                std::string delete_bitmap_val(MIN_BLOB_SPLIT_SIZE + 1, 'A');
+                cloud::blob_put(txn.get(), delete_bitmap_key, delete_bitmap_val, 0,
+                                MIN_BLOB_SPLIT_SIZE);
             }
         }
         if (is_last_tablet) {
@@ -6613,8 +6614,9 @@ TEST(CheckerTest, delete_bitmap_inverted_check_abnormal) {
                 // delete bitmaps may be spilitted into mulitiple KVs if too large
                 auto delete_bitmap_key =
                         meta_delete_bitmap_key({instance_id, tablet_id, rowset_id, ver, 0});
-                std::string delete_bitmap_val(1000, 'A');
-                cloud::blob_put(txn.get(), delete_bitmap_key, delete_bitmap_val, 0, 300);
+                std::string delete_bitmap_val(MIN_BLOB_SPLIT_SIZE + 1, 'A');
+                cloud::blob_put(txn.get(), delete_bitmap_key, delete_bitmap_val, 0,
+                                MIN_BLOB_SPLIT_SIZE);
             }
         }
     }

cloud/test/schema_kv_test.cpp

@@ -963,9 +963,16 @@ TEST(AlterSchemaKVTest, BlobRemoveBeforeOverwriteTest) {
 
     // Compute the schema KV key.
     auto schema_key = meta_schema_key({instance_id, index_id, schema_version});
+    auto put_three_blob_chunks = [&](Transaction* txn, std::string_view value) {
+        const size_t split_size = value.size() / 3 + 1;
+        for (size_t offset = 0, sequence = 0; offset < value.size();
+             offset += split_size, ++sequence) {
+            txn->put(encode_blob_key(schema_key, config::meta_schema_value_version, sequence),
+                     value.substr(offset, split_size));
+        }
+    };
 
-    // Overwrite the schema KV with a small split_size to create 3 blob chunks.
-    // The normal schema (~120 bytes) with split_size=50 yields 3 chunks.
+    // Overwrite the schema KV with 3 blob chunks.
     {
         std::unique_ptr<Transaction> txn;
         ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
@@ -973,13 +980,10 @@ TEST(AlterSchemaKVTest, BlobRemoveBeforeOverwriteTest) {
         ValueBuf old_val;
         ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val), TxnErrorCode::TXN_OK);
         old_val.remove(txn.get());
-        // Write with small split_size to create 3 chunks
         doris::TabletSchemaCloudPB schema;
         ASSERT_TRUE(old_val.to_pb(&schema));
         std::string serialized = schema.SerializeAsString();
-        size_t split_size = (serialized.size() / 3) + 1; // ensures exactly 3 chunks
-        cloud::blob_put(txn.get(), schema_key, schema, config::meta_schema_value_version,
-                        split_size);
+        put_three_blob_chunks(txn.get(), serialized);
         ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
     }
 
@@ -1020,9 +1024,7 @@ TEST(AlterSchemaKVTest, BlobRemoveBeforeOverwriteTest) {
         ASSERT_TRUE(old_val.to_pb(&schema));
         schema.set_disable_auto_compaction(false);
         std::string serialized = schema.SerializeAsString();
-        size_t split_size = (serialized.size() / 3) + 1;
-        cloud::blob_put(txn.get(), schema_key, schema, config::meta_schema_value_version,
-                        split_size);
+        put_three_blob_chunks(txn.get(), serialized);
         ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
     }
     {

cloud/test/txn_kv_test.cpp

@@ -417,7 +417,7 @@ TEST(TxnKvTest, PutLargeValueTest) {
     auto key = meta_schema_key({instance_id, 10005, 1});
     std::unique_ptr<Transaction> txn;
     ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
-    doris::cloud::blob_put(txn.get(), key, schema, 1, 100);
+    doris::cloud::blob_put(txn.get(), key, schema, 1, doris::cloud::MIN_BLOB_SPLIT_SIZE);
     ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
 
     // Check get
@@ -439,7 +439,7 @@ TEST(TxnKvTest, PutLargeValueTest) {
     // Check multi range get
     sp->set_call_back("memkv::Transaction::get", [](auto&& args) {
         auto* limit = doris::try_any_cast<int*>(args[0]);
-        *limit = 100;
+        *limit = 1;
     });
     err = doris::cloud::blob_get(txn.get(), key, &val_buf);
     ASSERT_EQ(err, TxnErrorCode::TXN_OK);