| // 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 <gtest/gtest.h> |
| |
| #include <algorithm> |
| #include <memory> |
| #include <string> |
| |
| #include "common/bvars.h" |
| #include "common/config.h" |
| #include "common/defer.h" |
| #include "meta-store/keys.h" |
| #include "meta-store/mem_txn_kv.h" |
| #include "meta-store/versioned_value.h" |
| #include "recycler/recycler.h" |
| |
| namespace doris::cloud { |
| namespace { |
| |
| constexpr std::string_view kInstanceId = "table_stream_recycler_instance"; |
| constexpr int64_t kBaseDbId = 1001; |
| constexpr int64_t kBaseTableId = 1002; |
| constexpr int64_t kStreamDbId = 1003; |
| constexpr int64_t kStreamId = 1004; |
| constexpr int64_t kPartitionId = 1005; |
| |
| std::string stream_offset_key(int64_t partition_id, int64_t stream_id = kStreamId) { |
| return table_stream_offset_key({std::string(kInstanceId), kBaseDbId, kBaseTableId, kStreamDbId, |
| stream_id, partition_id}); |
| } |
| |
| std::string versioned_stream_offset_key(int64_t partition_id, int64_t stream_id = kStreamId) { |
| return versioned::table_stream_offset_key({std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, stream_id, partition_id}); |
| } |
| |
| void put_offset(Transaction* txn, int64_t partition_id, int64_t stream_id = kStreamId) { |
| TableStreamOffsetPB offset; |
| offset.set_partition_id(partition_id); |
| offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED); |
| offset.set_offset_tso(100); |
| const auto latest_key = stream_offset_key(partition_id, stream_id); |
| const auto versioned_key = versioned_stream_offset_key(partition_id, stream_id); |
| txn->put(latest_key, offset.SerializeAsString()); |
| versioned_put(txn, versioned_key, Versionstamp(11, 0), offset.SerializeAsString()); |
| offset.set_offset_tso(110); |
| versioned_put(txn, versioned_key, Versionstamp(12, 0), offset.SerializeAsString()); |
| } |
| |
| void put_single_version_offset(Transaction* txn, int64_t partition_id) { |
| TableStreamOffsetPB offset; |
| offset.set_partition_id(partition_id); |
| offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED); |
| offset.set_offset_tso(100); |
| txn->put(stream_offset_key(partition_id), offset.SerializeAsString()); |
| versioned_put(txn, versioned_stream_offset_key(partition_id), Versionstamp(11, 0), |
| offset.SerializeAsString()); |
| } |
| |
| void remove_offset(Transaction* txn, int64_t partition_id, int64_t stream_id = kStreamId) { |
| txn->remove(stream_offset_key(partition_id, stream_id)); |
| versioned_remove_all(txn, versioned_stream_offset_key(partition_id, stream_id)); |
| } |
| |
| void put_stream_recycle_index(Transaction* txn, RecycleIndexPB::State state, |
| int64_t stream_id = kStreamId) { |
| RecycleIndexPB recycle_index; |
| recycle_index.set_db_id(kBaseDbId); |
| recycle_index.set_table_id(kBaseTableId); |
| recycle_index.set_creation_time(0); |
| recycle_index.set_expiration(0); |
| recycle_index.set_state(state); |
| recycle_index.set_object_type(IndexObjectTypePB::TABLE_STREAM); |
| recycle_index.set_stream_db_id(kStreamDbId); |
| txn->put(recycle_index_key({std::string(kInstanceId), stream_id}), |
| recycle_index.SerializeAsString()); |
| } |
| |
| void add_stream_identity(RecyclePartitionPB* recycle_partition, int64_t stream_id) { |
| TableStreamIdentityPB* identity = recycle_partition->add_table_streams(); |
| identity->set_base_db_id(kBaseDbId); |
| identity->set_base_table_id(kBaseTableId); |
| identity->set_stream_db_id(kStreamDbId); |
| identity->set_stream_id(stream_id); |
| } |
| |
| bool key_exists(TxnKv* txn_kv, std::string_view key) { |
| std::unique_ptr<Transaction> txn; |
| EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| std::string value; |
| return txn->get(key, &value, true) == TxnErrorCode::TXN_OK; |
| } |
| |
| size_t range_size(TxnKv* txn_kv, const std::string& prefix) { |
| std::unique_ptr<Transaction> txn; |
| EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| std::string end = prefix; |
| end.push_back('\xff'); |
| std::unique_ptr<RangeGetIterator> iter; |
| EXPECT_EQ(txn->get(prefix, end, &iter, true), TxnErrorCode::TXN_OK); |
| return iter->size(); |
| } |
| |
| InstanceRecycler make_recycler(const std::shared_ptr<TxnKv>& txn_kv) { |
| InstanceInfoPB instance; |
| instance.set_instance_id(std::string(kInstanceId)); |
| return InstanceRecycler(txn_kv, instance, RecyclerThreadPoolGroup {}, |
| std::make_shared<TxnLazyCommitter>(txn_kv)); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecycleStreamDeletesOnlyOffsets) { |
| 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; |
| }; |
| |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_offset(txn.get(), kPartitionId); |
| put_offset(txn.get(), kPartitionId + 1); |
| |
| RecycleIndexPB recycle_index; |
| recycle_index.set_db_id(kBaseDbId); |
| recycle_index.set_table_id(kBaseTableId); |
| recycle_index.set_creation_time(0); |
| recycle_index.set_expiration(0); |
| recycle_index.set_state(RecycleIndexPB::DROPPED); |
| recycle_index.set_object_type(IndexObjectTypePB::TABLE_STREAM); |
| recycle_index.set_stream_db_id(kStreamDbId); |
| txn->put(recycle_index_key({std::string(kInstanceId), kStreamId}), |
| recycle_index.SerializeAsString()); |
| const std::string unrelated_key = meta_tablet_key( |
| {std::string(kInstanceId), kBaseTableId, kStreamId, kPartitionId, 2001}); |
| txn->put(unrelated_key, "unrelated physical data"); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| const int recycle_round_before = g_bvar_recycler_instance_recycle_round.get( |
| {std::string(kInstanceId), "recycle_stream"}); |
| const int64_t recycle_total_before = |
| g_bvar_recycler_instance_recycle_total_num_since_started.get( |
| {std::string(kInstanceId), "recycle_stream"}); |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.recycle_indexes(), 0); |
| |
| EXPECT_FALSE( |
| key_exists(txn_kv.get(), recycle_index_key({std::string(kInstanceId), kStreamId}))); |
| EXPECT_EQ(range_size(txn_kv.get(), |
| table_stream_offset_key_prefix(std::string(kInstanceId), kBaseDbId, |
| kBaseTableId, kStreamDbId, kStreamId)), |
| 0); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned::table_stream_offset_key_prefix( |
| std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId)), |
| 0); |
| EXPECT_TRUE(key_exists(txn_kv.get(), unrelated_key)); |
| EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_num.get( |
| {std::string(kInstanceId), "recycle_stream"}), |
| 6); |
| EXPECT_EQ(g_bvar_recycler_instance_recycle_round.get( |
| {std::string(kInstanceId), "recycle_stream"}), |
| recycle_round_before + 1); |
| EXPECT_EQ(g_bvar_recycler_instance_recycle_total_num_since_started.get( |
| {std::string(kInstanceId), "recycle_stream"}), |
| recycle_total_before + 6); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecyclePreparedStreamAfterPartialOffsetInitialization) { |
| 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; |
| }; |
| |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_offset(txn.get(), kPartitionId); |
| put_offset(txn.get(), kPartitionId + 1); |
| put_offset(txn.get(), kPartitionId + 2); |
| txn->remove(stream_offset_key(kPartitionId + 1)); |
| versioned_remove_all(txn.get(), versioned_stream_offset_key(kPartitionId + 2)); |
| put_stream_recycle_index(txn.get(), RecycleIndexPB::PREPARED); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| ASSERT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId))); |
| ASSERT_FALSE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId + 1))); |
| ASSERT_GT(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId + 1)), 0); |
| ASSERT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId + 2))); |
| ASSERT_EQ(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId + 2)), 0); |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.recycle_indexes(), 0); |
| |
| EXPECT_FALSE( |
| key_exists(txn_kv.get(), recycle_index_key({std::string(kInstanceId), kStreamId}))); |
| EXPECT_EQ(range_size(txn_kv.get(), |
| table_stream_offset_key_prefix(std::string(kInstanceId), kBaseDbId, |
| kBaseTableId, kStreamDbId, kStreamId)), |
| 0); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned::table_stream_offset_key_prefix( |
| std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId)), |
| 0); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecycleStreamResumesFromRecyclingAfterPartialDeletion) { |
| 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; |
| }; |
| |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_offset(txn.get(), kPartitionId); |
| put_offset(txn.get(), kPartitionId + 1); |
| put_offset(txn.get(), kPartitionId + 2); |
| remove_offset(txn.get(), kPartitionId); |
| txn->remove(stream_offset_key(kPartitionId + 1)); |
| put_stream_recycle_index(txn.get(), RecycleIndexPB::RECYCLING); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| ASSERT_FALSE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId))); |
| ASSERT_EQ(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId)), 0); |
| ASSERT_FALSE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId + 1))); |
| ASSERT_GT(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId + 1)), 0); |
| ASSERT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId + 2))); |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.recycle_indexes(), 0); |
| ASSERT_EQ(recycler.recycle_indexes(), 0); |
| |
| EXPECT_FALSE( |
| key_exists(txn_kv.get(), recycle_index_key({std::string(kInstanceId), kStreamId}))); |
| EXPECT_EQ(range_size(txn_kv.get(), |
| table_stream_offset_key_prefix(std::string(kInstanceId), kBaseDbId, |
| kBaseTableId, kStreamDbId, kStreamId)), |
| 0); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned::table_stream_offset_key_prefix( |
| std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId)), |
| 0); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecycleStreamDeletesOffsetsAcrossRangePages) { |
| 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; |
| }; |
| |
| constexpr int64_t offset_count = 10001; |
| constexpr int64_t seed_batch_size = 500; |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| for (int64_t begin = 0; begin < offset_count; begin += seed_batch_size) { |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| const int64_t end = std::min(offset_count, begin + seed_batch_size); |
| for (int64_t i = begin; i < end; ++i) { |
| put_single_version_offset(txn.get(), kPartitionId + i); |
| } |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| } |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_stream_recycle_index(txn.get(), RecycleIndexPB::DROPPED); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| ASSERT_EQ(range_size(txn_kv.get(), |
| table_stream_offset_key_prefix(std::string(kInstanceId), kBaseDbId, |
| kBaseTableId, kStreamDbId, kStreamId)), |
| 10000); |
| ASSERT_EQ(range_size(txn_kv.get(), versioned::table_stream_offset_key_prefix( |
| std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId)), |
| 10000); |
| ASSERT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId + offset_count - 1))); |
| ASSERT_GT( |
| range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId + offset_count - 1)), |
| 0); |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.recycle_indexes(), 0); |
| |
| EXPECT_FALSE( |
| key_exists(txn_kv.get(), recycle_index_key({std::string(kInstanceId), kStreamId}))); |
| EXPECT_EQ(range_size(txn_kv.get(), |
| table_stream_offset_key_prefix(std::string(kInstanceId), kBaseDbId, |
| kBaseTableId, kStreamDbId, kStreamId)), |
| 0); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned::table_stream_offset_key_prefix( |
| std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId)), |
| 0); |
| } |
| |
| TEST(TableStreamRecyclerTest, StatisticsDispatchesStreamToOffsetScan) { |
| 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; |
| }; |
| |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_offset(txn.get(), kPartitionId); |
| put_offset(txn.get(), kPartitionId + 1); |
| |
| RecycleIndexPB recycle_index; |
| recycle_index.set_db_id(kBaseDbId); |
| recycle_index.set_table_id(kBaseTableId); |
| recycle_index.set_creation_time(0); |
| recycle_index.set_expiration(0); |
| recycle_index.set_state(RecycleIndexPB::DROPPED); |
| recycle_index.set_object_type(IndexObjectTypePB::TABLE_STREAM); |
| recycle_index.set_stream_db_id(kStreamDbId); |
| txn->put(recycle_index_key({std::string(kInstanceId), kStreamId}), |
| recycle_index.SerializeAsString()); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.scan_and_statistics_indexes(), 0); |
| |
| EXPECT_EQ(g_bvar_recycler_instance_last_round_to_recycle_num.get( |
| {std::string(kInstanceId), "recycle_stream"}), |
| 6); |
| EXPECT_EQ(g_bvar_recycler_instance_last_round_to_recycle_num.get( |
| {std::string(kInstanceId), "recycle_indexes"}), |
| 0); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecyclePartitionDeletesOnlyThatPartitionOffsets) { |
| 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; |
| }; |
| |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_offset(txn.get(), kPartitionId); |
| put_offset(txn.get(), kPartitionId + 1); |
| |
| RecyclePartitionPB recycle_partition; |
| recycle_partition.set_db_id(kBaseDbId); |
| recycle_partition.set_table_id(kBaseTableId); |
| recycle_partition.add_index_id(2001); |
| recycle_partition.set_creation_time(0); |
| recycle_partition.set_expiration(0); |
| recycle_partition.set_state(RecyclePartitionPB::DROPPED); |
| TableStreamIdentityPB* identity = recycle_partition.add_table_streams(); |
| identity->set_base_db_id(kBaseDbId); |
| identity->set_base_table_id(kBaseTableId); |
| identity->set_stream_db_id(kStreamDbId); |
| identity->set_stream_id(kStreamId); |
| txn->put(recycle_partition_key({std::string(kInstanceId), kPartitionId}), |
| recycle_partition.SerializeAsString()); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.recycle_partitions(), 0); |
| |
| const auto dropped_latest = |
| table_stream_offset_key({std::string(kInstanceId), kBaseDbId, kBaseTableId, kStreamDbId, |
| kStreamId, kPartitionId}); |
| const auto retained_latest = |
| table_stream_offset_key({std::string(kInstanceId), kBaseDbId, kBaseTableId, kStreamDbId, |
| kStreamId, kPartitionId + 1}); |
| EXPECT_FALSE(key_exists(txn_kv.get(), dropped_latest)); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned::table_stream_offset_key( |
| {std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId, kPartitionId})), |
| 0); |
| EXPECT_TRUE(key_exists(txn_kv.get(), retained_latest)); |
| EXPECT_GT(range_size(txn_kv.get(), versioned::table_stream_offset_key( |
| {std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId, kPartitionId + 1})), |
| 0); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecyclePartitionUsesPersistedStreamIdentity) { |
| 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; |
| }; |
| |
| constexpr std::string_view source_instance_id = "table_stream_recycler_source"; |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| |
| InstanceInfoPB source_instance; |
| source_instance.set_instance_id(std::string(source_instance_id)); |
| txn->put(instance_key({std::string(source_instance_id)}), source_instance.SerializeAsString()); |
| InstanceInfoPB child_instance; |
| child_instance.set_instance_id(std::string(kInstanceId)); |
| child_instance.set_multi_version_status(MULTI_VERSION_READ_WRITE); |
| child_instance.set_source_instance_id(std::string(source_instance_id)); |
| child_instance.set_source_snapshot_id(Versionstamp(20, 0).to_string()); |
| txn->put(instance_key({std::string(kInstanceId)}), child_instance.SerializeAsString()); |
| |
| put_offset(txn.get(), kPartitionId); |
| TableStreamOffsetPB source_offset; |
| source_offset.set_partition_id(kPartitionId); |
| source_offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED); |
| source_offset.set_offset_tso(90); |
| const auto source_latest = |
| table_stream_offset_key({std::string(source_instance_id), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId, kPartitionId}); |
| const auto source_versioned = versioned::table_stream_offset_key( |
| {std::string(source_instance_id), kBaseDbId, kBaseTableId, kStreamDbId, kStreamId, |
| kPartitionId}); |
| txn->put(source_latest, source_offset.SerializeAsString()); |
| versioned_put(txn.get(), source_versioned, Versionstamp(11, 0), |
| source_offset.SerializeAsString()); |
| |
| RecyclePartitionPB recycle_partition; |
| recycle_partition.set_db_id(kBaseDbId); |
| recycle_partition.set_table_id(kBaseTableId); |
| recycle_partition.add_index_id(2001); |
| recycle_partition.set_creation_time(0); |
| recycle_partition.set_expiration(0); |
| recycle_partition.set_state(RecyclePartitionPB::DROPPED); |
| TableStreamIdentityPB* identity = recycle_partition.add_table_streams(); |
| identity->set_base_db_id(kBaseDbId); |
| identity->set_base_table_id(kBaseTableId); |
| identity->set_stream_db_id(kStreamDbId); |
| identity->set_stream_id(kStreamId); |
| txn->put(recycle_partition_key({std::string(kInstanceId), kPartitionId}), |
| recycle_partition.SerializeAsString()); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| InstanceRecycler recycler(txn_kv, child_instance, RecyclerThreadPoolGroup {}, |
| std::make_shared<TxnLazyCommitter>(txn_kv)); |
| ASSERT_EQ(recycler.recycle_partitions(), 0); |
| |
| const auto child_latest = |
| table_stream_offset_key({std::string(kInstanceId), kBaseDbId, kBaseTableId, kStreamDbId, |
| kStreamId, kPartitionId}); |
| const auto child_versioned = |
| versioned::table_stream_offset_key({std::string(kInstanceId), kBaseDbId, kBaseTableId, |
| kStreamDbId, kStreamId, kPartitionId}); |
| EXPECT_FALSE(key_exists(txn_kv.get(), child_latest)); |
| EXPECT_EQ(range_size(txn_kv.get(), child_versioned), 0); |
| EXPECT_TRUE(key_exists(txn_kv.get(), source_latest)); |
| EXPECT_GT(range_size(txn_kv.get(), source_versioned), 0); |
| } |
| |
| TEST(TableStreamRecyclerTest, RecyclePartitionOffsetsResumesAcrossBatches) { |
| const bool old_force_immediate_recycle = config::force_immediate_recycle; |
| const int32_t old_batch_size = config::recycler_max_tasks_per_batch; |
| config::force_immediate_recycle = true; |
| config::recycler_max_tasks_per_batch = 2; |
| DORIS_CLOUD_DEFER { |
| config::force_immediate_recycle = old_force_immediate_recycle; |
| config::recycler_max_tasks_per_batch = old_batch_size; |
| }; |
| |
| constexpr int64_t stream_count = 5; |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| |
| RecyclePartitionPB recycle_partition; |
| recycle_partition.set_db_id(kBaseDbId); |
| recycle_partition.set_table_id(kBaseTableId); |
| recycle_partition.add_index_id(2001); |
| recycle_partition.set_creation_time(0); |
| recycle_partition.set_expiration(0); |
| recycle_partition.set_state(RecyclePartitionPB::RECYCLING); |
| for (int64_t i = 0; i < stream_count; ++i) { |
| const int64_t stream_id = kStreamId + i; |
| add_stream_identity(&recycle_partition, stream_id); |
| put_offset(txn.get(), kPartitionId, stream_id); |
| put_offset(txn.get(), kPartitionId + 1, stream_id); |
| } |
| // The first batch was already committed before the previous recycler stopped. |
| remove_offset(txn.get(), kPartitionId, kStreamId); |
| remove_offset(txn.get(), kPartitionId, kStreamId + 1); |
| txn->put(recycle_partition_key({std::string(kInstanceId), kPartitionId}), |
| recycle_partition.SerializeAsString()); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| for (int64_t i = 0; i < 2; ++i) { |
| const int64_t stream_id = kStreamId + i; |
| ASSERT_FALSE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId, stream_id))); |
| ASSERT_EQ(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId, stream_id)), |
| 0); |
| } |
| for (int64_t i = 2; i < stream_count; ++i) { |
| const int64_t stream_id = kStreamId + i; |
| ASSERT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId, stream_id))); |
| ASSERT_GT(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId, stream_id)), |
| 0); |
| } |
| |
| InstanceRecycler recycler = make_recycler(txn_kv); |
| ASSERT_EQ(recycler.recycle_partitions(), 0); |
| ASSERT_EQ(recycler.recycle_partitions(), 0); |
| |
| EXPECT_FALSE(key_exists(txn_kv.get(), |
| recycle_partition_key({std::string(kInstanceId), kPartitionId}))); |
| for (int64_t i = 0; i < stream_count; ++i) { |
| const int64_t stream_id = kStreamId + i; |
| EXPECT_FALSE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId, stream_id))); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId, stream_id)), |
| 0); |
| EXPECT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId + 1, stream_id))); |
| EXPECT_GT( |
| range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId + 1, stream_id)), |
| 0); |
| } |
| } |
| |
| TEST(TableStreamRecyclerTest, DropStreamRemovesOffsetsLeftByLegacyPartitionRecycle) { |
| 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; |
| }; |
| |
| auto txn_kv = std::make_shared<MemTxnKv>(); |
| ASSERT_EQ(txn_kv->init(), 0); |
| std::unique_ptr<Transaction> txn; |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_offset(txn.get(), kPartitionId); |
| |
| RecyclePartitionPB recycle_partition; |
| recycle_partition.set_db_id(kBaseDbId); |
| recycle_partition.set_table_id(kBaseTableId); |
| recycle_partition.add_index_id(2001); |
| recycle_partition.set_creation_time(0); |
| recycle_partition.set_expiration(0); |
| recycle_partition.set_state(RecyclePartitionPB::DROPPED); |
| txn->put(recycle_partition_key({std::string(kInstanceId), kPartitionId}), |
| recycle_partition.SerializeAsString()); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| InstanceRecycler partition_recycler = make_recycler(txn_kv); |
| ASSERT_EQ(partition_recycler.recycle_partitions(), 0); |
| ASSERT_FALSE(key_exists(txn_kv.get(), |
| recycle_partition_key({std::string(kInstanceId), kPartitionId}))); |
| ASSERT_TRUE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId))); |
| ASSERT_GT(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId)), 0); |
| |
| ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); |
| put_stream_recycle_index(txn.get(), RecycleIndexPB::DROPPED); |
| ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); |
| |
| InstanceRecycler stream_recycler = make_recycler(txn_kv); |
| ASSERT_EQ(stream_recycler.recycle_indexes(), 0); |
| |
| EXPECT_FALSE( |
| key_exists(txn_kv.get(), recycle_index_key({std::string(kInstanceId), kStreamId}))); |
| EXPECT_FALSE(key_exists(txn_kv.get(), stream_offset_key(kPartitionId))); |
| EXPECT_EQ(range_size(txn_kv.get(), versioned_stream_offset_key(kPartitionId)), 0); |
| } |
| |
| } // namespace |
| } // namespace doris::cloud |
