nprasad2077/doris
1
0 Fork 0
mirror of https://github.com/apache/doris.git synced 2026-11-04 18:29:59 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
doris/cloud/test/schema_kv_test.cpp
T
Jingzhe Jiaand贾敬哲 cbe1308fbc [feature](storage) Support row-level TTL (#65858) 
### What problem does this PR solve?

Issue Number: close #61056  #66084
Related PR: #None

Problem Summary:

Add create-time Row TTL support with a hidden TTL column for both
source-time and direct-expiration tables. The implementation removes
dynamic ALTER enablement and dedicated full schema-change/backfill
paths, keeps TTL configured only at CREATE TABLE time, and treats TTL as
a post-merge row visibility rule composed with delete sign. Storage
keeps ordinary compaction policy while filtering expired rows during
reads and natural rewrite paths. FE/BE tests and regression cases cover
property validation, hidden column construction, tablet schema
propagation, visibility filtering, partial updates, planning behavior,
and ALTER rejection.

Co-authored-by: 贾敬哲 <jiajingzhe.1350@bytedance.com>
2026-10-29 15:01:50 +08:00

1113 lines
52 KiB
C++
Raw Permalink Blame History

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include <brpc/controller.h>
#include <gen_cpp/cloud.pb.h>
#include <gen_cpp/olap_file.pb.h>
#include <google/protobuf/arena.h>
#include <gtest/gtest.h>
#include <random>
#include "common/config.h"
#include "common/defer.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-service/meta_service_schema.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"
#include "meta-store/txn_kv_error.h"
static std::string instance_id = "schema_kv_test";
namespace doris::cloud {
extern std::unique_ptr<MetaServiceProxy> get_meta_service();
static std::string next_rowset_id() {
static int cnt = 0;
return std::to_string(++cnt);
}
static void fill_schema(doris::TabletSchemaCloudPB* schema, int32_t schema_version) {
schema->set_schema_version(schema_version);
for (int i = 0; i < 10; ++i) {
auto column = schema->add_column();
column->set_unique_id(20000 + i);
column->set_type("INT");
}
}
static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id, const std::string& rowset_id,
int32_t schema_version) {
auto tablet = req.add_tablet_metas();
tablet->set_table_id(table_id);
tablet->set_index_id(index_id);
tablet->set_partition_id(partition_id);
tablet->set_tablet_id(tablet_id);
auto schema = tablet->mutable_schema();
fill_schema(schema, schema_version);
auto first_rowset = tablet->add_rs_metas();
first_rowset->set_rowset_id(0); // required
first_rowset->set_rowset_id_v2(rowset_id);
first_rowset->set_start_version(0);
first_rowset->set_end_version(1);
first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
}
static void create_tablet(MetaServiceProxy* meta_service, int64_t db_id, int64_t table_id,
int64_t index_id, int64_t partition_id, int64_t tablet_id,
const std::string& rowset_id, int32_t schema_version) {
brpc::Controller cntl;
CreateTabletsRequest req;
CreateTabletsResponse res;
req.set_db_id(db_id);
add_tablet(req, table_id, index_id, partition_id, tablet_id, rowset_id, schema_version);
meta_service->create_tablets(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}
static void get_rowset(MetaServiceProxy* meta_service, int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id, GetRowsetResponse& res) {
brpc::Controller cntl;
GetRowsetRequest req;
auto tablet_idx = req.mutable_idx();
tablet_idx->set_table_id(table_id);
tablet_idx->set_index_id(index_id);
tablet_idx->set_partition_id(partition_id);
tablet_idx->set_tablet_id(tablet_id);
req.set_start_version(0);
req.set_end_version(-1);
req.set_cumulative_compaction_cnt(0);
req.set_base_compaction_cnt(0);
req.set_cumulative_point(2);
meta_service->get_rowset(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}
static void check_get_tablet(MetaServiceProxy* meta_service, int64_t tablet_id,
int32_t schema_version) {
brpc::Controller cntl;
GetTabletRequest req;
GetTabletResponse res;
req.set_tablet_id(tablet_id);
meta_service->get_tablet(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
ASSERT_TRUE(res.has_tablet_meta()) << tablet_id;
EXPECT_TRUE(res.tablet_meta().has_schema()) << tablet_id;
EXPECT_EQ(res.tablet_meta().schema_version(), schema_version) << tablet_id;
};
TEST(DetachSchemaKVTest, TabletTest) {
auto meta_service = get_meta_service();
// meta_service->resource_mgr().reset(); // Do not use resource manager
auto sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
SyncPoint::get_instance()->clear_all_call_backs();
};
sp->set_call_back("get_instance_id", [&](auto&& args) {
auto* ret = try_any_cast_ret<std::string>(args);
ret->first = instance_id;
ret->second = true;
});
sp->enable_processing();
// new MS write with write_schema_kv=false, old MS read
{
constexpr auto db_id = 1, table_id = 10001, index_id = 10002, partition_id = 10003,
tablet_id = 10004;
config::write_schema_kv = false;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
// check saved values in txn_kv
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string tablet_key, tablet_val;
meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
ASSERT_EQ(txn->get(tablet_key, &tablet_val), TxnErrorCode::TXN_OK);
doris::TabletMetaCloudPB saved_tablet;
ASSERT_TRUE(saved_tablet.ParseFromString(tablet_val));
EXPECT_TRUE(saved_tablet.has_schema());
std::string rowset_key, rowset_val;
meta_rowset_key({instance_id, tablet_id, 1}, &rowset_key);
ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB saved_rowset;
ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
EXPECT_TRUE(saved_rowset.has_tablet_schema());
}
// old MS write, new MS read
{
constexpr auto table_id = 10011, index_id = 10012, partition_id = 10013, tablet_id = 10014;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
doris::TabletMetaCloudPB saved_tablet;
saved_tablet.set_table_id(table_id);
saved_tablet.set_index_id(index_id);
saved_tablet.set_partition_id(partition_id);
saved_tablet.set_tablet_id(tablet_id);
saved_tablet.mutable_schema()->set_schema_version(1);
auto column = saved_tablet.mutable_schema()->add_column();
column->set_unique_id(30001);
column->set_type("INT");
std::string tablet_key, tablet_val;
meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
ASSERT_TRUE(saved_tablet.SerializeToString(&tablet_val));
txn->put(tablet_key, tablet_val);
TabletIndexPB saved_tablet_idx;
saved_tablet_idx.set_table_id(table_id);
saved_tablet_idx.set_index_id(index_id);
saved_tablet_idx.set_partition_id(partition_id);
saved_tablet_idx.set_tablet_id(tablet_id);
std::string tablet_idx_key, tablet_idx_val;
meta_tablet_idx_key({instance_id, tablet_id}, &tablet_idx_key);
ASSERT_TRUE(saved_tablet_idx.SerializeToString(&tablet_idx_val));
txn->put(tablet_idx_key, tablet_idx_val);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
// check get tablet response
check_get_tablet(meta_service.get(), tablet_id, 1);
}
auto check_new_saved_tablet_val = [](Transaction* txn, int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id,
int32_t schema_version) {
std::string tablet_key, tablet_val;
meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
ASSERT_EQ(txn->get(tablet_key, &tablet_val), TxnErrorCode::TXN_OK);
doris::TabletMetaCloudPB saved_tablet;
ASSERT_TRUE(saved_tablet.ParseFromString(tablet_val));
EXPECT_FALSE(saved_tablet.has_schema()) << tablet_id;
EXPECT_EQ(saved_tablet.schema_version(), schema_version) << tablet_id;
};
// new MS write with write_schema_kv=true, new MS read
{
constexpr auto db_id = 1, table_id = 10021, index_id = 10022, partition_id = 10023,
tablet_id = 10024;
config::write_schema_kv = true;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
// check saved values in txn_kv
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
check_new_saved_tablet_val(txn.get(), table_id, index_id, partition_id, tablet_id, 1);
std::string rowset_key, rowset_val;
meta_rowset_key({instance_id, tablet_id, 1}, &rowset_key);
ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB saved_rowset;
ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
EXPECT_FALSE(saved_rowset.has_tablet_schema());
EXPECT_EQ(saved_rowset.index_id(), index_id);
EXPECT_EQ(saved_rowset.schema_version(), 1);
// check get tablet response
check_get_tablet(meta_service.get(), tablet_id, 1);
// check get rowset response
GetRowsetResponse get_rowset_res;
get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id, get_rowset_res);
ASSERT_EQ(get_rowset_res.rowset_meta_size(), 1);
EXPECT_TRUE(get_rowset_res.rowset_meta(0).has_tablet_schema());
EXPECT_EQ(get_rowset_res.rowset_meta(0).index_id(), index_id);
EXPECT_EQ(get_rowset_res.rowset_meta(0).schema_version(), 1);
ASSERT_TRUE(get_rowset_res.has_stats());
EXPECT_EQ(get_rowset_res.stats().num_rows(), 0);
EXPECT_EQ(get_rowset_res.stats().num_rowsets(), 1);
EXPECT_EQ(get_rowset_res.stats().num_segments(), 0);
EXPECT_EQ(get_rowset_res.stats().data_size(), 0);
EXPECT_EQ(get_rowset_res.stats().index_size(), 0);
EXPECT_EQ(get_rowset_res.stats().segment_size(), 0);
}
// new MS batch create tablets with write_schema_kv=true
{
config::write_schema_kv = true;
brpc::Controller cntl;
CreateTabletsRequest req;
CreateTabletsResponse res;
req.set_db_id(1);
add_tablet(req, 10031, 10032, 10033, 100031, next_rowset_id(), 1);
add_tablet(req, 10031, 10032, 10033, 100032, next_rowset_id(), 2);
add_tablet(req, 10031, 10032, 10033, 100033, next_rowset_id(), 2);
add_tablet(req, 10031, 10032, 10033, 100034, next_rowset_id(), 3);
add_tablet(req, 10031, 10032, 10033, 100035, next_rowset_id(), 3);
add_tablet(req, 10031, 10032, 10033, 100036, next_rowset_id(), 3);
add_tablet(req, 10031, 10034, 10033, 100037, next_rowset_id(), 1);
add_tablet(req, 10031, 10034, 10033, 100038, next_rowset_id(), 2);
add_tablet(req, 10031, 10034, 10033, 100039, next_rowset_id(), 2);
meta_service->create_tablets(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
// check saved values in txn_kv
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100031, 1);
check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100032, 2);
check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100033, 2);
check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100034, 3);
check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100035, 3);
check_new_saved_tablet_val(txn.get(), 10031, 10032, 10033, 100036, 3);
check_new_saved_tablet_val(txn.get(), 10031, 10034, 10033, 100037, 1);
check_new_saved_tablet_val(txn.get(), 10031, 10034, 10033, 100038, 2);
check_new_saved_tablet_val(txn.get(), 10031, 10034, 10033, 100039, 2);
// check get tablet response
check_get_tablet(meta_service.get(), 100031, 1);
check_get_tablet(meta_service.get(), 100032, 2);
check_get_tablet(meta_service.get(), 100033, 2);
check_get_tablet(meta_service.get(), 100034, 3);
check_get_tablet(meta_service.get(), 100035, 3);
check_get_tablet(meta_service.get(), 100036, 3);
check_get_tablet(meta_service.get(), 100037, 1);
check_get_tablet(meta_service.get(), 100038, 2);
check_get_tablet(meta_service.get(), 100039, 2);
}
}
TEST(DetachSchemaKVTest, PutSchemaKvTest) {
config::meta_schema_value_version = 1;
auto meta_service = get_meta_service();
int64_t index_id = 14221;
int64_t schema_version = 0;
std::string key = meta_schema_key({instance_id, index_id, schema_version});
std::string versioned_key = versioned::meta_schema_key({instance_id, index_id, schema_version});
doris::TabletSchemaCloudPB schema;
fill_schema(&schema, schema_version);
std::unique_ptr<Transaction> txn;
MetaServiceCode code = MetaServiceCode::OK;
std::string msg;
for (int i = 0; i < 2; ++i) {
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_schema_kv(code, msg, txn.get(), key, schema);
ASSERT_EQ(code, MetaServiceCode::OK);
put_versioned_schema_kv(code, msg, txn.get(), versioned_key, schema);
ASSERT_EQ(code, MetaServiceCode::OK);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
// verify the tablet schema is written
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
doris::TabletSchemaCloudPB saved_schema;
ValueBuf buf;
ASSERT_EQ(cloud::blob_get(txn.get(), key, &buf), TxnErrorCode::TXN_OK);
// verify the versioned tablet schema is written
ASSERT_EQ(document_get(txn.get(), versioned_key, &saved_schema), TxnErrorCode::TXN_OK);
EXPECT_EQ(saved_schema.schema_version(), schema_version);
}
{
// put new schema version
schema_version = 1;
schema.set_schema_version(schema_version);
key = meta_schema_key({instance_id, index_id, schema_version});
versioned_key = versioned::meta_schema_key({instance_id, index_id, schema_version});
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_schema_kv(code, msg, txn.get(), key, schema);
ASSERT_EQ(code, MetaServiceCode::OK);
put_versioned_schema_kv(code, msg, txn.get(), versioned_key, schema);
ASSERT_EQ(code, MetaServiceCode::OK);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
// verify the tablet schema is written
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
doris::TabletSchemaCloudPB saved_schema;
ValueBuf buf;
ASSERT_EQ(cloud::blob_get(txn.get(), key, &buf), TxnErrorCode::TXN_OK);
// verify the versioned tablet schema is written
ASSERT_EQ(document_get(txn.get(), versioned_key, &saved_schema), TxnErrorCode::TXN_OK);
EXPECT_EQ(saved_schema.schema_version(), schema_version);
}
}
static void check_stored_ttl_schema(Transaction* txn, const std::string& key,
const std::string& versioned_key, const std::string& original) {
doris::TabletSchemaCloudPB saved_schema;
ValueBuf buf;
ASSERT_EQ(blob_get(txn, key, &buf), TxnErrorCode::TXN_OK);
ASSERT_TRUE(buf.to_pb(&saved_schema));
EXPECT_EQ(saved_schema.SerializeAsString(), original);
ASSERT_EQ(document_get(txn, versioned_key, &saved_schema), TxnErrorCode::TXN_OK);
EXPECT_EQ(saved_schema.SerializeAsString(), original);
}
TEST(DetachSchemaKVTest, RowTtlSchemaRoundTripPreservesColumnOrder) {
auto meta_service = get_meta_service();
constexpr int64_t index_id = 14321;
constexpr int32_t schema_version = 7;
const auto key = meta_schema_key({instance_id, index_id, schema_version});
const auto versioned_key = versioned::meta_schema_key({instance_id, index_id, schema_version});
doris::TabletSchemaCloudPB schema;
fill_schema(&schema, schema_version);
schema.mutable_column()->SwapElements(0, schema.column_size() - 1);
schema.mutable_column(0)->set_name("__DORIS_TTL_COL__");
schema.mutable_column(0)->set_type("DATETIMEV2");
schema.set_ttl_col_idx(0);
schema.set_row_ttl_duration_us(10);
schema.set_row_ttl_time_zone_offset_seconds(3600);
const auto original = schema.SerializeAsString();
// Exercise both the initial write and the existing-schema comparison in both KV formats.
for (int i = 0; i < 2; ++i) {
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
MetaServiceCode code = MetaServiceCode::OK;
std::string msg;
put_schema_kv(code, msg, txn.get(), key, schema);
ASSERT_EQ(code, MetaServiceCode::OK) << msg;
put_versioned_schema_kv(code, msg, txn.get(), versioned_key, schema);
ASSERT_EQ(code, MetaServiceCode::OK) << msg;
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
EXPECT_EQ(schema.SerializeAsString(), original);
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
check_stored_ttl_schema(txn.get(), key, versioned_key, original);
}
}
static void begin_txn(MetaServiceProxy* meta_service, int64_t db_id, const std::string& label,
int64_t table_id, int64_t& txn_id) {
brpc::Controller cntl;
BeginTxnRequest req;
BeginTxnResponse res;
auto txn_info = req.mutable_txn_info();
txn_info->set_db_id(db_id);
txn_info->set_label(label);
txn_info->add_table_ids(table_id);
txn_info->set_timeout_ms(36000);
meta_service->begin_txn(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label;
ASSERT_TRUE(res.has_txn_id()) << label;
txn_id = res.txn_id();
}
static void commit_txn(MetaServiceProxy* meta_service, int64_t db_id, int64_t txn_id,
const std::string& label) {
brpc::Controller cntl;
CommitTxnRequest req;
CommitTxnResponse res;
req.set_db_id(db_id);
req.set_txn_id(txn_id);
meta_service->commit_txn(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label;
}
static doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id,
const std::string& rowset_id, int32_t schema_version,
int64_t version = -1,
const TabletSchemaCloudPB* schema = nullptr) {
doris::RowsetMetaCloudPB rowset;
rowset.set_rowset_id(0); // required
rowset.set_rowset_id_v2(rowset_id);
rowset.set_tablet_id(tablet_id);
rowset.set_txn_id(txn_id);
rowset.set_num_rows(100);
rowset.set_num_segments(1);
rowset.set_data_disk_size(10000);
rowset.set_index_disk_size(1000);
rowset.set_total_disk_size(11000);
if (version > 0) {
rowset.set_start_version(version);
rowset.set_end_version(version);
}
rowset.mutable_tablet_schema()->set_schema_version(schema_version);
rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
if (schema != nullptr) {
rowset.mutable_tablet_schema()->CopyFrom(*schema);
rowset.mutable_tablet_schema()->set_schema_version(schema_version);
}
return rowset;
}
static void prepare_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
CreateRowsetResponse& res) {
brpc::Controller cntl;
auto arena = res.GetArena();
auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
req->mutable_rowset_meta()->CopyFrom(rowset);
meta_service->prepare_rowset(&cntl, req, &res, nullptr);
if (!arena) delete req;
}
static void commit_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
CreateRowsetResponse& res) {
brpc::Controller cntl;
auto arena = res.GetArena();
auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
req->mutable_rowset_meta()->CopyFrom(rowset);
meta_service->commit_rowset(&cntl, req, &res, nullptr);
if (!arena) delete req;
}
static void insert_rowset(MetaServiceProxy* meta_service, int64_t db_id, const std::string& label,
int64_t table_id, int64_t tablet_id, int32_t schema_version,
const TabletSchemaCloudPB* schema = nullptr) {
int64_t txn_id = 0;
ASSERT_NO_FATAL_FAILURE(begin_txn(meta_service, db_id, label, table_id, txn_id));
CreateRowsetResponse res;
auto rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), schema_version, -1, schema);
rowset.set_has_variant_type_in_schema(schema != nullptr);
prepare_rowset(meta_service, rowset, res);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label;
res.Clear();
ASSERT_NO_FATAL_FAILURE(commit_rowset(meta_service, rowset, res));
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << label << ", msg=" << res.status().msg();
commit_txn(meta_service, db_id, txn_id, label);
}
static TabletSchemaCloudPB getVariantSchema() {
TabletSchemaCloudPB schema;
schema.set_schema_version(10086);
// columns
ColumnPB var;
var.set_type("VARIANT");
var.set_unique_id(10);
ColumnPB var_sub1;
var_sub1.set_type("INT");
var_sub1.set_unique_id(-1);
schema.add_column()->CopyFrom(var_sub1);
ColumnPB var_sub2;
var_sub2.set_type("DOUBLE");
var_sub2.set_unique_id(-1);
schema.add_column()->CopyFrom(var_sub2);
ColumnPB var_sparse_sub1;
var_sparse_sub1.set_type("DOUBLE");
var_sparse_sub1.set_unique_id(-1);
var.add_sparse_columns()->CopyFrom(var_sparse_sub1);
schema.add_column()->CopyFrom(var);
// indexes
doris::TabletIndexPB index1;
index1.set_index_id(111);
index1.set_index_suffix_name("aaabbbccc");
schema.add_index()->CopyFrom(index1);
doris::TabletIndexPB index2;
index2.set_index_id(222);
schema.add_index()->CopyFrom(index2);
return schema;
}
TEST(DetachSchemaKVTest, RowsetTest) {
auto meta_service = get_meta_service();
// meta_service->resource_mgr().reset(); // Do not use resource manager
auto sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
SyncPoint::get_instance()->clear_all_call_backs();
};
sp->set_call_back("get_instance_id", [&](auto&& args) {
auto* ret = try_any_cast_ret<std::string>(args);
ret->first = instance_id;
ret->second = true;
});
sp->enable_processing();
constexpr int64_t db_id = 10000;
// new MS write with write_schema_kv=false, old MS read
{
constexpr auto table_id = 10001, index_id = 10002, partition_id = 10003, tablet_id = 10004;
config::write_schema_kv = false;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
ASSERT_NO_FATAL_FAILURE(
insert_rowset(meta_service.get(), db_id, "101", table_id, tablet_id, 2)); // [2-2]
// check saved values in txn_kv
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_key, rowset_val;
meta_rowset_key({instance_id, tablet_id, 2}, &rowset_key); // [2-2]
ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB saved_rowset;
ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
ASSERT_TRUE(saved_rowset.has_tablet_schema());
EXPECT_EQ(saved_rowset.tablet_schema().schema_version(), 2);
}
// old MS write, new MS read
{
constexpr auto table_id = 10011, index_id = 10012, partition_id = 10013, tablet_id = 10014;
config::write_schema_kv = false;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
auto saved_rowset = create_rowset(10015, tablet_id, next_rowset_id(), 2, 2);
std::string rowset_key, rowset_val;
meta_rowset_key({instance_id, tablet_id, 2}, &rowset_key); // version=[2-2]
ASSERT_TRUE(saved_rowset.SerializeToString(&rowset_val));
txn->put(rowset_key, rowset_val);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
// check get rowset response
GetRowsetResponse get_rowset_res;
get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id, get_rowset_res);
ASSERT_EQ(get_rowset_res.rowset_meta_size(), 2);
ASSERT_EQ(get_rowset_res.rowset_meta(0).end_version(), 1); // [0-1]
ASSERT_TRUE(get_rowset_res.rowset_meta(0).has_tablet_schema());
EXPECT_EQ(get_rowset_res.rowset_meta(0).tablet_schema().schema_version(), 1);
EXPECT_EQ(get_rowset_res.rowset_meta(0).index_id(), index_id);
EXPECT_EQ(get_rowset_res.rowset_meta(0).schema_version(), 1);
ASSERT_EQ(get_rowset_res.rowset_meta(1).end_version(), 2); // [2-2]
ASSERT_TRUE(get_rowset_res.rowset_meta(1).has_tablet_schema());
EXPECT_EQ(get_rowset_res.rowset_meta(1).tablet_schema().schema_version(), 2);
EXPECT_EQ(get_rowset_res.rowset_meta(1).index_id(), index_id);
EXPECT_EQ(get_rowset_res.rowset_meta(1).schema_version(), 2);
}
// new MS write with write_schema_kv=true, new MS read
{
constexpr auto table_id = 10021, index_id = 10022, partition_id = 10023, tablet_id = 10024;
config::write_schema_kv = true;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
ASSERT_NO_FATAL_FAILURE(
insert_rowset(meta_service.get(), db_id, "201", table_id, tablet_id, 2)); // [2-2]
// check saved values in txn_kv
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_key, rowset_val;
meta_rowset_key({instance_id, 10024, 2}, &rowset_key); // [2-2]
ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB saved_rowset;
ASSERT_TRUE(saved_rowset.ParseFromString(rowset_val));
EXPECT_FALSE(saved_rowset.has_tablet_schema());
EXPECT_EQ(saved_rowset.index_id(), index_id);
EXPECT_EQ(saved_rowset.schema_version(), 2);
// check get rowset response
GetRowsetResponse get_rowset_res;
get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id, get_rowset_res);
ASSERT_EQ(get_rowset_res.rowset_meta_size(), 2);
ASSERT_EQ(get_rowset_res.rowset_meta(0).end_version(), 1); // [0-1]
ASSERT_TRUE(get_rowset_res.rowset_meta(0).has_tablet_schema());
EXPECT_EQ(get_rowset_res.rowset_meta(0).tablet_schema().schema_version(), 1);
EXPECT_EQ(get_rowset_res.rowset_meta(0).index_id(), index_id);
EXPECT_EQ(get_rowset_res.rowset_meta(0).schema_version(), 1);
ASSERT_EQ(get_rowset_res.rowset_meta(1).end_version(), 2); // [2-2]
ASSERT_TRUE(get_rowset_res.rowset_meta(1).has_tablet_schema());
EXPECT_EQ(get_rowset_res.rowset_meta(1).tablet_schema().schema_version(), 2);
EXPECT_EQ(get_rowset_res.rowset_meta(1).index_id(), index_id);
EXPECT_EQ(get_rowset_res.rowset_meta(1).schema_version(), 2);
ASSERT_TRUE(get_rowset_res.has_stats());
EXPECT_EQ(get_rowset_res.stats().num_rows(), 100);
EXPECT_EQ(get_rowset_res.stats().num_rowsets(), 2);
EXPECT_EQ(get_rowset_res.stats().num_segments(), 1);
EXPECT_EQ(get_rowset_res.stats().data_size(), 11000);
EXPECT_EQ(get_rowset_res.stats().index_size(), 1000);
EXPECT_EQ(get_rowset_res.stats().segment_size(), 10000);
}
// new MS read rowsets committed by both old and new MS
auto insert_and_get_rowset = [&meta_service](int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id,
int label_base,
google::protobuf::Arena* arena = nullptr,
const TabletSchemaCloudPB* schema = nullptr) {
config::write_schema_kv = false;
std::mt19937 rng(std::chrono::system_clock::now().time_since_epoch().count());
std::uniform_int_distribution<int> dist1(1, 4);
std::uniform_int_distribution<int> dist2(2, 7);
std::vector<int> schema_versions {1};
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(),
schema_versions[0]));
for (int i = 0; i < 10; ++i) {
schema_versions.push_back(dist1(rng));
ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), db_id,
std::to_string(++label_base), table_id, tablet_id,
schema_versions.back(), schema));
}
config::write_schema_kv = true;
for (int i = 0; i < 15; ++i) {
schema_versions.push_back(dist2(rng));
ASSERT_NO_FATAL_FAILURE(insert_rowset(meta_service.get(), db_id,
std::to_string(++label_base), table_id, tablet_id,
schema_versions.back(), schema));
}
// check get rowset response
auto get_rowset_res = google::protobuf::Arena::CreateMessage<GetRowsetResponse>(arena);
DORIS_CLOUD_DEFER {
if (!arena) delete get_rowset_res;
};
get_rowset(meta_service.get(), table_id, index_id, partition_id, tablet_id,
*get_rowset_res);
ASSERT_EQ(get_rowset_res->rowset_meta_size(), schema_versions.size());
for (int i = 0; i < schema_versions.size(); ++i) {
auto& rowset = get_rowset_res->rowset_meta(i);
ASSERT_EQ(rowset.end_version(), i + 1);
ASSERT_TRUE(rowset.has_tablet_schema());
EXPECT_EQ(rowset.tablet_schema().schema_version(), schema_versions[i]);
EXPECT_EQ(rowset.index_id(), index_id);
EXPECT_EQ(rowset.schema_version(), schema_versions[i]);
}
ASSERT_TRUE(get_rowset_res->has_stats());
EXPECT_EQ(get_rowset_res->stats().num_rows(), 2500);
EXPECT_EQ(get_rowset_res->stats().num_rowsets(), 26);
EXPECT_EQ(get_rowset_res->stats().num_segments(), 25);
EXPECT_EQ(get_rowset_res->stats().data_size(), 275000);
EXPECT_EQ(get_rowset_res->stats().index_size(), 25000);
EXPECT_EQ(get_rowset_res->stats().segment_size(), 250000);
if (schema != nullptr) {
auto schema_version = get_rowset_res->rowset_meta(10).schema_version();
get_rowset_res->mutable_rowset_meta(10)->mutable_tablet_schema()->set_schema_version(
10086);
std::cout << get_rowset_res->rowset_meta(10).tablet_schema().ShortDebugString()
<< std::endl;
std::cout << schema->ShortDebugString() << std::endl;
EXPECT_EQ(get_rowset_res->rowset_meta(10).tablet_schema().column(2).type(),
schema->column(2).type());
EXPECT_EQ(get_rowset_res->rowset_meta(10).tablet_schema().index(0).index_suffix_name(),
schema->index(0).index_suffix_name());
EXPECT_EQ(get_rowset_res->rowset_meta(10).tablet_schema().index(1).index_id(),
schema->index(1).index_id());
get_rowset_res->mutable_rowset_meta(10)->mutable_tablet_schema()->set_schema_version(
schema_version);
}
};
insert_and_get_rowset(10031, 10032, 10033, 10034, 300);
// use arena
google::protobuf::Arena arena;
insert_and_get_rowset(10041, 10042, 10043, 10044, 400, &arena);
TabletSchemaCloudPB schema = getVariantSchema();
insert_and_get_rowset(10051, 10052, 10053, 10054, 500, &arena, &schema);
}
TEST(DetachSchemaKVTest, InsertExistedRowsetTest) {
auto meta_service = get_meta_service();
// meta_service->resource_mgr().reset(); // Do not use resource manager
auto sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
SyncPoint::get_instance()->clear_all_call_backs();
config::enable_recycle_delete_rowset_key_check = true;
};
sp->set_call_back("get_instance_id", [&](auto&& args) {
auto* ret = try_any_cast_ret<std::string>(args);
ret->first = instance_id;
ret->second = true;
});
sp->enable_processing();
config::enable_recycle_delete_rowset_key_check = false;
int64_t db_id = 1000;
// old MS commit rowset, new MS commit rowset again
{
constexpr auto table_id = 10001, index_id = 10002, partition_id = 10003, tablet_id = 10004;
config::write_schema_kv = false;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
int64_t txn_id = -1;
std::string label = "test_abort_txn_label";
{
brpc::Controller cntl;
BeginTxnRequest req;
BeginTxnResponse res;
req.set_cloud_unique_id("test_cloud_unique_id");
auto* txn_info = req.mutable_txn_info();
txn_info->set_db_id(db_id);
txn_info->set_label(label);
txn_info->add_table_ids(table_id);
txn_info->set_timeout_ms(36000);
meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
&req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
txn_id = res.txn_id();
LOG(INFO) << "Step 1: Transaction started, txn_id=" << txn_id;
}
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
auto committed_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
std::string tmp_rowset_key, tmp_rowset_val;
// 0:instance_id 1:txn_id 2:tablet_id
meta_rowset_tmp_key({instance_id, txn_id, tablet_id}, &tmp_rowset_key);
ASSERT_TRUE(committed_rowset.SerializeToString(&tmp_rowset_val));
txn->put(tmp_rowset_key, tmp_rowset_val);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
CreateRowsetResponse res;
auto new_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
prepare_rowset(meta_service.get(), new_rowset, res);
ASSERT_EQ(res.status().code(), MetaServiceCode::ALREADY_EXISTED);
ASSERT_TRUE(res.has_existed_rowset_meta());
EXPECT_EQ(res.existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
EXPECT_EQ(res.existed_rowset_meta().index_id(), index_id);
EXPECT_EQ(res.existed_rowset_meta().schema_version(), 2);
ASSERT_TRUE(res.existed_rowset_meta().has_tablet_schema());
EXPECT_EQ(res.existed_rowset_meta().tablet_schema().schema_version(), 2);
res.Clear();
commit_rowset(meta_service.get(), new_rowset, res);
ASSERT_EQ(res.status().code(), MetaServiceCode::ALREADY_EXISTED);
ASSERT_TRUE(res.has_existed_rowset_meta());
EXPECT_EQ(res.existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
EXPECT_EQ(res.existed_rowset_meta().index_id(), index_id);
EXPECT_EQ(res.existed_rowset_meta().schema_version(), 2);
ASSERT_TRUE(res.existed_rowset_meta().has_tablet_schema());
EXPECT_EQ(res.existed_rowset_meta().tablet_schema().schema_version(), 2);
}
// new MS commit rowset, new MS commit rowset again
auto insert_existed_rowset = [&meta_service, db_id](int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id,
int64_t txn_id,
google::protobuf::Arena* arena = nullptr) {
config::write_schema_kv = true;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(), 1));
auto committed_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
auto res = google::protobuf::Arena::CreateMessage<CreateRowsetResponse>(arena);
DORIS_CLOUD_DEFER {
if (!arena) delete res;
};
prepare_rowset(meta_service.get(), committed_rowset, *res);
ASSERT_EQ(res->status().code(), MetaServiceCode::OK);
res->Clear();
commit_rowset(meta_service.get(), committed_rowset, *res);
ASSERT_EQ(res->status().code(), MetaServiceCode::OK);
res->Clear();
auto new_rowset = create_rowset(txn_id, tablet_id, next_rowset_id(), 2, 2);
prepare_rowset(meta_service.get(), new_rowset, *res);
ASSERT_EQ(res->status().code(), MetaServiceCode::ALREADY_EXISTED);
ASSERT_TRUE(res->has_existed_rowset_meta());
EXPECT_EQ(res->existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
EXPECT_EQ(res->existed_rowset_meta().index_id(), index_id);
EXPECT_EQ(res->existed_rowset_meta().schema_version(), 2);
ASSERT_TRUE(res->existed_rowset_meta().has_tablet_schema());
EXPECT_EQ(res->existed_rowset_meta().tablet_schema().schema_version(), 2);
res->Clear();
commit_rowset(meta_service.get(), new_rowset, *res);
ASSERT_EQ(res->status().code(), MetaServiceCode::ALREADY_EXISTED);
ASSERT_TRUE(res->has_existed_rowset_meta());
EXPECT_EQ(res->existed_rowset_meta().rowset_id_v2(), committed_rowset.rowset_id_v2());
EXPECT_EQ(res->existed_rowset_meta().index_id(), index_id);
EXPECT_EQ(res->existed_rowset_meta().schema_version(), 2);
ASSERT_TRUE(res->existed_rowset_meta().has_tablet_schema());
EXPECT_EQ(res->existed_rowset_meta().tablet_schema().schema_version(), 2);
};
insert_existed_rowset(10011, 10012, 10013, 10014, 10015);
google::protobuf::Arena arena;
insert_existed_rowset(10021, 10022, 10023, 10024, 10025, &arena);
}
TEST(SchemaKVTest, InsertExistedRowsetTest) {
auto meta_service = get_meta_service();
// meta_service->resource_mgr().reset(); // Do not use resource manager
auto sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
SyncPoint::get_instance()->clear_all_call_backs();
};
sp->set_call_back("get_instance_id", [&](auto&& args) {
auto* ret = try_any_cast_ret<std::string>(args);
ret->first = instance_id;
ret->second = true;
});
sp->enable_processing();
constexpr int64_t db_id = 2000;
config::write_schema_kv = true;
config::meta_schema_value_version = 0;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
next_rowset_id(), 1));
check_get_tablet(meta_service.get(), 10004, 1);
config::meta_schema_value_version = 1;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
next_rowset_id(), 2));
check_get_tablet(meta_service.get(), 10005, 2);
}
static void check_schema(MetaServiceProxy* meta_service, int64_t tablet_id, int32_t schema_version,
bool expected_disable_auto_compaction = false) {
brpc::Controller cntl;
GetTabletRequest req;
GetTabletResponse res;
req.set_tablet_id(tablet_id);
meta_service->get_tablet(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
ASSERT_TRUE(res.has_tablet_meta()) << tablet_id;
EXPECT_TRUE(res.tablet_meta().has_schema()) << tablet_id;
EXPECT_EQ(res.tablet_meta().schema_version(), schema_version) << tablet_id;
EXPECT_EQ(res.tablet_meta().schema().column_size(), 10) << tablet_id;
EXPECT_EQ(res.tablet_meta().schema().disable_auto_compaction(),
expected_disable_auto_compaction)
<< tablet_id;
};
static void update_tablet(MetaServiceProxy* meta_service, int64_t tablet_id) {
brpc::Controller cntl;
UpdateTabletRequest req;
UpdateTabletResponse res;
auto meta_info = req.add_tablet_meta_infos();
meta_info->set_disable_auto_compaction(true);
meta_info->set_tablet_id(tablet_id);
meta_service->update_tablet(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}
TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
//case 1 config::write_schema_kv = true;
constexpr int64_t db_id = 3000;
{
auto meta_service = get_meta_service();
config::write_schema_kv = true;
//config::meta_schema_value_version = 0;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
next_rowset_id(), 0));
check_get_tablet(meta_service.get(), 10004, 0);
check_schema(meta_service.get(), 10004, 0);
//config::meta_schema_value_version = 1;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
next_rowset_id(), 2));
check_get_tablet(meta_service.get(), 10005, 2);
update_tablet(meta_service.get(), 10005);
check_schema(meta_service.get(), 10005, 2, true);
}
//case 2 config::write_schema_kv = false;
{
auto meta_service = get_meta_service();
config::write_schema_kv = false;
auto defer1 =
std::make_unique<std::function<void()>>([]() { config::write_schema_kv = true; });
//config::meta_schema_value_version = 0;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
next_rowset_id(), 0));
check_get_tablet(meta_service.get(), 10004, 0);
check_schema(meta_service.get(), 10004, 0);
//config::meta_schema_value_version = 1;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
next_rowset_id(), 2));
check_get_tablet(meta_service.get(), 10005, 2);
update_tablet(meta_service.get(), 10005);
check_schema(meta_service.get(), 10005, 2, true);
}
//case 3 config::write_schema_kv = false, create tablet, config::write_schema_kv = true;
{
auto meta_service = get_meta_service();
config::write_schema_kv = false;
auto defer1 =
std::make_unique<std::function<void()>>([]() { config::write_schema_kv = true; });
//config::meta_schema_value_version = 0;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
next_rowset_id(), 0));
check_get_tablet(meta_service.get(), 10004, 0);
check_schema(meta_service.get(), 10004, 0);
//config::meta_schema_value_version = 1;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
next_rowset_id(), 2));
check_get_tablet(meta_service.get(), 10005, 2);
config::write_schema_kv = true;
update_tablet(meta_service.get(), 10005);
check_schema(meta_service.get(), 10005, 2, true);
}
//case 4 config::write_schema_kv = false, create tablet, config::write_schema_kv = true;
// meta_schema_value_version = 0, meta_schema_value_version = 1
{
auto meta_service = get_meta_service();
config::write_schema_kv = false;
auto defer1 = std::make_unique<std::function<void()>>([]() {
config::write_schema_kv = true;
config::meta_schema_value_version = 1;
});
config::meta_schema_value_version = 0;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10004,
next_rowset_id(), 0));
check_get_tablet(meta_service.get(), 10004, 0);
check_schema(meta_service.get(), 10004, 0);
config::meta_schema_value_version = 1;
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, 10001, 10002, 10003, 10005,
next_rowset_id(), 2));
check_get_tablet(meta_service.get(), 10005, 2);
config::write_schema_kv = true;
update_tablet(meta_service.get(), 10005);
check_schema(meta_service.get(), 10005, 2, true);
}
}
// Helper: count the number of blob chunks for a given base key.
static int count_blob_chunks(Transaction* txn, const std::string& base_key) {
std::string begin_key = base_key;
std::string end_key = base_key;
encode_int64(INT64_MAX, &end_key);
std::unique_ptr<RangeGetIterator> iter;
auto err = txn->get(begin_key, end_key, &iter);
EXPECT_EQ(err, TxnErrorCode::TXN_OK);
if (err != TxnErrorCode::TXN_OK) return -1;
int count = 0;
while (iter->has_next()) {
iter->next();
++count;
}
return count;
}
// Verify that ValueBuf::remove() before blob_put is necessary to clean up
// orphaned blob chunks when the number of chunks decreases.
TEST(AlterSchemaKVTest, BlobRemoveBeforeOverwriteTest) {
auto meta_service = get_meta_service();
auto sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
sp->clear_all_call_backs();
sp->disable_processing();
};
sp->set_call_back("get_instance_id", [&](auto&& args) {
auto* ret = try_any_cast_ret<std::string>(args);
ret->first = instance_id;
ret->second = true;
});
sp->enable_processing();
constexpr int64_t db_id = 4000;
constexpr int64_t table_id = 40001;
constexpr int64_t index_id = 40002;
constexpr int64_t partition_id = 40003;
constexpr int64_t tablet_id = 40004;
constexpr int32_t schema_version = 0;
config::write_schema_kv = true;
config::meta_schema_value_version = 1;
// Create a tablet with a normal 10-column schema.
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id, index_id,
partition_id, tablet_id, next_rowset_id(),
schema_version));
check_get_tablet(meta_service.get(), tablet_id, schema_version);
// 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 3 blob chunks.
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
// First remove existing blob chunks
ValueBuf old_val;
ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val), TxnErrorCode::TXN_OK);
old_val.remove(txn.get());
doris::TabletSchemaCloudPB schema;
ASSERT_TRUE(old_val.to_pb(&schema));
std::string serialized = schema.SerializeAsString();
put_three_blob_chunks(txn.get(), serialized);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
// Verify we now have 3 blob chunks.
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 3);
}
// Case 1: update_tablet WITH remove (default behavior).
// The remove cleans up old 3 chunks, then blob_put writes 1 chunk
// (with default split_size=90KB, the ~120 byte schema fits in 1 chunk).
update_tablet(meta_service.get(), tablet_id);
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 1)
<< "With remove: should have 1 chunk (old 3 cleaned up)";
// Verify schema is correct
ValueBuf val;
ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &val), TxnErrorCode::TXN_OK);
doris::TabletSchemaCloudPB schema;
ASSERT_TRUE(val.to_pb(&schema));
EXPECT_TRUE(schema.disable_auto_compaction());
}
check_schema(meta_service.get(), tablet_id, schema_version, true);
// Reset: overwrite schema KV again with 3 blob chunks and
// disable_auto_compaction=false so update_tablet will modify it again.
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
ValueBuf old_val;
ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val), TxnErrorCode::TXN_OK);
old_val.remove(txn.get());
doris::TabletSchemaCloudPB schema;
ASSERT_TRUE(old_val.to_pb(&schema));
schema.set_disable_auto_compaction(false);
std::string serialized = schema.SerializeAsString();
put_three_blob_chunks(txn.get(), serialized);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 3);
}
// Case 2: update_tablet WITHOUT remove (inject skip via sync point).
// This simulates what would happen if we didn't have the remove logic.
sp->set_call_back("update_tablet::skip_schema_remove",
[](auto&& args) { *try_any_cast<bool*>(args[0]) = true; });
update_tablet(meta_service.get(), tablet_id);
sp->clear_call_back("update_tablet::skip_schema_remove");
{
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
int chunks = count_blob_chunks(txn.get(), schema_key);
EXPECT_EQ(chunks, 3) << "Without remove: should have 3 chunks (2 orphaned + 1 overwritten)";
// blob_get reads all 3 chunks and concatenates them, producing
// corrupted data that cannot be parsed correctly.
ValueBuf val;
ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &val), TxnErrorCode::TXN_OK);
doris::TabletSchemaCloudPB schema;
// The deserialized schema may parse but contain garbage from
// orphaned chunks, or the column count will be wrong.
if (val.to_pb(&schema)) {
// If it parses, the data is corrupted - the column count or
// content won't match the expected schema.
LOG(INFO) << "Without remove: parsed schema has " << schema.column_size()
<< " columns (expected 10), disable_auto_compaction="
<< schema.disable_auto_compaction();
} else {
LOG(INFO) << "Without remove: schema parse failed as expected (corrupted data)";
}
}
}
} // namespace doris::cloud
Reference in new issue View Git Blame Copy Permalink