Google Git Google Git
Sign in
apache/doris/refs/heads/stack-demo/./cloud/test/meta_service_table_stream_test.cpp
blob: d087340db64dd8d66b712b4f5f47ce3e1f5649f5 [file]
// 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 <gtest/gtest.h>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <initializer_list>
#include <limits>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <thread>
#include <vector>
#include "common/config.h"
#include "common/defer.h"
#include "common/lexical_util.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-store/keys.h"
#include "meta-store/txn_kv.h"
#include "meta-store/txn_kv_error.h"
#include "meta-store/versioned_value.h"
#include "recycler/recycler.h"
#include "resource-manager/resource_manager.h"
#include "snapshot/snapshot_manager.h"
namespace doris::cloud {
extern std::unique_ptr<MetaServiceProxy> get_meta_service(bool mock_resource_mgr);
extern void add_tablet(CreateTabletsRequest& request, int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id);
extern doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id, int partition_id,
int64_t version, int num_rows);
extern TxnErrorCode read_operation_log(Transaction* txn, std::string_view log_key,
Versionstamp* log_version, OperationLogPB* operation_log);
class MetaServiceTableStreamTest : public ::testing::Test {
protected:
void SetUp() override {
service_ = get_meta_service(false);
instance_id_ = "table_stream_read_state_instance";
cloud_unique_id_ = "1:" + instance_id_ + ":test";
identity_.set_base_db_id(1001);
identity_.set_base_table_id(1002);
identity_.set_stream_db_id(1003);
identity_.set_stream_id(1004);
}
void set_multi_version_status(MultiVersionStatus mode) {
multi_version_status_ = mode;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
InstanceInfoPB instance;
instance.set_instance_id(instance_id_);
instance.set_multi_version_status(mode);
txn->put(instance_key({instance_id_}), instance.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
auto [code, message] = service_->resource_mgr()->refresh_instance(instance_id_);
ASSERT_EQ(code, MetaServiceCode::OK) << message;
}
void put_partition_mapping(Transaction* txn, int64_t partition_id) {
PartitionIndexPB partition;
partition.set_db_id(identity_.base_db_id());
partition.set_table_id(identity_.base_table_id());
txn->put(versioned::partition_index_key({instance_id_, partition_id}),
partition.SerializeAsString());
}
void put_latest_partition_state(Transaction* txn, int64_t partition_id, int64_t visible_version,
int64_t commit_tso, std::optional<int64_t> offset_tso) {
VersionPB version;
version.set_version(visible_version);
version.set_commit_tso(commit_tso);
txn->put(partition_version_key({instance_id_, identity_.base_db_id(),
identity_.base_table_id(), partition_id}),
version.SerializeAsString());
if (offset_tso.has_value()) {
TableStreamOffsetPB offset;
offset.set_partition_id(partition_id);
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(*offset_tso);
offset.set_last_consumption_time_ms(1234);
txn->put(table_stream_offset_key({instance_id_, identity_.base_db_id(),
identity_.base_table_id(), identity_.stream_db_id(),
identity_.stream_id(), partition_id}),
offset.SerializeAsString());
}
}
void put_versioned_partition_state(Transaction* txn, int64_t partition_id,
int64_t visible_version, int64_t commit_tso,
std::optional<int64_t> offset_tso) {
put_partition_mapping(txn, partition_id);
versioned_put(txn, versioned::meta_partition_key({instance_id_, partition_id}),
Versionstamp(41, 0), "");
VersionPB version;
version.set_version(visible_version);
version.set_commit_tso(commit_tso);
versioned_put(txn, versioned::partition_version_key({instance_id_, partition_id}),
Versionstamp(42, 0), version.SerializeAsString());
if (offset_tso.has_value()) {
TableStreamOffsetPB offset;
offset.set_partition_id(partition_id);
offset.set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
offset.set_offset_tso(*offset_tso);
versioned_put(txn,
versioned::table_stream_offset_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), partition_id}),
Versionstamp(43, 0), offset.SerializeAsString());
}
}
GetTableStreamOffsetResponse get_read_state(MetaServiceProxy* service,
const std::vector<int64_t>& partitions) {
GetTableStreamOffsetRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
auto* binding = request.add_bindings();
binding->mutable_identity()->CopyFrom(identity_);
for (int64_t partition_id : partitions) {
binding->add_partition_ids(partition_id);
}
GetTableStreamOffsetResponse response;
brpc::Controller controller;
service->get_table_stream_offset(&controller, &request, &response, nullptr);
return response;
}
GetTableStreamOffsetResponse get_read_state(const std::vector<int64_t>& partitions) {
return get_read_state(service_.get(), partitions);
}
GetTableStreamOffsetResponse get_read_state(std::initializer_list<int64_t> partitions) {
return get_read_state(std::vector<int64_t>(partitions));
}
int64_t begin_target_transaction(const std::string& label) {
return begin_transaction(cloud_unique_id_, label);
}
int64_t begin_transaction(const std::string& cloud_unique_id, const std::string& label) {
BeginTxnRequest request;
request.set_cloud_unique_id(cloud_unique_id);
TxnInfoPB* txn_info = request.mutable_txn_info();
txn_info->set_db_id(3001);
txn_info->set_label(label);
txn_info->add_table_ids(3002);
txn_info->set_timeout_ms(36000);
BeginTxnResponse response;
brpc::Controller controller;
service_->begin_txn(&controller, &request, &response, nullptr);
EXPECT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
return response.txn_id();
}
void register_instance(const std::string& instance_id, MultiVersionStatus mode) {
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
InstanceInfoPB instance;
instance.set_instance_id(instance_id);
instance.set_multi_version_status(mode);
txn->put(instance_key({instance_id}), instance.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
auto [code, message] = service_->resource_mgr()->refresh_instance(instance_id);
ASSERT_EQ(code, MetaServiceCode::OK) << message;
}
void mark_transaction_visible(const std::string& instance_id, int64_t txn_id,
Transaction* txn) {
std::string txn_info_value;
ASSERT_EQ(txn->get(txn_info_key({instance_id, 3001, txn_id}), &txn_info_value),
TxnErrorCode::TXN_OK);
TxnInfoPB txn_info;
ASSERT_TRUE(txn_info.ParseFromString(txn_info_value));
txn_info.set_status(TXN_STATUS_VISIBLE);
txn->put(txn_info_key({instance_id, 3001, txn_id}), txn_info.SerializeAsString());
}
CommitTxnRequest make_consume_request(int64_t txn_id, int64_t partition_id,
TableStreamOffsetStatePB expected_state,
int64_t expected_tso, int64_t next_tso) {
CommitTxnRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.set_db_id(3001);
request.set_txn_id(txn_id);
TableStreamUpdatePB* stream_update = request.add_table_stream_updates();
stream_update->mutable_identity()->CopyFrom(identity_);
TableStreamPartitionUpdatePB* partition_update = stream_update->add_partition_updates();
partition_update->set_partition_id(partition_id);
partition_update->set_expected_state(expected_state);
if (expected_state != TABLE_STREAM_OFFSET_UNKNOWN) {
partition_update->set_expected_offset_tso(expected_tso);
}
partition_update->set_next_offset_tso(next_tso);
return request;
}
CommitTxnResponse commit_transaction(const CommitTxnRequest& request) {
CommitTxnResponse response;
brpc::Controller controller;
service_->commit_txn(&controller, &request, &response, nullptr);
return response;
}
std::pair<CommitTxnResponse, CommitTxnResponse> commit_concurrently(
const CommitTxnRequest& first_request, const CommitTxnRequest& second_request) {
std::mutex mutex;
std::condition_variable condition;
int commits_ready = 0;
bool release_commits = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
sync_point->clear_all_call_backs();
sync_point->clear_trace();
sync_point->disable_processing();
};
sync_point->set_call_back("commit_txn_immediately::before_commit", [&](auto&&) {
std::unique_lock lock(mutex);
++commits_ready;
condition.notify_all();
condition.wait(lock, [&] { return release_commits; });
});
sync_point->enable_processing();
CommitTxnResponse first_response;
CommitTxnResponse second_response;
std::thread first_thread([&] { first_response = commit_transaction(first_request); });
std::thread second_thread([&] { second_response = commit_transaction(second_request); });
bool both_commits_ready = false;
{
std::unique_lock lock(mutex);
both_commits_ready = condition.wait_for(lock, std::chrono::seconds(10),
[&] { return commits_ready == 2; });
release_commits = true;
condition.notify_all();
}
first_thread.join();
second_thread.join();
EXPECT_TRUE(both_commits_ready);
EXPECT_EQ(commits_ready, 2);
return {std::move(first_response), std::move(second_response)};
}
CommitTxnResponse consume_partition(int64_t txn_id, int64_t partition_id,
TableStreamOffsetStatePB expected_state,
int64_t expected_tso, int64_t next_tso) {
return commit_transaction(
make_consume_request(txn_id, partition_id, expected_state, expected_tso, next_tso));
}
void create_target_tablet(int64_t db_id, int64_t table_id, int64_t index_id,
int64_t partition_id, int64_t tablet_id) {
CreateTabletsRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.set_db_id(db_id);
add_tablet(request, table_id, index_id, partition_id, tablet_id);
CreateTabletsResponse response;
brpc::Controller controller;
service_->create_tablets(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
VersionPB version;
version.set_version(1);
txn->put(partition_version_key({instance_id_, db_id, table_id, partition_id}),
version.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
void stage_target_rowset(int64_t txn_id, int64_t partition_id, int64_t tablet_id) {
const doris::RowsetMetaCloudPB rowset =
create_rowset(txn_id, tablet_id, partition_id, -1, 100);
auto call = [&](bool prepare) {
CreateRowsetRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.mutable_rowset_meta()->CopyFrom(rowset);
CreateRowsetResponse response;
brpc::Controller controller;
if (prepare) {
service_->prepare_rowset(&controller, &request, &response, nullptr);
} else {
service_->commit_rowset(&controller, &request, &response, nullptr);
}
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
};
ASSERT_NO_FATAL_FAILURE(call(true));
ASSERT_NO_FATAL_FAILURE(call(false));
}
TableStreamOffsetPB get_latest_offset(const TableStreamIdentityPB& identity,
int64_t partition_id) {
std::unique_ptr<Transaction> txn;
EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(table_stream_offset_key(
{instance_id_, identity.base_db_id(), identity.base_table_id(),
identity.stream_db_id(), identity.stream_id(), partition_id}),
&value),
TxnErrorCode::TXN_OK);
TableStreamOffsetPB offset;
EXPECT_TRUE(offset.ParseFromString(value));
return offset;
}
TableStreamOffsetPB get_latest_offset(int64_t partition_id) {
return get_latest_offset(identity_, partition_id);
}
size_t count_keys(std::string_view begin, std::string_view end) {
std::unique_ptr<Transaction> txn;
EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::unique_ptr<RangeGetIterator> iter;
EXPECT_EQ(txn->get(begin, end, &iter, true, 0), TxnErrorCode::TXN_OK);
return iter->size();
}
std::vector<std::pair<std::string, std::string>> read_keys(std::string_view begin,
std::string_view end) {
std::unique_ptr<Transaction> txn;
EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
FullRangeGetOptions options;
options.txn = txn.get();
auto iter = service_->txn_kv()->full_range_get(std::string(begin), std::string(end),
std::move(options));
std::vector<std::pair<std::string, std::string>> entries;
for (auto kv = iter->next(); kv.has_value(); kv = iter->next()) {
entries.emplace_back(kv->first, kv->second);
}
EXPECT_TRUE(iter->is_valid());
return entries;
}
std::unique_ptr<MetaServiceProxy> restart_meta_service() {
const std::shared_ptr<TxnKv> txn_kv = service_->txn_kv();
auto resource_manager = std::make_shared<ResourceManager>(txn_kv);
auto [code, message] = resource_manager->refresh_instance(instance_id_);
EXPECT_EQ(code, MetaServiceCode::OK) << message;
auto rate_limiter = std::make_shared<RateLimiter>();
auto snapshot_manager = std::make_shared<SnapshotManager>(txn_kv);
auto restarted_impl = std::make_unique<MetaServiceImpl>(txn_kv, resource_manager,
rate_limiter, snapshot_manager);
return std::make_unique<MetaServiceProxy>(std::move(restarted_impl));
}
void set_clone_source(const std::string& source_instance_id, Versionstamp snapshot_version) {
multi_version_status_ = MULTI_VERSION_READ_WRITE;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
InstanceInfoPB instance;
instance.set_instance_id(instance_id_);
instance.set_multi_version_status(MULTI_VERSION_READ_WRITE);
instance.set_source_instance_id(source_instance_id);
instance.set_source_snapshot_id(SnapshotManager::serialize_snapshot_id(snapshot_version));
txn->put(instance_key({instance_id_}), instance.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
auto [code, message] = service_->resource_mgr()->refresh_instance(instance_id_);
ASSERT_EQ(code, MetaServiceCode::OK) << message;
}
Versionstamp put_auto_versioned_offset(const std::string& target_instance_id,
const TableStreamIdentityPB& identity,
int64_t partition_id, int64_t offset_tso,
bool write_latest = false) {
std::unique_ptr<Transaction> txn;
EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
TableStreamOffsetPB offset;
offset.set_partition_id(partition_id);
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(offset_tso);
const TableStreamOffsetKeyInfo key_info {target_instance_id, identity.base_db_id(),
identity.base_table_id(), identity.stream_db_id(),
identity.stream_id(), partition_id};
const std::string value = offset.SerializeAsString();
if (write_latest) {
txn->put(table_stream_offset_key(key_info), value);
}
txn->enable_get_versionstamp();
versioned_put(
txn.get(),
versioned::table_stream_offset_key(versioned::TableStreamOffsetKeyInfo(key_info)),
value);
EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
Versionstamp version;
EXPECT_EQ(txn->get_versionstamp(&version), TxnErrorCode::TXN_OK);
return version;
}
std::unique_ptr<MetaServiceProxy> service_;
std::string instance_id_;
std::string cloud_unique_id_;
TableStreamIdentityPB identity_;
MultiVersionStatus multi_version_status_ = MULTI_VERSION_DISABLED;
};
TEST_F(MetaServiceTableStreamTest, ReadLatestAndUnknownOffsets) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetTableStreamOffsetResponse response = get_read_state({2001, 2002});
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), 2);
const auto& consumed = response.bindings(0).partition_states(0);
EXPECT_EQ(consumed.partition_id(), 2001);
EXPECT_EQ(consumed.offset_state(), TABLE_STREAM_OFFSET_CONSUMED);
EXPECT_EQ(consumed.offset_tso(), 100);
EXPECT_EQ(consumed.end_tso(), 130);
EXPECT_EQ(consumed.visible_version(), 8);
EXPECT_EQ(consumed.last_consumption_time_ms(), 1234);
const auto& unknown = response.bindings(0).partition_states(1);
EXPECT_EQ(unknown.partition_id(), 2002);
EXPECT_EQ(unknown.offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
EXPECT_FALSE(unknown.has_offset_tso());
EXPECT_EQ(unknown.end_tso(), 140);
EXPECT_EQ(unknown.visible_version(), 9);
}
TEST_F(MetaServiceTableStreamTest, ReadVersionAndOffsetFromOneSnapshot) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kPartitionId = 2001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kPartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
bool updated = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
sync_point->disable_processing();
sync_point->clear_all_call_backs();
};
sync_point->set_call_back("get_table_stream_offset::after_read_versions", [&](auto&&) {
std::unique_ptr<Transaction> update_txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&update_txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(update_txn.get(), kPartitionId, 9, 140, 120);
ASSERT_EQ(update_txn->commit(), TxnErrorCode::TXN_OK);
updated = true;
});
sync_point->enable_processing();
const GetTableStreamOffsetResponse response = get_read_state({kPartitionId});
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_TRUE(updated);
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
const TableStreamPartitionReadStatePB& state = response.bindings(0).partition_states(0);
EXPECT_EQ(state.visible_version(), 8);
EXPECT_EQ(state.end_tso(), 130);
EXPECT_EQ(state.offset_tso(), 100);
sync_point->disable_processing();
sync_point->clear_all_call_backs();
const GetTableStreamOffsetResponse next_response = get_read_state({kPartitionId});
ASSERT_EQ(next_response.status().code(), MetaServiceCode::OK) << next_response.status().msg();
const TableStreamPartitionReadStatePB& next_state =
next_response.bindings(0).partition_states(0);
EXPECT_EQ(next_state.visible_version(), 9);
EXPECT_EQ(next_state.end_tso(), 140);
EXPECT_EQ(next_state.offset_tso(), 120);
}
TEST_F(MetaServiceTableStreamTest, ReadStateSurvivesMetaServiceRestartAndDoesNotWriteKv) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
const std::string all_keys_begin(1, '\x00');
const std::string all_keys_end(1, '\xff');
const auto keys_before = read_keys(all_keys_begin, all_keys_end);
const GetTableStreamOffsetResponse first_response = get_read_state({2001, 2002});
ASSERT_EQ(first_response.status().code(), MetaServiceCode::OK) << first_response.status().msg();
EXPECT_EQ(read_keys(all_keys_begin, all_keys_end), keys_before);
const std::shared_ptr<TxnKv> txn_kv = service_->txn_kv();
auto resource_manager = std::make_shared<ResourceManager>(txn_kv);
auto [code, message] = resource_manager->refresh_instance(instance_id_);
ASSERT_EQ(code, MetaServiceCode::OK) << message;
auto rate_limiter = std::make_shared<RateLimiter>();
auto snapshot_manager = std::make_shared<SnapshotManager>(txn_kv);
auto restarted_impl = std::make_unique<MetaServiceImpl>(txn_kv, resource_manager, rate_limiter,
snapshot_manager);
auto restarted_service = std::make_unique<MetaServiceProxy>(std::move(restarted_impl));
const GetTableStreamOffsetResponse restarted_response =
get_read_state(restarted_service.get(), {2001, 2002});
ASSERT_EQ(restarted_response.status().code(), MetaServiceCode::OK)
<< restarted_response.status().msg();
EXPECT_EQ(restarted_response.SerializeAsString(), first_response.SerializeAsString());
EXPECT_EQ(read_keys(all_keys_begin, all_keys_end), keys_before);
}
TEST_F(MetaServiceTableStreamTest, CreateResumesAcrossMetaServiceRestarts) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kFirstPartitionId = 2001;
constexpr int64_t kSecondPartitionId = 2002;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kFirstPartitionId, 8, 130, std::nullopt);
put_latest_partition_state(txn.get(), kSecondPartitionId, 9, 140, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
IndexRequest index_request;
index_request.set_cloud_unique_id(cloud_unique_id_);
index_request.set_db_id(identity_.base_db_id());
index_request.set_table_id(identity_.base_table_id());
index_request.add_index_ids(identity_.stream_id());
index_request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
index_request.set_stream_db_id(identity_.stream_db_id());
index_request.set_expiration(::time(nullptr) + 3600);
IndexResponse index_response;
brpc::Controller prepare_controller;
service_->prepare_index(&prepare_controller, &index_request, &index_response, nullptr);
ASSERT_EQ(index_response.status().code(), MetaServiceCode::OK) << index_response.status().msg();
auto make_partition_request = [&](int64_t partition_id, int64_t offset_tso) {
PartitionRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.add_index_ids(identity_.stream_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity_.stream_db_id());
request.add_partition_ids(partition_id);
TableStreamOffsetPB* offset = request.add_table_stream_offsets();
offset->set_partition_id(partition_id);
offset->set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
offset->set_offset_tso(offset_tso);
return request;
};
const PartitionRequest first_partition_request = make_partition_request(kFirstPartitionId, 100);
const PartitionRequest second_partition_request =
make_partition_request(kSecondPartitionId, 110);
auto restarted_after_prepare = restart_meta_service();
PartitionResponse partition_response;
brpc::Controller first_partition_controller;
restarted_after_prepare->commit_partition(&first_partition_controller, &first_partition_request,
&partition_response, nullptr);
ASSERT_EQ(partition_response.status().code(), MetaServiceCode::OK)
<< partition_response.status().msg();
auto restarted_after_partial_commit = restart_meta_service();
partition_response.Clear();
brpc::Controller retry_partition_controller;
restarted_after_partial_commit->commit_partition(
&retry_partition_controller, &first_partition_request, &partition_response, nullptr);
ASSERT_EQ(partition_response.status().code(), MetaServiceCode::OK)
<< partition_response.status().msg();
partition_response.Clear();
brpc::Controller second_partition_controller;
restarted_after_partial_commit->commit_partition(
&second_partition_controller, &second_partition_request, &partition_response, nullptr);
ASSERT_EQ(partition_response.status().code(), MetaServiceCode::OK)
<< partition_response.status().msg();
index_response.Clear();
brpc::Controller commit_index_controller;
restarted_after_partial_commit->commit_index(&commit_index_controller, &index_request,
&index_response, nullptr);
ASSERT_EQ(index_response.status().code(), MetaServiceCode::OK) << index_response.status().msg();
EXPECT_EQ(get_latest_offset(kFirstPartitionId).offset_tso(), 100);
EXPECT_EQ(get_latest_offset(kSecondPartitionId).offset_tso(), 110);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value, true),
TxnErrorCode::TXN_KEY_NOT_FOUND);
Versionstamp offset_version;
EXPECT_EQ(versioned_get(
txn.get(),
versioned::table_stream_offset_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), kFirstPartitionId}),
&offset_version, &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
TEST_F(MetaServiceTableStreamTest, ReadVersionedState) {
set_multi_version_status(MULTI_VERSION_READ_WRITE);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_versioned_partition_state(txn.get(), 2001, 18, 230, 200);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetTableStreamOffsetResponse response = get_read_state({2001});
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
const auto& state = response.bindings(0).partition_states(0);
EXPECT_EQ(state.offset_state(), TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
EXPECT_EQ(state.offset_tso(), 200);
EXPECT_EQ(state.end_tso(), 230);
EXPECT_EQ(state.visible_version(), 18);
}
TEST_F(MetaServiceTableStreamTest,
ReadVersionedStateFromCloneChainInBatchSurvivesRestartAndDoesNotWriteKv) {
const std::string source_instance_id = "table_stream_batch_read_source";
set_clone_source(source_instance_id, Versionstamp(100, 0));
const std::vector<int64_t> partition_ids = {2001, 2002, 2003};
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
for (size_t i = 0; i < partition_ids.size(); ++i) {
int64_t partition_id = partition_ids[i];
PartitionIndexPB partition;
partition.set_db_id(identity_.base_db_id());
partition.set_table_id(identity_.base_table_id());
txn->put(versioned::partition_index_key({source_instance_id, partition_id}),
partition.SerializeAsString());
versioned_put(txn.get(), versioned::meta_partition_key({source_instance_id, partition_id}),
Versionstamp(51 + i, 0), "");
VersionPB version;
version.set_version(10 + i);
version.set_commit_tso(1000 + i);
versioned_put(txn.get(),
versioned::partition_version_key({source_instance_id, partition_id}),
Versionstamp(60 + i, 0), version.SerializeAsString());
}
auto put_offset = [&](const std::string& target_instance_id, int64_t partition_id,
Versionstamp versionstamp, int64_t offset_tso) {
TableStreamOffsetPB offset;
offset.set_partition_id(partition_id);
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(offset_tso);
versioned_put(
txn.get(),
versioned::table_stream_offset_key(
{target_instance_id, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), partition_id}),
versionstamp, offset.SerializeAsString());
};
put_offset(source_instance_id, 2001, Versionstamp(70, 0), 701);
put_offset(source_instance_id, 2002, Versionstamp(71, 0), 702);
put_offset(instance_id_, 2002, Versionstamp(200, 0), 802);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
const std::string all_keys_begin(1, '\x00');
const std::string all_keys_end(1, '\xff');
const auto keys_before = read_keys(all_keys_begin, all_keys_end);
const GetTableStreamOffsetResponse response = get_read_state(partition_ids);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), partition_ids.size());
EXPECT_EQ(response.bindings(0).partition_states(0).offset_tso(), 701);
EXPECT_EQ(response.bindings(0).partition_states(1).offset_tso(), 802);
EXPECT_EQ(response.bindings(0).partition_states(2).offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
EXPECT_EQ(read_keys(all_keys_begin, all_keys_end), keys_before);
auto restarted_service = restart_meta_service();
const GetTableStreamOffsetResponse restarted_response =
get_read_state(restarted_service.get(), partition_ids);
ASSERT_EQ(restarted_response.status().code(), MetaServiceCode::OK)
<< restarted_response.status().msg();
EXPECT_EQ(restarted_response.SerializeAsString(), response.SerializeAsString());
EXPECT_EQ(read_keys(all_keys_begin, all_keys_end), keys_before);
}
TEST_F(MetaServiceTableStreamTest, ReadLargePartitionBatch) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int kPartitionCount = 1201;
std::vector<int64_t> partition_ids;
partition_ids.reserve(kPartitionCount);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
for (int i = 0; i < kPartitionCount; ++i) {
int64_t partition_id = 10000 + i;
partition_ids.push_back(partition_id);
put_latest_partition_state(txn.get(), partition_id, i + 1, 100000 + i, 90000 + i);
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetTableStreamOffsetResponse response = get_read_state(partition_ids);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.bindings(0).partition_states_size(), kPartitionCount);
EXPECT_EQ(response.bindings(0).partition_states(0).partition_id(), partition_ids.front());
EXPECT_EQ(response.bindings(0).partition_states(0).offset_tso(), 90000);
EXPECT_EQ(response.bindings(0).partition_states(kPartitionCount - 1).partition_id(),
partition_ids.back());
EXPECT_EQ(response.bindings(0).partition_states(kPartitionCount - 1).offset_tso(),
90000 + kPartitionCount - 1);
}
TEST_F(MetaServiceTableStreamTest, ReadMultipleBindingsInBatch) {
set_multi_version_status(MULTI_VERSION_READ_WRITE);
constexpr int kBindingCount = 101;
constexpr int64_t kPartitionId = 2001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
PartitionIndexPB partition;
partition.set_db_id(identity_.base_db_id());
partition.set_table_id(identity_.base_table_id());
txn->put(versioned::partition_index_key({instance_id_, kPartitionId}),
partition.SerializeAsString());
versioned_put(txn.get(), versioned::meta_partition_key({instance_id_, kPartitionId}),
Versionstamp(20, 0), "");
VersionPB version;
version.set_version(8);
version.set_commit_tso(130);
versioned_put(txn.get(), versioned::partition_version_key({instance_id_, kPartitionId}),
Versionstamp(21, 0), version.SerializeAsString());
GetTableStreamOffsetRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
for (int i = 0; i < kBindingCount; ++i) {
int64_t stream_id = identity_.stream_id() + i;
TableStreamPartitionSetPB* binding = request.add_bindings();
binding->mutable_identity()->CopyFrom(identity_);
binding->mutable_identity()->set_stream_id(stream_id);
binding->add_partition_ids(kPartitionId);
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetTableStreamOffsetResponse response;
brpc::Controller controller;
service_->get_table_stream_offset(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.bindings_size(), kBindingCount);
for (const TableStreamReadBindingResultPB& binding : response.bindings()) {
ASSERT_EQ(binding.partition_states_size(), 1);
EXPECT_EQ(binding.partition_states(0).offset_state(), TABLE_STREAM_OFFSET_UNKNOWN);
}
}
TEST_F(MetaServiceTableStreamTest, ReadWriteOnlyState) {
set_multi_version_status(MULTI_VERSION_WRITE_ONLY);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 28, 330, 300);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetTableStreamOffsetResponse response = get_read_state({2001});
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
const auto& state = response.bindings(0).partition_states(0);
EXPECT_EQ(state.offset_state(), TABLE_STREAM_OFFSET_CONSUMED);
EXPECT_EQ(state.offset_tso(), 300);
EXPECT_EQ(state.end_tso(), 330);
EXPECT_EQ(state.visible_version(), 28);
}
TEST_F(MetaServiceTableStreamTest, WaitsForPendingSourceTxnAndRereadsSnapshot) {
set_multi_version_status(MULTI_VERSION_DISABLED);
const int64_t pending_txn_id = begin_target_transaction("pending-source-version");
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string txn_info_value;
ASSERT_EQ(txn->get(txn_info_key({instance_id_, 3001, pending_txn_id}), &txn_info_value),
TxnErrorCode::TXN_OK);
TxnInfoPB txn_info;
ASSERT_TRUE(txn_info.ParseFromString(txn_info_value));
txn_info.set_status(TXN_STATUS_VISIBLE);
txn->put(txn_info_key({instance_id_, 3001, pending_txn_id}), txn_info.SerializeAsString());
VersionPB pending_version;
pending_version.set_commit_tso(130);
pending_version.add_pending_txn_ids(pending_txn_id);
txn->put(partition_version_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(), 2001}),
pending_version.SerializeAsString());
TableStreamOffsetPB offset;
offset.set_partition_id(2001);
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(100);
txn->put(table_stream_offset_key({instance_id_, identity_.base_db_id(),
identity_.base_table_id(), identity_.stream_db_id(),
identity_.stream_id(), 2001}),
offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
bool after_wait_called = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
sync_point->disable_processing();
sync_point->clear_all_call_backs();
};
sync_point->set_call_back("get_table_stream_offset::after_wait_for_pending_txns", [&](auto&&) {
std::unique_ptr<Transaction> update_txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&update_txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(update_txn.get(), 2001, 2, 130, 120);
ASSERT_EQ(update_txn->commit(), TxnErrorCode::TXN_OK);
after_wait_called = true;
});
sync_point->enable_processing();
GetTableStreamOffsetResponse response = get_read_state({2001});
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_TRUE(after_wait_called);
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
const TableStreamPartitionReadStatePB& state = response.bindings(0).partition_states(0);
EXPECT_EQ(state.visible_version(), 2);
EXPECT_EQ(state.end_tso(), 130);
EXPECT_EQ(state.offset_tso(), 120);
}
TEST_F(MetaServiceTableStreamTest, WaitsForPendingVersionFromCloneSourceInstance) {
const std::string source_instance_id = "table_stream_pending_version_source";
register_instance(source_instance_id, MULTI_VERSION_READ_WRITE);
set_clone_source(source_instance_id, Versionstamp(100, 0));
const int64_t pending_txn_id = begin_transaction(fmt::format("1:{}:test", source_instance_id),
"pending-clone-source-version");
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
mark_transaction_visible(source_instance_id, pending_txn_id, txn.get());
PartitionIndexPB partition;
partition.set_db_id(identity_.base_db_id());
partition.set_table_id(identity_.base_table_id());
txn->put(versioned::partition_index_key({source_instance_id, 2001}),
partition.SerializeAsString());
versioned_put(txn.get(), versioned::meta_partition_key({source_instance_id, 2001}),
Versionstamp(50, 0), "");
VersionPB pending_version;
pending_version.set_commit_tso(130);
pending_version.add_pending_txn_ids(pending_txn_id);
versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
Versionstamp(60, 0), pending_version.SerializeAsString());
TableStreamOffsetPB offset;
offset.set_partition_id(2001);
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(100);
versioned_put(txn.get(),
versioned::table_stream_offset_key(
{source_instance_id, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), 2001}),
Versionstamp(70, 0), offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
bool after_wait_called = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
sync_point->disable_processing();
sync_point->clear_all_call_backs();
};
sync_point->set_call_back("get_table_stream_offset::after_wait_for_pending_txns", [&](auto&&) {
std::unique_ptr<Transaction> update_txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&update_txn), TxnErrorCode::TXN_OK);
VersionPB committed_version;
committed_version.set_version(2);
committed_version.set_commit_tso(130);
versioned_put(update_txn.get(),
versioned::partition_version_key({source_instance_id, 2001}),
Versionstamp(60, 0), committed_version.SerializeAsString());
offset.set_offset_tso(120);
versioned_put(
update_txn.get(),
versioned::table_stream_offset_key(
{source_instance_id, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), 2001}),
Versionstamp(70, 0), offset.SerializeAsString());
ASSERT_EQ(update_txn->commit(), TxnErrorCode::TXN_OK);
after_wait_called = true;
});
sync_point->enable_processing();
GetTableStreamOffsetResponse response = get_read_state({2001});
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_TRUE(after_wait_called);
ASSERT_EQ(response.bindings_size(), 1);
ASSERT_EQ(response.bindings(0).partition_states_size(), 1);
const TableStreamPartitionReadStatePB& state = response.bindings(0).partition_states(0);
EXPECT_EQ(state.visible_version(), 2);
EXPECT_EQ(state.end_tso(), 130);
EXPECT_EQ(state.offset_tso(), 120);
}
TEST_F(MetaServiceTableStreamTest, SingleGetVersionWaitsUsingCloneSourceInstance) {
const std::string source_instance_id = "table_stream_single_pending_version_source";
register_instance(source_instance_id, MULTI_VERSION_READ_WRITE);
set_clone_source(source_instance_id, Versionstamp(100, 0));
const int64_t pending_txn_id = begin_transaction(fmt::format("1:{}:test", source_instance_id),
"pending-clone-single-version");
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
mark_transaction_visible(source_instance_id, pending_txn_id, txn.get());
VersionPB pending_version;
pending_version.set_commit_tso(130);
pending_version.add_pending_txn_ids(pending_txn_id);
versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
Versionstamp(60, 0), pending_version.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetVersionRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.set_partition_id(2001);
request.set_wait_for_pending_txn(true);
GetVersionResponse response;
brpc::Controller controller;
service_->get_version(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(response.version(), 2);
ASSERT_EQ(response.commit_tsos_size(), 1);
EXPECT_EQ(response.commit_tsos(0), 130);
}
TEST_F(MetaServiceTableStreamTest, BatchGetVersionWaitsUsingCloneSourceInstance) {
const std::string source_instance_id = "table_stream_batch_pending_version_source";
register_instance(source_instance_id, MULTI_VERSION_READ_WRITE);
set_clone_source(source_instance_id, Versionstamp(100, 0));
const int64_t pending_txn_id = begin_transaction(fmt::format("1:{}:test", source_instance_id),
"pending-clone-batch-version");
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
mark_transaction_visible(source_instance_id, pending_txn_id, txn.get());
PartitionIndexPB partition;
partition.set_db_id(identity_.base_db_id());
partition.set_table_id(identity_.base_table_id());
txn->put(versioned::partition_index_key({source_instance_id, 2001}),
partition.SerializeAsString());
VersionPB pending_version;
pending_version.set_commit_tso(130);
pending_version.add_pending_txn_ids(pending_txn_id);
versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
Versionstamp(60, 0), pending_version.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetVersionRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.set_batch_mode(true);
request.set_wait_for_pending_txn(true);
request.add_db_ids(identity_.base_db_id());
request.add_table_ids(identity_.base_table_id());
request.add_partition_ids(2001);
GetVersionResponse response;
brpc::Controller controller;
service_->get_version(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(response.versions_size(), 1);
ASSERT_EQ(response.commit_tsos_size(), 1);
EXPECT_EQ(response.versions(0), 2);
EXPECT_EQ(response.commit_tsos(0), 130);
}
TEST_F(MetaServiceTableStreamTest, RejectDuplicateBindingsAndPartitions) {
GetTableStreamOffsetRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
auto* binding = request.add_bindings();
binding->mutable_identity()->CopyFrom(identity_);
binding->add_partition_ids(2001);
binding->add_partition_ids(2001);
GetTableStreamOffsetResponse response;
brpc::Controller controller;
service_->get_table_stream_offset(&controller, &request, &response, nullptr);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
request.mutable_bindings()->Clear();
for (int i = 0; i < 2; ++i) {
binding = request.add_bindings();
binding->mutable_identity()->CopyFrom(identity_);
binding->add_partition_ids(2001 + i);
}
response.Clear();
brpc::Controller second_controller;
service_->get_table_stream_offset(&second_controller, &request, &response, nullptr);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
}
TEST_F(MetaServiceTableStreamTest, RejectEnabledModeAndRecyclingStream) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
InstanceInfoPB instance;
instance.set_instance_id(instance_id_);
instance.set_multi_version_status(MULTI_VERSION_ENABLED);
txn->put(instance_key({instance_id_}), instance.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
GetTableStreamOffsetResponse response = get_read_state({2001});
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
set_multi_version_status(MULTI_VERSION_DISABLED);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
RecycleIndexPB recycle_index;
recycle_index.set_table_id(identity_.base_table_id());
recycle_index.set_state(RecycleIndexPB::RECYCLING);
recycle_index.set_object_type(TABLE_STREAM);
txn->put(recycle_index_key({instance_id_, identity_.stream_id()}),
recycle_index.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
response = get_read_state({2001});
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
}
TEST_F(MetaServiceTableStreamTest, CommitUpdatesLatestOffsetAndIsIdempotent) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
int64_t txn_id = begin_target_transaction("consume-existing-offset");
CommitTxnResponse response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
TableStreamOffsetPB offset = get_latest_offset(2001);
EXPECT_EQ(offset.state(), TABLE_STREAM_OFFSET_CONSUMED);
EXPECT_EQ(offset.offset_tso(), 120);
EXPECT_GT(offset.last_consumption_time_ms(), 0);
response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 120);
txn_id = begin_target_transaction("consume-unknown-offset");
response = consume_partition(txn_id, 2002, TABLE_STREAM_OFFSET_UNKNOWN, 0, 140);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(2002).offset_tso(), 140);
}
TEST_F(MetaServiceTableStreamTest, CommitMultipleStreamsAtomically) {
set_multi_version_status(MULTI_VERSION_DISABLED);
TableStreamIdentityPB second_identity = identity_;
second_identity.set_stream_db_id(1005);
second_identity.set_stream_id(1006);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
TableStreamOffsetPB second_offset;
second_offset.set_partition_id(2001);
second_offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
second_offset.set_offset_tso(105);
second_offset.set_last_consumption_time_ms(1234);
txn->put(table_stream_offset_key(
{instance_id_, second_identity.base_db_id(), second_identity.base_table_id(),
second_identity.stream_db_id(), second_identity.stream_id(), 2001}),
second_offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
auto make_request = [&](int64_t txn_id, int64_t second_expected_tso) {
CommitTxnRequest request =
make_consume_request(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
TableStreamUpdatePB* second_stream_update = request.add_table_stream_updates();
second_stream_update->mutable_identity()->CopyFrom(second_identity);
TableStreamPartitionUpdatePB* second_partition_update =
second_stream_update->add_partition_updates();
second_partition_update->set_partition_id(2001);
second_partition_update->set_expected_state(TABLE_STREAM_OFFSET_CONSUMED);
second_partition_update->set_expected_offset_tso(second_expected_tso);
second_partition_update->set_next_offset_tso(125);
return request;
};
CommitTxnResponse response = commit_transaction(
make_request(begin_target_transaction("consume-multiple-streams-stale"), 104));
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(identity_, 2001).offset_tso(), 100);
EXPECT_EQ(get_latest_offset(second_identity, 2001).offset_tso(), 105);
response = commit_transaction(
make_request(begin_target_transaction("consume-multiple-streams"), 105));
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(identity_, 2001).offset_tso(), 120);
EXPECT_EQ(get_latest_offset(second_identity, 2001).offset_tso(), 125);
}
TEST_F(MetaServiceTableStreamTest, CommitTargetRowsetAndOffsetAtomically) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSourcePartitionId = 2001;
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
int64_t stale_txn_id = begin_target_transaction("consume-target-rowset-stale");
stage_target_rowset(stale_txn_id, kTargetPartitionId, kTargetTabletId);
CommitTxnRequest stale_request = make_consume_request(stale_txn_id, kSourcePartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 99, 120);
CommitTxnResponse response = commit_transaction(stale_request);
ASSERT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 100);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, stale_txn_id, kTargetTabletId}), &value),
TxnErrorCode::TXN_OK);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB target_version;
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 1);
int64_t committed_txn_id = begin_target_transaction("consume-target-rowset");
stage_target_rowset(committed_txn_id, kTargetPartitionId, kTargetTabletId);
CommitTxnRequest commit_request = make_consume_request(committed_txn_id, kSourcePartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
commit_request.set_commit_tso(999);
response = commit_transaction(commit_request);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, committed_txn_id, kTargetTabletId}),
&value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB committed_rowset;
ASSERT_TRUE(committed_rowset.ParseFromString(value));
EXPECT_EQ(committed_rowset.txn_id(), committed_txn_id);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 2);
EXPECT_EQ(target_version.commit_tso(), 999);
response = commit_transaction(commit_request);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
ASSERT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(committed_rowset.ParseFromString(value));
EXPECT_EQ(committed_rowset.txn_id(), committed_txn_id);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 2);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 3}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
TEST_F(MetaServiceTableStreamTest, SourcePublishDoesNotConflictWithConsumptionCommit) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSourcePartitionId = 2001;
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
const int64_t consume_txn_id = begin_target_transaction("consume-during-source-publish");
stage_target_rowset(consume_txn_id, kTargetPartitionId, kTargetTabletId);
CommitTxnRequest consume_request = make_consume_request(consume_txn_id, kSourcePartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
consume_request.set_commit_tso(999);
std::mutex mutex;
std::condition_variable condition;
bool consume_ready = false;
bool release_consume = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
{
std::lock_guard lock(mutex);
release_consume = true;
}
condition.notify_all();
sync_point->disable_processing();
sync_point->clear_all_call_backs();
};
sync_point->set_call_back("commit_txn_immediately::before_commit", [&](auto&&) {
std::unique_lock lock(mutex);
consume_ready = true;
condition.notify_all();
condition.wait(lock, [&] { return release_consume; });
});
sync_point->enable_processing();
CommitTxnResponse consume_response;
std::thread consume_thread([&] { consume_response = commit_transaction(consume_request); });
bool reached_commit = false;
{
std::unique_lock lock(mutex);
reached_commit =
condition.wait_for(lock, std::chrono::seconds(10), [&] { return consume_ready; });
}
TxnErrorCode create_publish_txn = TxnErrorCode::TXN_OK;
TxnErrorCode commit_publish_txn = TxnErrorCode::TXN_OK;
if (reached_commit) {
create_publish_txn = service_->txn_kv()->create_txn(&txn);
if (create_publish_txn == TxnErrorCode::TXN_OK) {
VersionPB published_version;
published_version.set_version(9);
published_version.set_commit_tso(160);
txn->put(partition_version_key({instance_id_, identity_.base_db_id(),
identity_.base_table_id(), kSourcePartitionId}),
published_version.SerializeAsString());
commit_publish_txn = txn->commit();
}
}
{
std::lock_guard lock(mutex);
release_consume = true;
}
condition.notify_all();
consume_thread.join();
ASSERT_TRUE(reached_commit);
ASSERT_EQ(create_publish_txn, TxnErrorCode::TXN_OK);
ASSERT_EQ(commit_publish_txn, TxnErrorCode::TXN_OK);
ASSERT_EQ(consume_response.status().code(), MetaServiceCode::OK)
<< consume_response.status().msg();
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
ASSERT_EQ(txn->get(partition_version_key({instance_id_, identity_.base_db_id(),
identity_.base_table_id(), kSourcePartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB source_version;
ASSERT_TRUE(source_version.ParseFromString(value));
EXPECT_EQ(source_version.version(), 9);
EXPECT_EQ(source_version.commit_tso(), 160);
EXPECT_EQ(
txn->get(meta_rowset_tmp_key({instance_id_, consume_txn_id, kTargetTabletId}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB committed_rowset;
ASSERT_TRUE(committed_rowset.ParseFromString(value));
EXPECT_EQ(committed_rowset.txn_id(), consume_txn_id);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB target_version;
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 2);
EXPECT_EQ(target_version.commit_tso(), 999);
ASSERT_EQ(txn->get(txn_info_key({instance_id_, kTargetDbId, consume_txn_id}), &value),
TxnErrorCode::TXN_OK);
TxnInfoPB txn_info;
ASSERT_TRUE(txn_info.ParseFromString(value));
EXPECT_EQ(txn_info.status(), TXN_STATUS_VISIBLE);
}
TEST_F(MetaServiceTableStreamTest, CommitRetryAfterMetaServiceRestartIsExactlyOnce) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSourcePartitionId = 2001;
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
const int64_t txn_id = begin_target_transaction("consume-retry-after-ms-restart");
stage_target_rowset(txn_id, kTargetPartitionId, kTargetTabletId);
CommitTxnRequest request = make_consume_request(txn_id, kSourcePartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
request.set_commit_tso(999);
// The commit succeeds, but its response is lost before the caller observes it.
const CommitTxnResponse lost_response = commit_transaction(request);
ASSERT_EQ(lost_response.status().code(), MetaServiceCode::OK) << lost_response.status().msg();
auto restarted_service = restart_meta_service();
CommitTxnResponse retry_response;
brpc::Controller retry_controller;
restarted_service->commit_txn(&retry_controller, &request, &retry_response, nullptr);
ASSERT_EQ(retry_response.status().code(), MetaServiceCode::OK) << retry_response.status().msg();
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, txn_id, kTargetTabletId}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_OK);
doris::RowsetMetaCloudPB committed_rowset;
ASSERT_TRUE(committed_rowset.ParseFromString(value));
EXPECT_EQ(committed_rowset.txn_id(), txn_id);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 3}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB target_version;
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 2);
EXPECT_EQ(target_version.commit_tso(), 999);
}
TEST_F(MetaServiceTableStreamTest, CommitRejectsRecyclingStreamBeforePublishingRowsetAndOffset) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSourcePartitionId = 2001;
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
const int64_t txn_id = begin_target_transaction("consume-recycling-stream");
stage_target_rowset(txn_id, kTargetPartitionId, kTargetTabletId);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
RecycleIndexPB recycle_index;
recycle_index.set_db_id(identity_.base_db_id());
recycle_index.set_table_id(identity_.base_table_id());
recycle_index.set_stream_db_id(identity_.stream_db_id());
recycle_index.set_state(RecycleIndexPB::RECYCLING);
recycle_index.set_object_type(IndexObjectTypePB::TABLE_STREAM);
txn->put(recycle_index_key({instance_id_, identity_.stream_id()}),
recycle_index.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
const CommitTxnRequest request = make_consume_request(txn_id, kSourcePartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
const CommitTxnResponse response = commit_transaction(request);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 100);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, txn_id, kTargetTabletId}), &value),
TxnErrorCode::TXN_OK);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB target_version;
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 1);
}
TEST_F(MetaServiceTableStreamTest, DropPartitionConflictsWithInFlightConsumption) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSourcePartitionId = 2001;
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
const int64_t txn_id = begin_target_transaction("consume-while-dropping-source-partition");
stage_target_rowset(txn_id, kTargetPartitionId, kTargetTabletId);
const CommitTxnRequest consume_request = make_consume_request(
txn_id, kSourcePartitionId, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
std::mutex mutex;
std::condition_variable condition;
bool commit_ready = false;
bool release_commit = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
{
std::lock_guard lock(mutex);
release_commit = true;
}
condition.notify_all();
sync_point->disable_processing();
sync_point->clear_all_call_backs();
};
sync_point->set_call_back("commit_txn_immediately::before_commit", [&](auto&&) {
std::unique_lock lock(mutex);
commit_ready = true;
condition.notify_all();
condition.wait(lock, [&] { return release_commit; });
});
sync_point->enable_processing();
CommitTxnResponse consume_response;
std::thread consume_thread([&] { consume_response = commit_transaction(consume_request); });
bool reached_commit = false;
{
std::unique_lock lock(mutex);
reached_commit =
condition.wait_for(lock, std::chrono::seconds(10), [&] { return commit_ready; });
}
PartitionRequest drop_request;
PartitionResponse drop_response;
if (reached_commit) {
drop_request.set_cloud_unique_id(cloud_unique_id_);
drop_request.set_db_id(identity_.base_db_id());
drop_request.set_table_id(identity_.base_table_id());
drop_request.add_index_ids(identity_.base_table_id() + 1);
drop_request.add_partition_ids(kSourcePartitionId);
drop_request.mutable_table_streams()->Add()->CopyFrom(identity_);
drop_request.set_expiration(::time(nullptr) + 3600);
brpc::Controller drop_controller;
service_->drop_partition(&drop_controller, &drop_request, &drop_response, nullptr);
}
{
std::lock_guard lock(mutex);
release_commit = true;
}
condition.notify_all();
consume_thread.join();
ASSERT_TRUE(reached_commit);
ASSERT_EQ(drop_response.status().code(), MetaServiceCode::OK) << drop_response.status().msg();
EXPECT_EQ(consume_response.status().code(), MetaServiceCode::KV_TXN_CONFLICT)
<< consume_response.status().msg();
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 100);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
ASSERT_EQ(txn->get(recycle_partition_key({instance_id_, kSourcePartitionId}), &value),
TxnErrorCode::TXN_OK);
RecyclePartitionPB recycle_partition;
ASSERT_TRUE(recycle_partition.ParseFromString(value));
EXPECT_EQ(recycle_partition.state(), RecyclePartitionPB::DROPPED);
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, txn_id, kTargetTabletId}), &value),
TxnErrorCode::TXN_OK);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB target_version;
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 1);
}
TEST_F(MetaServiceTableStreamTest, ConcurrentOffsetCasAllowsOnlyOneCommit) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kPartitionId = 2001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kPartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
const CommitTxnRequest first_request =
make_consume_request(begin_target_transaction("concurrent-cas-first"), kPartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
const CommitTxnRequest second_request =
make_consume_request(begin_target_transaction("concurrent-cas-second"), kPartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 125);
auto [first_response, second_response] = commit_concurrently(first_request, second_request);
const int successful_commits = (first_response.status().code() == MetaServiceCode::OK ? 1 : 0) +
(second_response.status().code() == MetaServiceCode::OK ? 1 : 0);
const int conflicted_commits =
(first_response.status().code() == MetaServiceCode::KV_TXN_CONFLICT ? 1 : 0) +
(second_response.status().code() == MetaServiceCode::KV_TXN_CONFLICT ? 1 : 0);
EXPECT_EQ(successful_commits, 1);
EXPECT_EQ(conflicted_commits, 1);
const int64_t final_offset = get_latest_offset(kPartitionId).offset_tso();
EXPECT_TRUE(final_offset == 120 || final_offset == 125);
}
TEST_F(MetaServiceTableStreamTest, ConcurrentUnknownOffsetCasAllowsOnlyOneFirstConsumer) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kPartitionId = 2001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kPartitionId, 8, 130, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
const CommitTxnRequest first_request =
make_consume_request(begin_target_transaction("concurrent-unknown-first"), kPartitionId,
TABLE_STREAM_OFFSET_UNKNOWN, 0, 120);
const CommitTxnRequest second_request =
make_consume_request(begin_target_transaction("concurrent-unknown-second"),
kPartitionId, TABLE_STREAM_OFFSET_UNKNOWN, 0, 125);
auto [first_response, second_response] = commit_concurrently(first_request, second_request);
const int successful_commits = (first_response.status().code() == MetaServiceCode::OK ? 1 : 0) +
(second_response.status().code() == MetaServiceCode::OK ? 1 : 0);
const int conflicted_commits =
(first_response.status().code() == MetaServiceCode::KV_TXN_CONFLICT ? 1 : 0) +
(second_response.status().code() == MetaServiceCode::KV_TXN_CONFLICT ? 1 : 0);
EXPECT_EQ(successful_commits, 1);
EXPECT_EQ(conflicted_commits, 1);
const int64_t final_offset = get_latest_offset(kPartitionId).offset_tso();
EXPECT_TRUE(final_offset == 120 || final_offset == 125);
}
TEST_F(MetaServiceTableStreamTest, ConcurrentIndependentOffsetUpdatesDoNotConflict) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSharedPartitionId = 2101;
constexpr int64_t kFirstPartitionId = 2201;
constexpr int64_t kSecondPartitionId = 2202;
TableStreamIdentityPB second_identity = identity_;
second_identity.set_stream_db_id(identity_.stream_db_id() + 1);
second_identity.set_stream_id(identity_.stream_id() + 1);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSharedPartitionId, 8, 130, std::nullopt);
put_latest_partition_state(txn.get(), kFirstPartitionId, 9, 140, std::nullopt);
put_latest_partition_state(txn.get(), kSecondPartitionId, 10, 150, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
CommitTxnRequest first_request =
make_consume_request(begin_target_transaction("concurrent-stream-first"),
kSharedPartitionId, TABLE_STREAM_OFFSET_UNKNOWN, 0, 120);
CommitTxnRequest second_request =
make_consume_request(begin_target_transaction("concurrent-stream-second"),
kSharedPartitionId, TABLE_STREAM_OFFSET_UNKNOWN, 0, 125);
second_request.mutable_table_stream_updates(0)->mutable_identity()->CopyFrom(second_identity);
auto [first_stream_response, second_stream_response] =
commit_concurrently(first_request, second_request);
ASSERT_EQ(first_stream_response.status().code(), MetaServiceCode::OK)
<< first_stream_response.status().msg();
ASSERT_EQ(second_stream_response.status().code(), MetaServiceCode::OK)
<< second_stream_response.status().msg();
EXPECT_EQ(get_latest_offset(identity_, kSharedPartitionId).offset_tso(), 120);
EXPECT_EQ(get_latest_offset(second_identity, kSharedPartitionId).offset_tso(), 125);
first_request = make_consume_request(begin_target_transaction("concurrent-partition-first"),
kFirstPartitionId, TABLE_STREAM_OFFSET_UNKNOWN, 0, 135);
second_request = make_consume_request(begin_target_transaction("concurrent-partition-second"),
kSecondPartitionId, TABLE_STREAM_OFFSET_UNKNOWN, 0, 145);
auto [first_partition_response, second_partition_response] =
commit_concurrently(first_request, second_request);
ASSERT_EQ(first_partition_response.status().code(), MetaServiceCode::OK)
<< first_partition_response.status().msg();
ASSERT_EQ(second_partition_response.status().code(), MetaServiceCode::OK)
<< second_partition_response.status().msg();
EXPECT_EQ(get_latest_offset(kFirstPartitionId).offset_tso(), 135);
EXPECT_EQ(get_latest_offset(kSecondPartitionId).offset_tso(), 145);
}
TEST_F(MetaServiceTableStreamTest, MultiVersionModeChangeConflictsWithInFlightConsumption) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kPartitionId = 2001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kPartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
const CommitTxnRequest request = make_consume_request(
begin_target_transaction("consume-while-changing-multi-version-mode"), kPartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
std::mutex mutex;
std::condition_variable condition;
bool commit_ready = false;
bool release_commit = false;
auto* sync_point = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
{
std::lock_guard lock(mutex);
release_commit = true;
}
condition.notify_all();
sync_point->disable_processing();
sync_point->clear_all_call_backs();
};
sync_point->set_call_back("commit_txn_immediately::before_commit", [&](auto&&) {
std::unique_lock lock(mutex);
commit_ready = true;
condition.notify_all();
condition.wait(lock, [&] { return release_commit; });
});
sync_point->enable_processing();
CommitTxnResponse response;
std::thread commit_thread([&] { response = commit_transaction(request); });
bool reached_commit = false;
{
std::unique_lock lock(mutex);
reached_commit =
condition.wait_for(lock, std::chrono::seconds(10), [&] { return commit_ready; });
}
TxnErrorCode create_mode_txn = TxnErrorCode::TXN_OK;
TxnErrorCode commit_mode_txn = TxnErrorCode::TXN_OK;
if (reached_commit) {
create_mode_txn = service_->txn_kv()->create_txn(&txn);
if (create_mode_txn == TxnErrorCode::TXN_OK) {
InstanceInfoPB instance;
instance.set_instance_id(instance_id_);
instance.set_multi_version_status(MULTI_VERSION_WRITE_ONLY);
txn->put(instance_key({instance_id_}), instance.SerializeAsString());
commit_mode_txn = txn->commit();
}
}
{
std::lock_guard lock(mutex);
release_commit = true;
}
condition.notify_all();
commit_thread.join();
ASSERT_TRUE(reached_commit);
ASSERT_EQ(create_mode_txn, TxnErrorCode::TXN_OK);
ASSERT_EQ(commit_mode_txn, TxnErrorCode::TXN_OK);
EXPECT_EQ(response.status().code(), MetaServiceCode::KV_TXN_CONFLICT)
<< response.status().msg();
EXPECT_EQ(get_latest_offset(kPartitionId).offset_tso(), 100);
}
TEST_F(MetaServiceTableStreamTest, CommitLargePartitionBatch) {
set_multi_version_status(MULTI_VERSION_WRITE_ONLY);
constexpr int kPartitionCount = 1201;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
for (int i = 0; i < kPartitionCount; ++i) {
int64_t partition_id = 30000 + i;
put_latest_partition_state(txn.get(), partition_id, i + 1, 40000 + i, std::nullopt);
}
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
CommitTxnRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.set_db_id(3001);
request.set_txn_id(begin_target_transaction("consume-large-partition-batch"));
TableStreamUpdatePB* stream_update = request.add_table_stream_updates();
stream_update->mutable_identity()->CopyFrom(identity_);
for (int i = 0; i < kPartitionCount; ++i) {
int64_t partition_id = 30000 + i;
TableStreamPartitionUpdatePB* update = stream_update->add_partition_updates();
update->set_partition_id(partition_id);
update->set_expected_state(TABLE_STREAM_OFFSET_UNKNOWN);
update->set_next_offset_tso(40000 + i);
}
CommitTxnResponse response = commit_transaction(request);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(30000).offset_tso(), 40000);
EXPECT_EQ(get_latest_offset(30000 + kPartitionCount / 2).offset_tso(),
40000 + kPartitionCount / 2);
EXPECT_EQ(get_latest_offset(30000 + kPartitionCount - 1).offset_tso(),
40000 + kPartitionCount - 1);
}
TEST_F(MetaServiceTableStreamTest, CommitRejectsStaleAndInvalidOffsets) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
int64_t txn_id = begin_target_transaction("consume-stale-offset");
CommitTxnResponse response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 99, 120);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
txn_id = begin_target_transaction("consume-backward-offset");
response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 90);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
txn_id = begin_target_transaction("consume-beyond-visible-tso");
response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 131);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
}
TEST_F(MetaServiceTableStreamTest, CommitValidatesOffsetTsoDomain) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, -1);
put_latest_partition_state(txn.get(), 2002, 9, 140, std::nullopt);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
CommitTxnResponse response =
consume_partition(begin_target_transaction("consume-empty-boundary-existing"), 2001,
TABLE_STREAM_OFFSET_CONSUMED, -1, -1);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(2001).offset_tso(), -1);
response = consume_partition(begin_target_transaction("consume-empty-boundary-unknown"), 2002,
TABLE_STREAM_OFFSET_UNKNOWN, 0, -1);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(2002).offset_tso(), -1);
for (int64_t invalid_tso : {0, -2}) {
response = consume_partition(
begin_target_transaction("consume-invalid-expected-" + std::to_string(invalid_tso)),
2001, TABLE_STREAM_OFFSET_CONSUMED, invalid_tso, -1);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), -1);
response = consume_partition(
begin_target_transaction("consume-invalid-next-" + std::to_string(invalid_tso)),
2001, TABLE_STREAM_OFFSET_CONSUMED, -1, invalid_tso);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), -1);
}
}
TEST_F(MetaServiceTableStreamTest, CommitUsesCloneEffectiveOffset) {
const std::string source_instance_id = "table_stream_read_state_source";
set_clone_source(source_instance_id, Versionstamp(100, 0));
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
PartitionIndexPB partition_index;
partition_index.set_db_id(identity_.base_db_id());
partition_index.set_table_id(identity_.base_table_id());
txn->put(versioned::partition_index_key({source_instance_id, 2001}),
partition_index.SerializeAsString());
versioned_put(txn.get(), versioned::meta_partition_key({source_instance_id, 2001}),
Versionstamp(41, 0), "");
VersionPB version;
version.set_version(18);
version.set_commit_tso(230);
versioned_put(txn.get(), versioned::partition_version_key({source_instance_id, 2001}),
Versionstamp(42, 0), version.SerializeAsString());
TableStreamOffsetPB inherited_offset;
inherited_offset.set_partition_id(2001);
inherited_offset.set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
inherited_offset.set_offset_tso(200);
versioned_put(txn.get(),
versioned::table_stream_offset_key(
{source_instance_id, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), 2001}),
Versionstamp(43, 0), inherited_offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
int64_t txn_id = begin_target_transaction("consume-inherited-offset-as-unknown");
CommitTxnResponse response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_UNKNOWN, 0, 220);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
txn_id = begin_target_transaction("consume-inherited-offset");
response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING, 200, 220);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 220);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
Versionstamp offset_version;
ASSERT_EQ(
versioned_get(txn.get(),
versioned::table_stream_offset_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), 2001}),
&offset_version, &value),
TxnErrorCode::TXN_OK);
TableStreamOffsetPB versioned_offset;
ASSERT_TRUE(versioned_offset.ParseFromString(value));
EXPECT_EQ(versioned_offset.offset_tso(), 220);
OperationLogPB operation_log;
Versionstamp log_version;
ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
&operation_log),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(operation_log.has_commit_txn());
ASSERT_EQ(operation_log.commit_txn().table_stream_offset_gc_size(), 1);
const TableStreamPartitionSetPB& offset_gc =
operation_log.commit_txn().table_stream_offset_gc(0);
EXPECT_EQ(offset_gc.identity().SerializeAsString(), identity_.SerializeAsString());
ASSERT_EQ(offset_gc.partition_ids_size(), 1);
EXPECT_EQ(offset_gc.partition_ids(0), 2001);
}
TEST_F(MetaServiceTableStreamTest, CommitRejectsLatestOffsetWithoutVersionedOffset) {
set_multi_version_status(MULTI_VERSION_READ_WRITE);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_versioned_partition_state(txn.get(), 2001, 8, 130, std::nullopt);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
int64_t txn_id = begin_target_transaction("consume-inconsistent-local-offset");
CommitTxnResponse response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_UNKNOWN, 0, 120);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
}
TEST_F(MetaServiceTableStreamTest, CommitRejectsDifferentLatestAndVersionedOffsetValues) {
set_multi_version_status(MULTI_VERSION_READ_WRITE);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_versioned_partition_state(txn.get(), 2001, 8, 130, std::nullopt);
TableStreamOffsetPB latest_offset;
latest_offset.set_partition_id(2001);
latest_offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
latest_offset.set_offset_tso(100);
latest_offset.set_last_consumption_time_ms(1234);
TableStreamOffsetKeyInfo key_info {instance_id_,
identity_.base_db_id(),
identity_.base_table_id(),
identity_.stream_db_id(),
identity_.stream_id(),
2001};
txn->put(table_stream_offset_key(key_info), latest_offset.SerializeAsString());
TableStreamOffsetPB versioned_offset = latest_offset;
versioned_offset.set_last_consumption_time_ms(1235);
versioned_put(txn.get(),
versioned::table_stream_offset_key(versioned::TableStreamOffsetKeyInfo(key_info)),
Versionstamp(43, 0), versioned_offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
int64_t txn_id = begin_target_transaction("consume-inconsistent-offset-values");
CommitTxnResponse response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).last_consumption_time_ms(), 1234);
}
TEST_F(MetaServiceTableStreamTest, CommitRejectsUnsupportedModeAndMissingCommitTso) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
VersionPB version;
version.set_version(8);
txn->put(partition_version_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(), 2001}),
version.SerializeAsString());
TableStreamOffsetPB offset;
offset.set_partition_id(2001);
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(100);
txn->put(table_stream_offset_key({instance_id_, identity_.base_db_id(),
identity_.base_table_id(), identity_.stream_db_id(),
identity_.stream_id(), 2001}),
offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
int64_t txn_id = begin_target_transaction("consume-missing-visible-tso");
CommitTxnResponse response =
consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 110);
EXPECT_EQ(response.status().code(), MetaServiceCode::VERSION_NOT_FOUND);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
set_multi_version_status(MULTI_VERSION_ENABLED);
txn_id = begin_target_transaction("consume-enabled-mode");
response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 110);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 100);
}
TEST_F(MetaServiceTableStreamTest, CommitRejectsUnsupportedTxnModesAndForcesImmediate) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), 2001, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
CommitTxnRequest request = make_consume_request(begin_target_transaction("consume-2pc"), 2001,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
request.set_is_2pc(true);
EXPECT_EQ(commit_transaction(request).status().code(), MetaServiceCode::INVALID_ARGUMENT);
request = make_consume_request(begin_target_transaction("consume-txn-load"), 2001,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
request.set_is_txn_load(true);
EXPECT_EQ(commit_transaction(request).status().code(), MetaServiceCode::INVALID_ARGUMENT);
request = make_consume_request(begin_target_transaction("consume-sub-txn"), 2001,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
request.add_sub_txn_infos()->set_sub_txn_id(1);
EXPECT_EQ(commit_transaction(request).status().code(), MetaServiceCode::INVALID_ARGUMENT);
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
const int old_rowset_threshold = config::txn_lazy_commit_rowsets_thresold;
const int old_fuzzy_possibility = config::cloud_txn_lazy_commit_fuzzy_possibility;
const bool old_lazy_commit_enabled = config::enable_cloud_txn_lazy_commit;
config::txn_lazy_commit_rowsets_thresold = 0;
config::cloud_txn_lazy_commit_fuzzy_possibility = 100;
config::enable_cloud_txn_lazy_commit = true;
DORIS_CLOUD_DEFER {
config::txn_lazy_commit_rowsets_thresold = old_rowset_threshold;
config::cloud_txn_lazy_commit_fuzzy_possibility = old_fuzzy_possibility;
config::enable_cloud_txn_lazy_commit = old_lazy_commit_enabled;
SyncPoint::get_instance()->clear_all_call_backs();
SyncPoint::get_instance()->clear_trace();
SyncPoint::get_instance()->disable_processing();
};
bool eventually_committed = false;
SyncPoint::get_instance()->set_call_back("commit_txn_eventually::finish",
[&](auto&&) { eventually_committed = true; });
SyncPoint::get_instance()->enable_processing();
const int64_t immediate_txn_id = begin_target_transaction("consume-immediate");
stage_target_rowset(immediate_txn_id, kTargetPartitionId, kTargetTabletId);
request = make_consume_request(immediate_txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
request.set_enable_txn_lazy_commit(true);
request.set_commit_tso(999);
CommitTxnResponse response = commit_transaction(request);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
EXPECT_FALSE(response.is_lazy_commit());
EXPECT_FALSE(eventually_committed);
EXPECT_EQ(get_latest_offset(2001).offset_tso(), 120);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, immediate_txn_id, kTargetTabletId}),
&value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_OK);
}
TEST_F(MetaServiceTableStreamTest, CommitTxnTooLargeDoesNotFallBackToLazyCommit) {
set_multi_version_status(MULTI_VERSION_DISABLED);
constexpr int64_t kSourcePartitionId = 2001;
constexpr int64_t kTargetDbId = 3001;
constexpr int64_t kTargetTableId = 3002;
constexpr int64_t kTargetIndexId = 3003;
constexpr int64_t kTargetPartitionId = 4001;
constexpr int64_t kTargetTabletId = 5001;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), kSourcePartitionId, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
create_target_tablet(kTargetDbId, kTargetTableId, kTargetIndexId, kTargetPartitionId,
kTargetTabletId);
const int64_t txn_id = begin_target_transaction("consume-too-large");
stage_target_rowset(txn_id, kTargetPartitionId, kTargetTabletId);
const int old_rowset_threshold = config::txn_lazy_commit_rowsets_thresold;
const int old_fuzzy_possibility = config::cloud_txn_lazy_commit_fuzzy_possibility;
const bool old_lazy_commit_enabled = config::enable_cloud_txn_lazy_commit;
config::txn_lazy_commit_rowsets_thresold = 0;
config::cloud_txn_lazy_commit_fuzzy_possibility = 100;
config::enable_cloud_txn_lazy_commit = true;
DORIS_CLOUD_DEFER {
config::txn_lazy_commit_rowsets_thresold = old_rowset_threshold;
config::cloud_txn_lazy_commit_fuzzy_possibility = old_fuzzy_possibility;
config::enable_cloud_txn_lazy_commit = old_lazy_commit_enabled;
SyncPoint::get_instance()->clear_all_call_backs();
SyncPoint::get_instance()->clear_trace();
SyncPoint::get_instance()->disable_processing();
};
bool immediate_commit_injected = false;
bool eventually_committed = false;
SyncPoint::get_instance()->set_call_back(
"commit_txn_immediately::before_commit", [&](auto&& args) {
auto* err = try_any_cast<TxnErrorCode*>(args[0]);
*err = TxnErrorCode::TXN_BYTES_TOO_LARGE;
auto* code = try_any_cast<MetaServiceCode*>(args[1]);
*code = cast_as<ErrCategory::COMMIT>(*err);
auto* pred = try_any_cast<bool*>(args.back());
*pred = true;
immediate_commit_injected = true;
});
SyncPoint::get_instance()->set_call_back("commit_txn_eventually::finish",
[&](auto&&) { eventually_committed = true; });
SyncPoint::get_instance()->enable_processing();
CommitTxnRequest request = make_consume_request(txn_id, kSourcePartitionId,
TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
request.set_enable_txn_lazy_commit(true);
request.set_commit_tso(999);
const CommitTxnResponse response = commit_transaction(request);
EXPECT_EQ(response.status().code(), MetaServiceCode::INVALID_ARGUMENT);
EXPECT_NE(response.status().msg().find("table stream offset updates cannot use lazy commit"),
std::string::npos);
EXPECT_TRUE(immediate_commit_injected);
EXPECT_FALSE(eventually_committed);
EXPECT_FALSE(response.is_lazy_commit());
EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 100);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, txn_id, kTargetTabletId}), &value),
TxnErrorCode::TXN_OK);
EXPECT_EQ(txn->get(meta_rowset_key({instance_id_, kTargetTabletId, 2}), &value),
TxnErrorCode::TXN_KEY_NOT_FOUND);
ASSERT_EQ(txn->get(partition_version_key(
{instance_id_, kTargetDbId, kTargetTableId, kTargetPartitionId}),
&value),
TxnErrorCode::TXN_OK);
VersionPB target_version;
ASSERT_TRUE(target_version.ParseFromString(value));
EXPECT_EQ(target_version.version(), 1);
}
TEST_F(MetaServiceTableStreamTest, CommitWritesOffsetsForEachMultiVersionMode) {
struct ModeCase {
MultiVersionStatus mode;
const char* label;
bool succeeds;
bool writes_versioned;
};
const std::vector<ModeCase> mode_cases = {
{MULTI_VERSION_DISABLED, "disabled", true, false},
{MULTI_VERSION_WRITE_ONLY, "write-only", true, true},
{MULTI_VERSION_READ_WRITE, "read-write", true, true},
{MULTI_VERSION_ENABLED, "enabled", false, false},
};
for (size_t i = 0; i < mode_cases.size(); ++i) {
const ModeCase& mode_case = mode_cases[i];
SCOPED_TRACE(mode_case.label);
const int64_t partition_id = 2101 + i;
const std::string log_prefix = versioned::log_key({instance_id_});
if (mode_case.mode == MULTI_VERSION_READ_WRITE) {
set_multi_version_status(mode_case.mode);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_versioned_partition_state(txn.get(), partition_id, 8, 130, std::nullopt);
TableStreamOffsetPB latest_offset;
latest_offset.set_partition_id(partition_id);
latest_offset.set_state(TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING);
latest_offset.set_offset_tso(100);
const TableStreamOffsetKeyInfo key_info {instance_id_,
identity_.base_db_id(),
identity_.base_table_id(),
identity_.stream_db_id(),
identity_.stream_id(),
partition_id};
txn->put(table_stream_offset_key(key_info), latest_offset.SerializeAsString());
versioned_put(txn.get(),
versioned::table_stream_offset_key(
versioned::TableStreamOffsetKeyInfo(key_info)),
latest_offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
} else {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
put_latest_partition_state(txn.get(), partition_id, 8, 130, 100);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
if (mode_case.mode != MULTI_VERSION_DISABLED) {
set_multi_version_status(mode_case.mode);
}
}
const size_t log_count_before = count_keys(log_prefix, lexical_end(log_prefix));
const TableStreamOffsetStatePB expected_state =
mode_case.mode == MULTI_VERSION_READ_WRITE
? TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING
: TABLE_STREAM_OFFSET_CONSUMED;
const CommitTxnResponse response = consume_partition(
begin_target_transaction(fmt::format("consume-mode-{}", mode_case.label)),
partition_id, expected_state, 100, 120);
EXPECT_EQ(response.status().code(),
mode_case.succeeds ? MetaServiceCode::OK : MetaServiceCode::INVALID_ARGUMENT)
<< response.status().msg();
EXPECT_EQ(get_latest_offset(partition_id).offset_tso(), mode_case.succeeds ? 120 : 100);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
Versionstamp value_version;
std::string value;
const TxnErrorCode versioned_err = versioned_get(
txn.get(),
versioned::table_stream_offset_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), partition_id}),
&value_version, &value);
if (mode_case.writes_versioned) {
ASSERT_EQ(versioned_err, TxnErrorCode::TXN_OK);
TableStreamOffsetPB versioned_offset;
ASSERT_TRUE(versioned_offset.ParseFromString(value));
EXPECT_EQ(versioned_offset.state(), TABLE_STREAM_OFFSET_CONSUMED);
EXPECT_EQ(versioned_offset.offset_tso(), 120);
} else {
EXPECT_EQ(versioned_err, TxnErrorCode::TXN_KEY_NOT_FOUND);
}
EXPECT_EQ(count_keys(log_prefix, lexical_end(log_prefix)),
log_count_before + (mode_case.writes_versioned ? 1 : 0));
}
}
TEST_F(MetaServiceTableStreamTest, DropWritesTypedRecycleIndexInDisabledMode) {
set_multi_version_status(MULTI_VERSION_DISABLED);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
IndexRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.add_index_ids(identity_.stream_id());
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity_.stream_db_id());
request.set_expiration(0);
IndexResponse response;
brpc::Controller controller;
service_->drop_index(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
ASSERT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value),
TxnErrorCode::TXN_OK);
RecycleIndexPB recycle_index;
ASSERT_TRUE(recycle_index.ParseFromString(value));
EXPECT_EQ(recycle_index.state(), RecycleIndexPB::DROPPED);
EXPECT_EQ(recycle_index.object_type(), IndexObjectTypePB::TABLE_STREAM);
EXPECT_EQ(recycle_index.db_id(), identity_.base_db_id());
EXPECT_EQ(recycle_index.table_id(), identity_.base_table_id());
EXPECT_EQ(recycle_index.stream_db_id(), identity_.stream_db_id());
}
TEST_F(MetaServiceTableStreamTest, DropWritesTypedOperationLogInVersionedMode) {
set_multi_version_status(MULTI_VERSION_READ_WRITE);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
IndexRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.add_index_ids(identity_.stream_id());
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity_.stream_db_id());
request.set_expiration(0);
IndexResponse response;
brpc::Controller controller;
service_->drop_index(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
OperationLogPB operation_log;
Versionstamp log_version;
ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
&operation_log),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(operation_log.has_drop_index());
ASSERT_TRUE(operation_log.has_min_timestamp());
EXPECT_EQ(operation_log.min_timestamp(),
static_cast<uint64_t>(std::numeric_limits<int64_t>::max()));
const DropIndexLogPB& drop_index = operation_log.drop_index();
EXPECT_EQ(drop_index.object_type(), IndexObjectTypePB::TABLE_STREAM);
EXPECT_EQ(drop_index.db_id(), identity_.base_db_id());
EXPECT_EQ(drop_index.table_id(), identity_.base_table_id());
EXPECT_EQ(drop_index.stream_db_id(), identity_.stream_db_id());
ASSERT_EQ(drop_index.index_ids_size(), 1);
EXPECT_EQ(drop_index.index_ids(0), identity_.stream_id());
std::string value;
EXPECT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value, true),
TxnErrorCode::TXN_KEY_NOT_FOUND);
}
TEST_F(MetaServiceTableStreamTest, DropRetryAfterMetaServiceRestartConverges) {
const bool old_force_immediate_recycle = config::force_immediate_recycle;
config::force_immediate_recycle = true;
DORIS_CLOUD_DEFER {
config::force_immediate_recycle = old_force_immediate_recycle;
};
set_multi_version_status(MULTI_VERSION_READ_WRITE);
constexpr int64_t kPartitionId = 2001;
put_auto_versioned_offset(instance_id_, identity_, kPartitionId, 100, true);
IndexRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.add_index_ids(identity_.stream_id());
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity_.stream_db_id());
request.set_expiration(0);
// The first DROP succeeds, but its response is lost before the caller observes it.
IndexResponse lost_response;
brpc::Controller first_drop_controller;
service_->drop_index(&first_drop_controller, &request, &lost_response, nullptr);
ASSERT_EQ(lost_response.status().code(), MetaServiceCode::OK) << lost_response.status().msg();
const std::string operation_log_prefix = versioned::log_key({instance_id_});
const size_t first_drop_log_count =
count_keys(operation_log_prefix, lexical_end(operation_log_prefix));
ASSERT_GT(first_drop_log_count, 0);
auto restarted_service = restart_meta_service();
IndexResponse retry_response;
brpc::Controller retry_drop_controller;
restarted_service->drop_index(&retry_drop_controller, &request, &retry_response, nullptr);
ASSERT_EQ(retry_response.status().code(), MetaServiceCode::OK) << retry_response.status().msg();
EXPECT_GE(count_keys(operation_log_prefix, lexical_end(operation_log_prefix)),
first_drop_log_count);
InstanceInfoPB instance;
instance.set_instance_id(instance_id_);
instance.set_status(InstanceInfoPB::NORMAL);
instance.set_multi_version_status(MULTI_VERSION_READ_WRITE);
InstanceRecycler recycler(service_->txn_kv(), instance, RecyclerThreadPoolGroup {},
std::make_shared<TxnLazyCommitter>(service_->txn_kv()));
ASSERT_EQ(recycler.init(), 0);
ASSERT_EQ(recycler.recycle_operation_logs(), 0);
EXPECT_EQ(count_keys(operation_log_prefix, lexical_end(operation_log_prefix)), 0);
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
ASSERT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value, true),
TxnErrorCode::TXN_OK);
RecycleIndexPB recycle_index;
ASSERT_TRUE(recycle_index.ParseFromString(value));
EXPECT_EQ(recycle_index.state(), RecycleIndexPB::DROPPED);
EXPECT_EQ(recycle_index.object_type(), IndexObjectTypePB::TABLE_STREAM);
ASSERT_EQ(recycler.recycle_indexes(), 0);
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
EXPECT_EQ(txn->get(recycle_index_key({instance_id_, identity_.stream_id()}), &value, true),
TxnErrorCode::TXN_KEY_NOT_FOUND);
const std::string latest_offset_prefix = table_stream_offset_key_prefix(
instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id());
const std::string versioned_offset_prefix = versioned::table_stream_offset_key_prefix(
instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id());
EXPECT_EQ(count_keys(latest_offset_prefix, lexical_end(latest_offset_prefix)), 0);
EXPECT_EQ(count_keys(versioned_offset_prefix, lexical_end(versioned_offset_prefix)), 0);
}
TEST_F(MetaServiceTableStreamTest, DropUsesEarliestLocalVersionedOffset) {
const std::string source_instance_id = "table_stream_drop_source";
const Versionstamp source_version =
put_auto_versioned_offset(source_instance_id, identity_, 2001, 10);
set_clone_source(source_instance_id, Versionstamp(source_version.version() + 1, 0));
// Partition 2002 sorts after partition 2001 but contains the earliest local version.
const Versionstamp min_local_version =
put_auto_versioned_offset(instance_id_, identity_, 2002, 30);
put_auto_versioned_offset(instance_id_, identity_, 2001, 10);
put_auto_versioned_offset(instance_id_, identity_, 2001, 20);
IndexRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.add_index_ids(identity_.stream_id());
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity_.stream_db_id());
request.set_expiration(0);
IndexResponse response;
brpc::Controller controller;
service_->drop_index(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
OperationLogPB operation_log;
Versionstamp log_version;
ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
&operation_log),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(operation_log.has_drop_index());
ASSERT_TRUE(operation_log.has_min_timestamp());
EXPECT_LT(source_version.version(), min_local_version.version());
EXPECT_EQ(operation_log.min_timestamp(), min_local_version.version());
EXPECT_LT(operation_log.min_timestamp(), log_version.version());
}
TEST_F(MetaServiceTableStreamTest, DropFindsMinimumInLargeVersionedOffsetHistory) {
set_multi_version_status(MULTI_VERSION_READ_WRITE);
const int default_full_range_batch_size = FullRangeGetOptions().batch_limit;
std::unique_ptr<Transaction> txn;
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
TableStreamOffsetPB offset;
offset.set_state(TABLE_STREAM_OFFSET_CONSUMED);
offset.set_offset_tso(100);
for (int i = 0; i < default_full_range_batch_size; ++i) {
offset.set_partition_id(2001);
versioned_put(
txn.get(),
versioned::table_stream_offset_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), offset.partition_id()}),
Versionstamp(1000 + i, 0), offset.SerializeAsString());
}
// This key sorts after all partition 2001 history but carries the global minimum version.
offset.set_partition_id(2002);
versioned_put(txn.get(),
versioned::table_stream_offset_key(
{instance_id_, identity_.base_db_id(), identity_.base_table_id(),
identity_.stream_db_id(), identity_.stream_id(), offset.partition_id()}),
Versionstamp(50, 0), offset.SerializeAsString());
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
IndexRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.add_index_ids(identity_.stream_id());
request.set_db_id(identity_.base_db_id());
request.set_table_id(identity_.base_table_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity_.stream_db_id());
request.set_expiration(0);
IndexResponse response;
brpc::Controller controller;
service_->drop_index(&controller, &request, &response, nullptr);
ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
OperationLogPB operation_log;
Versionstamp log_version;
ASSERT_EQ(read_operation_log(txn.get(), versioned::log_key({instance_id_}), &log_version,
&operation_log),
TxnErrorCode::TXN_OK);
ASSERT_TRUE(operation_log.has_drop_index());
ASSERT_TRUE(operation_log.has_min_timestamp());
EXPECT_EQ(operation_log.min_timestamp(), 50);
}
TEST_F(MetaServiceTableStreamTest, DropAndRecycleCloneChildOffsets) {
const bool old_force_immediate_recycle = config::force_immediate_recycle;
config::force_immediate_recycle = true;
DORIS_CLOUD_DEFER {
config::force_immediate_recycle = old_force_immediate_recycle;
};
const std::string source_instance_id = "table_stream_drop_recycle_source";
TableStreamIdentityPB local_identity = identity_;
TableStreamIdentityPB inherited_identity = identity_;
inherited_identity.set_stream_id(identity_.stream_id() + 1);
put_auto_versioned_offset(source_instance_id, local_identity, 2001, 90);
const Versionstamp source_version =
put_auto_versioned_offset(source_instance_id, inherited_identity, 2001, 80);
const Versionstamp source_snapshot = Versionstamp::next(source_version);
set_clone_source(source_instance_id, source_snapshot);
put_auto_versioned_offset(instance_id_, local_identity, 2001, 100, true);
auto drop_stream = [&](const TableStreamIdentityPB& identity) {
IndexRequest request;
request.set_cloud_unique_id(cloud_unique_id_);
request.add_index_ids(identity.stream_id());
request.set_db_id(identity.base_db_id());
request.set_table_id(identity.base_table_id());
request.set_object_type(IndexObjectTypePB::TABLE_STREAM);
request.set_stream_db_id(identity.stream_db_id());
request.set_expiration(0);
IndexResponse response;
brpc::Controller controller;
service_->drop_index(&controller, &request, &response, nullptr);
EXPECT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
};
drop_stream(local_identity);
drop_stream(inherited_identity);
InstanceInfoPB child_instance;
child_instance.set_instance_id(instance_id_);
child_instance.set_status(InstanceInfoPB::NORMAL);
child_instance.set_multi_version_status(MULTI_VERSION_READ_WRITE);
child_instance.set_source_instance_id(source_instance_id);
child_instance.set_source_snapshot_id(SnapshotManager::serialize_snapshot_id(source_snapshot));
InstanceRecycler recycler(service_->txn_kv(), child_instance, RecyclerThreadPoolGroup {},
std::make_shared<TxnLazyCommitter>(service_->txn_kv()));
ASSERT_EQ(recycler.init(), 0);
ASSERT_EQ(recycler.recycle_operation_logs(), 0);
auto key_exists = [&](std::string_view key) {
std::unique_ptr<Transaction> txn;
EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string value;
return txn->get(key, &value, true) == TxnErrorCode::TXN_OK;
};
EXPECT_TRUE(key_exists(recycle_index_key({instance_id_, local_identity.stream_id()})));
EXPECT_TRUE(key_exists(recycle_index_key({instance_id_, inherited_identity.stream_id()})));
ASSERT_EQ(recycler.recycle_indexes(), 0);
const auto offset_count = [&](const std::string& prefix) {
std::unique_ptr<Transaction> txn;
EXPECT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::unique_ptr<RangeGetIterator> iter;
EXPECT_EQ(txn->get(prefix, lexical_end(prefix), &iter, true, 0), TxnErrorCode::TXN_OK);
return iter->size();
};
EXPECT_EQ(offset_count(table_stream_offset_key_prefix(
instance_id_, local_identity.base_db_id(), local_identity.base_table_id(),
local_identity.stream_db_id(), local_identity.stream_id())),
0);
EXPECT_EQ(offset_count(versioned::table_stream_offset_key_prefix(
instance_id_, local_identity.base_db_id(), local_identity.base_table_id(),
local_identity.stream_db_id(), local_identity.stream_id())),
0);
EXPECT_EQ(offset_count(versioned::table_stream_offset_key_prefix(
instance_id_, inherited_identity.base_db_id(),
inherited_identity.base_table_id(), inherited_identity.stream_db_id(),
inherited_identity.stream_id())),
0);
EXPECT_FALSE(key_exists(recycle_index_key({instance_id_, local_identity.stream_id()})));
EXPECT_FALSE(key_exists(recycle_index_key({instance_id_, inherited_identity.stream_id()})));
EXPECT_GT(
offset_count(versioned::table_stream_offset_key_prefix(
source_instance_id, local_identity.base_db_id(), local_identity.base_table_id(),
local_identity.stream_db_id(), local_identity.stream_id())),
0);
EXPECT_GT(offset_count(versioned::table_stream_offset_key_prefix(
source_instance_id, inherited_identity.base_db_id(),
inherited_identity.base_table_id(), inherited_identity.stream_db_id(),
inherited_identity.stream_id())),
0);
}
} // namespace doris::cloud