You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "caipengbo (via GitHub)" <gi...@apache.org> on 2024/01/28 11:19:43 UTC
[PR] Add raw-key-value migration type [kvrocks]
caipengbo opened a new pull request, #2067:
URL: https://github.com/apache/kvrocks/pull/2067
Add unit test later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#issuecomment-1916293965
@caipengbo Sorry for didn't send out the comments. ðŸ˜
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1470783190
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
+ return batch->Send();
+}
+
+Status SlotMigrator::sendSnapshotByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Migrating snapshot of slot " << migrating_slot_ << " by raw key value";
+
+ rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
+ read_options.snapshot = slot_snapshot_;
+ engine::DBIterator iter(storage_, read_options);
+ auto prefix = ComposeSlotKeyPrefix(namespace_, migrating_slot_);
+
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ for (iter.Seek(prefix); iter.Valid() && iter.Key().starts_with(prefix); iter.Next()) {
+ auto redis_type = iter.Type();
+ std::string log_data;
+ if (redis_type == RedisType::kRedisList) {
+ redis::WriteBatchLogData batch_log_data(redis_type, {std::to_string(RedisCommand::kRedisCmdRPush)});
+ log_data = batch_log_data.Encode();
+ } else {
+ redis::WriteBatchLogData batch_log_data(redis_type);
+ log_data = batch_log_data.Encode();
+ }
+ batch_sender.SetPrefixLogData(log_data);
+
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(engine::kMetadataColumnFamilyName), iter.Key(), iter.Value()));
+
+ auto subkey_iter = iter.GetSubKeyIterator();
+ if (!subkey_iter) {
+ continue;
+ }
+
+ for (subkey_iter->Seek(); subkey_iter->Valid(); subkey_iter->Next()) {
+ GET_OR_RET(batch_sender.Put(subkey_iter->ColumnFamilyHandle(), subkey_iter->Key(), subkey_iter->Value()));
+
+ if (redis_type == RedisType::kRedisZSet) {
+ InternalKey internal_key(subkey_iter->Key(), storage_->IsSlotIdEncoded());
+ auto score_key = subkey_iter->Value().ToString();
+ score_key.append(subkey_iter->UserKey().ToString());
+ auto score_key_bytes =
+ InternalKey(iter.Key(), score_key, internal_key.GetVersion(), storage_->IsSlotIdEncoded()).Encode();
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(kColumnFamilyIDZSetScore), score_key_bytes, Slice()));
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate snapshot, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+Status SlotMigrator::syncWALByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Syncing WAL of slot " << migrating_slot_ << " by raw key value";
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ int epoch = 1;
+ uint64_t wal_incremental_seq = 0;
+
+ while (epoch <= kMaxLoopTimes) {
+ if (catchUpIncrementalWAL()) {
+ break;
+ }
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated incremental data, epoch: {}, seq from {} to {}", epoch, wal_begin_seq_,
+ wal_incremental_seq);
+ wal_begin_seq_ = wal_incremental_seq;
+ epoch++;
+ }
+
+ setForbiddenSlot(migrating_slot_);
+
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ if (wal_incremental_seq > wal_begin_seq_) {
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate last incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated last incremental data after set forbidden slot, seq from {} to {}",
+ wal_begin_seq_, wal_incremental_seq);
+ }
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate incremental data, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+bool SlotMigrator::catchUpIncrementalWAL() {
+ uint64_t gap = storage_->GetDB()->GetLatestSequenceNumber() - wal_begin_seq_;
+ if (gap <= seq_gap_limit_) {
+ LOG(INFO) << fmt::format("[migrate] Incremental data sequence gap: {}, less than limit: {}, set forbidden slot: {}",
+ gap, seq_gap_limit_, migrating_slot_.load());
+ return true;
+ }
+ return false;
+}
+
+Status SlotMigrator::migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender *batch_sender) {
+ engine::WALIterator wal_iter(storage_, migrating_slot_);
+ uint64_t start_seq = wal_begin_seq_ + 1;
+ for (wal_iter.Seek(start_seq); wal_iter.Valid(); wal_iter.Next()) {
+ if (wal_iter.NextSequenceNumber() > end_seq + 1) {
+ break;
+ }
+ auto item = wal_iter.Item();
+ switch (item.type) {
+ case engine::WALItem::Type::kTypeLogData: {
+ GET_OR_RET(batch_sender->PutLogData(item.key));
+ break;
+ }
+ case engine::WALItem::Type::kTypePut: {
+ GET_OR_RET(batch_sender->Put(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)),
+ item.key, item.value));
+ break;
+ }
+ case engine::WALItem::Type::kTypeDelete: {
+ GET_OR_RET(
+ batch_sender->Delete(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)), item.key));
+ break;
+ }
+ default:
Review Comment:
The delete range API is only used when flushing the db/all, so it should be a challenge to handle this while migrating. Can you add this switch branch and comment a bit the reason like:
```
// Do nothing in DeleteRange due to it might cross multiple slots. It's only used in
// FLUSHDB/FLUSHALL commands for now and maybe we can disable them while migrating.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1470757455
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
Review Comment:
`SetMigrateBatchSize` will be called in config set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1470760247
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
Review Comment:
oh, I missed this place. Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow to migrate cluster slot by raw write batch [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#issuecomment-1916498561
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2067) **Quality Gate passed**
The SonarCloud Quality Gate passed, but some issues were introduced.
[29 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2067&resolved=false&inNewCodePeriod=true)
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2067&resolved=false&inNewCodePeriod=true)
[63.7% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2067&metric=new_coverage&view=list)
[0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2067&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2067)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1470786096
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
+ return batch->Send();
+}
+
+Status SlotMigrator::sendSnapshotByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Migrating snapshot of slot " << migrating_slot_ << " by raw key value";
+
+ rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
+ read_options.snapshot = slot_snapshot_;
+ engine::DBIterator iter(storage_, read_options);
+ auto prefix = ComposeSlotKeyPrefix(namespace_, migrating_slot_);
+
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ for (iter.Seek(prefix); iter.Valid() && iter.Key().starts_with(prefix); iter.Next()) {
+ auto redis_type = iter.Type();
+ std::string log_data;
+ if (redis_type == RedisType::kRedisList) {
+ redis::WriteBatchLogData batch_log_data(redis_type, {std::to_string(RedisCommand::kRedisCmdRPush)});
+ log_data = batch_log_data.Encode();
+ } else {
+ redis::WriteBatchLogData batch_log_data(redis_type);
+ log_data = batch_log_data.Encode();
+ }
+ batch_sender.SetPrefixLogData(log_data);
+
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(engine::kMetadataColumnFamilyName), iter.Key(), iter.Value()));
+
+ auto subkey_iter = iter.GetSubKeyIterator();
+ if (!subkey_iter) {
+ continue;
+ }
+
+ for (subkey_iter->Seek(); subkey_iter->Valid(); subkey_iter->Next()) {
+ GET_OR_RET(batch_sender.Put(subkey_iter->ColumnFamilyHandle(), subkey_iter->Key(), subkey_iter->Value()));
+
+ if (redis_type == RedisType::kRedisZSet) {
+ InternalKey internal_key(subkey_iter->Key(), storage_->IsSlotIdEncoded());
+ auto score_key = subkey_iter->Value().ToString();
+ score_key.append(subkey_iter->UserKey().ToString());
+ auto score_key_bytes =
+ InternalKey(iter.Key(), score_key, internal_key.GetVersion(), storage_->IsSlotIdEncoded()).Encode();
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(kColumnFamilyIDZSetScore), score_key_bytes, Slice()));
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate snapshot, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+Status SlotMigrator::syncWALByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Syncing WAL of slot " << migrating_slot_ << " by raw key value";
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ int epoch = 1;
+ uint64_t wal_incremental_seq = 0;
+
+ while (epoch <= kMaxLoopTimes) {
+ if (catchUpIncrementalWAL()) {
+ break;
+ }
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated incremental data, epoch: {}, seq from {} to {}", epoch, wal_begin_seq_,
+ wal_incremental_seq);
+ wal_begin_seq_ = wal_incremental_seq;
+ epoch++;
+ }
+
+ setForbiddenSlot(migrating_slot_);
+
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ if (wal_incremental_seq > wal_begin_seq_) {
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate last incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated last incremental data after set forbidden slot, seq from {} to {}",
+ wal_begin_seq_, wal_incremental_seq);
+ }
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate incremental data, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+bool SlotMigrator::catchUpIncrementalWAL() {
+ uint64_t gap = storage_->GetDB()->GetLatestSequenceNumber() - wal_begin_seq_;
+ if (gap <= seq_gap_limit_) {
+ LOG(INFO) << fmt::format("[migrate] Incremental data sequence gap: {}, less than limit: {}, set forbidden slot: {}",
+ gap, seq_gap_limit_, migrating_slot_.load());
+ return true;
+ }
+ return false;
+}
+
+Status SlotMigrator::migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender *batch_sender) {
+ engine::WALIterator wal_iter(storage_, migrating_slot_);
+ uint64_t start_seq = wal_begin_seq_ + 1;
+ for (wal_iter.Seek(start_seq); wal_iter.Valid(); wal_iter.Next()) {
+ if (wal_iter.NextSequenceNumber() > end_seq + 1) {
+ break;
+ }
+ auto item = wal_iter.Item();
+ switch (item.type) {
+ case engine::WALItem::Type::kTypeLogData: {
+ GET_OR_RET(batch_sender->PutLogData(item.key));
+ break;
+ }
+ case engine::WALItem::Type::kTypePut: {
+ GET_OR_RET(batch_sender->Put(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)),
+ item.key, item.value));
+ break;
+ }
+ case engine::WALItem::Type::kTypeDelete: {
+ GET_OR_RET(
+ batch_sender->Delete(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)), item.key));
+ break;
+ }
+ default:
Review Comment:
Sound good :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#issuecomment-1913579257
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2067) **Quality Gate failed**
Failed conditions
[5.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2067&metric=new_coverage&view=list) (required ≥ 60%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2067)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#issuecomment-1916115608
> Just a few questions inline.
I have modified the question you mentioned before. Are there any other questions? @git-hulk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow to migrate cluster slot by raw write batch [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#issuecomment-1916961259
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2067) **Quality Gate passed**
The SonarCloud Quality Gate passed, but some issues were introduced.
[26 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2067&resolved=false&inNewCodePeriod=true)
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2067&resolved=false&inNewCodePeriod=true)
[61.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2067&metric=new_coverage&view=list)
[0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2067&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2067)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "caipengbo (via GitHub)" <gi...@apache.org>.
caipengbo commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1470759694
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
+ return batch->Send();
+}
+
+Status SlotMigrator::sendSnapshotByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Migrating snapshot of slot " << migrating_slot_ << " by raw key value";
+
+ rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
+ read_options.snapshot = slot_snapshot_;
+ engine::DBIterator iter(storage_, read_options);
+ auto prefix = ComposeSlotKeyPrefix(namespace_, migrating_slot_);
+
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ for (iter.Seek(prefix); iter.Valid() && iter.Key().starts_with(prefix); iter.Next()) {
+ auto redis_type = iter.Type();
+ std::string log_data;
+ if (redis_type == RedisType::kRedisList) {
+ redis::WriteBatchLogData batch_log_data(redis_type, {std::to_string(RedisCommand::kRedisCmdRPush)});
+ log_data = batch_log_data.Encode();
+ } else {
+ redis::WriteBatchLogData batch_log_data(redis_type);
+ log_data = batch_log_data.Encode();
+ }
+ batch_sender.SetPrefixLogData(log_data);
+
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(engine::kMetadataColumnFamilyName), iter.Key(), iter.Value()));
+
+ auto subkey_iter = iter.GetSubKeyIterator();
+ if (!subkey_iter) {
+ continue;
+ }
+
+ for (subkey_iter->Seek(); subkey_iter->Valid(); subkey_iter->Next()) {
+ GET_OR_RET(batch_sender.Put(subkey_iter->ColumnFamilyHandle(), subkey_iter->Key(), subkey_iter->Value()));
+
+ if (redis_type == RedisType::kRedisZSet) {
+ InternalKey internal_key(subkey_iter->Key(), storage_->IsSlotIdEncoded());
+ auto score_key = subkey_iter->Value().ToString();
+ score_key.append(subkey_iter->UserKey().ToString());
+ auto score_key_bytes =
+ InternalKey(iter.Key(), score_key, internal_key.GetVersion(), storage_->IsSlotIdEncoded()).Encode();
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(kColumnFamilyIDZSetScore), score_key_bytes, Slice()));
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate snapshot, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+Status SlotMigrator::syncWALByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Syncing WAL of slot " << migrating_slot_ << " by raw key value";
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ int epoch = 1;
+ uint64_t wal_incremental_seq = 0;
+
+ while (epoch <= kMaxLoopTimes) {
+ if (catchUpIncrementalWAL()) {
+ break;
+ }
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated incremental data, epoch: {}, seq from {} to {}", epoch, wal_begin_seq_,
+ wal_incremental_seq);
+ wal_begin_seq_ = wal_incremental_seq;
+ epoch++;
+ }
+
+ setForbiddenSlot(migrating_slot_);
+
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ if (wal_incremental_seq > wal_begin_seq_) {
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate last incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated last incremental data after set forbidden slot, seq from {} to {}",
+ wal_begin_seq_, wal_incremental_seq);
+ }
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate incremental data, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+bool SlotMigrator::catchUpIncrementalWAL() {
+ uint64_t gap = storage_->GetDB()->GetLatestSequenceNumber() - wal_begin_seq_;
+ if (gap <= seq_gap_limit_) {
+ LOG(INFO) << fmt::format("[migrate] Incremental data sequence gap: {}, less than limit: {}, set forbidden slot: {}",
+ gap, seq_gap_limit_, migrating_slot_.load());
+ return true;
+ }
+ return false;
+}
+
+Status SlotMigrator::migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender *batch_sender) {
+ engine::WALIterator wal_iter(storage_, migrating_slot_);
+ uint64_t start_seq = wal_begin_seq_ + 1;
+ for (wal_iter.Seek(start_seq); wal_iter.Valid(); wal_iter.Next()) {
+ if (wal_iter.NextSequenceNumber() > end_seq + 1) {
+ break;
+ }
+ auto item = wal_iter.Item();
+ switch (item.type) {
+ case engine::WALItem::Type::kTypeLogData: {
+ GET_OR_RET(batch_sender->PutLogData(item.key));
+ break;
+ }
+ case engine::WALItem::Type::kTypePut: {
+ GET_OR_RET(batch_sender->Put(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)),
+ item.key, item.value));
+ break;
+ }
+ case engine::WALItem::Type::kTypeDelete: {
+ GET_OR_RET(
+ batch_sender->Delete(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)), item.key));
+ break;
+ }
+ default:
Review Comment:
I don't have a good way to handle delete range because it may overwrite multiple slots. Do you have any good ideas?
I don't see that the redis-command migration handles delete range either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1468850588
##########
src/storage/storage.h:
##########
@@ -177,6 +177,7 @@ class Storage {
bool IsClosing() const { return db_closing_; }
std::string GetName() const { return config_->db_name; }
rocksdb::ColumnFamilyHandle *GetCFHandle(const std::string &name);
+ rocksdb::ColumnFamilyHandle *GetCFHandle(ColumnFamilyID name);
Review Comment:
Can rename it to cfd_id
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,154 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
+ return batch->Send();
+}
+
+Status SlotMigrator::sendSnapshotByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Migrating snapshot of slot " << migrating_slot_ << " by raw key value";
+
+ rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
+ read_options.snapshot = slot_snapshot_;
+ engine::DBIterator iter(storage_, read_options);
+ auto prefix = ComposeSlotKeyPrefix(namespace_, migrating_slot_);
+
+ BatchSender batch_sender(migrating_slot_, *dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ for (iter.Seek(prefix); iter.Valid() && iter.Key().starts_with(prefix); iter.Next()) {
+ auto redis_type = iter.Type();
+ std::string log_data;
+ if (redis_type == RedisType::kRedisList) {
+ redis::WriteBatchLogData batch_log_data(redis_type, {std::to_string(RedisCommand::kRedisCmdRPush)});
+ log_data = batch_log_data.Encode();
+ } else {
+ redis::WriteBatchLogData batch_log_data(redis_type);
+ log_data = batch_log_data.Encode();
+ }
+ batch_sender.SetPrefixLogData(log_data);
+
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(engine::kMetadataColumnFamilyName), iter.Key(), iter.Value()));
+
+ auto subkey_iter = iter.GetSubKeyIterator();
+ if (!subkey_iter) {
+ continue;
+ }
+
+ for (subkey_iter->Seek(); subkey_iter->Valid(); subkey_iter->Next()) {
+ GET_OR_RET(batch_sender.Put(subkey_iter->ColumnFamilyHandle(), subkey_iter->Key(), subkey_iter->Value()));
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate snapshot, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+Status SlotMigrator::syncWALByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Syncing WAL of slot " << migrating_slot_ << " by raw key value";
+ BatchSender batch_sender(migrating_slot_, *dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ int epoch = 1;
+ uint64_t wal_incremental_seq = 0;
+
+ while (epoch <= kMaxLoopTimes) {
+ if (catchUpIncrementalWAL()) {
+ break;
+ }
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated incremental data, epoch: {}, seq from {} to {}", epoch, wal_begin_seq_,
+ wal_incremental_seq);
+ wal_begin_seq_ = wal_incremental_seq;
+ epoch++;
+ }
+
+ setForbiddenSlot(migrating_slot_);
+
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ if (wal_incremental_seq > wal_begin_seq_) {
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate last incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated last incremental data after set forbidden slot, seq from {} to {}",
+ wal_begin_seq_, wal_incremental_seq);
+ }
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate incremental data, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+bool SlotMigrator::catchUpIncrementalWAL() {
+ uint64_t gap = storage_->GetDB()->GetLatestSequenceNumber() - wal_begin_seq_;
+ if (gap <= seq_gap_limit_) {
+ LOG(INFO) << fmt::format("[migrate] Incremental data sequence gap: {}, less than limit: {}, set forbidden slot: {}",
+ gap, seq_gap_limit_, migrating_slot_.load());
+ return true;
+ }
+ return false;
+}
+
+Status SlotMigrator::migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender *batch_sender) {
+ engine::WALIterator wal_iter(storage_, migrating_slot_);
+ uint64_t start_seq = wal_begin_seq_ + 1;
+ for (wal_iter.Seek(start_seq); wal_iter.Valid(); wal_iter.Next()) {
+ if (wal_iter.NextSequenceNumber() > end_seq + 1) {
+ break;
+ }
+ auto item = wal_iter.Item();
+ switch (item.type) {
+ case engine::WALItem::Type::kTypeLogData: {
+ GET_OR_RET(batch_sender->PutLogData(item.key));
+ break;
+ }
+ case engine::WALItem::Type::kTypePut: {
+ GET_OR_RET(batch_sender->Put(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)),
+ item.key, item.value));
+ break;
+ }
+ case engine::WALItem::Type::kTypeDelete: {
+ GET_OR_RET(
+ batch_sender->Delete(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)), item.key));
+ break;
+ }
+ default:
+ break;
+ }
+ if (batch_sender->IsFull()) {
+ GET_OR_RET(sendMigrationBatch(batch_sender));
+ }
+ }
+
+ // send the remaining data
+ return sendMigrationBatch(batch_sender);
+}
Review Comment:
```suggestion
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#issuecomment-1914871914
@caipengbo Great! Overall is quite clear and concise. Just a few questions inline.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add raw-key-value migration type [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #2067:
URL: https://github.com/apache/kvrocks/pull/2067#discussion_r1469722167
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
+ return batch->Send();
+}
+
+Status SlotMigrator::sendSnapshotByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Migrating snapshot of slot " << migrating_slot_ << " by raw key value";
+
+ rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
+ read_options.snapshot = slot_snapshot_;
+ engine::DBIterator iter(storage_, read_options);
+ auto prefix = ComposeSlotKeyPrefix(namespace_, migrating_slot_);
+
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ for (iter.Seek(prefix); iter.Valid() && iter.Key().starts_with(prefix); iter.Next()) {
+ auto redis_type = iter.Type();
+ std::string log_data;
+ if (redis_type == RedisType::kRedisList) {
+ redis::WriteBatchLogData batch_log_data(redis_type, {std::to_string(RedisCommand::kRedisCmdRPush)});
+ log_data = batch_log_data.Encode();
+ } else {
+ redis::WriteBatchLogData batch_log_data(redis_type);
+ log_data = batch_log_data.Encode();
+ }
+ batch_sender.SetPrefixLogData(log_data);
+
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(engine::kMetadataColumnFamilyName), iter.Key(), iter.Value()));
+
+ auto subkey_iter = iter.GetSubKeyIterator();
+ if (!subkey_iter) {
+ continue;
+ }
+
+ for (subkey_iter->Seek(); subkey_iter->Valid(); subkey_iter->Next()) {
+ GET_OR_RET(batch_sender.Put(subkey_iter->ColumnFamilyHandle(), subkey_iter->Key(), subkey_iter->Value()));
+
+ if (redis_type == RedisType::kRedisZSet) {
+ InternalKey internal_key(subkey_iter->Key(), storage_->IsSlotIdEncoded());
+ auto score_key = subkey_iter->Value().ToString();
+ score_key.append(subkey_iter->UserKey().ToString());
+ auto score_key_bytes =
+ InternalKey(iter.Key(), score_key, internal_key.GetVersion(), storage_->IsSlotIdEncoded()).Encode();
+ GET_OR_RET(batch_sender.Put(storage_->GetCFHandle(kColumnFamilyIDZSetScore), score_key_bytes, Slice()));
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ if (batch_sender.IsFull()) {
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+ }
+ }
+
+ GET_OR_RET(sendMigrationBatch(&batch_sender));
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate snapshot, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+Status SlotMigrator::syncWALByRawKV() {
+ uint64_t start_ts = util::GetTimeStampMS();
+ LOG(INFO) << "[migrate] Syncing WAL of slot " << migrating_slot_ << " by raw key value";
+ BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_, migrate_batch_bytes_per_sec_);
+
+ int epoch = 1;
+ uint64_t wal_incremental_seq = 0;
+
+ while (epoch <= kMaxLoopTimes) {
+ if (catchUpIncrementalWAL()) {
+ break;
+ }
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated incremental data, epoch: {}, seq from {} to {}", epoch, wal_begin_seq_,
+ wal_incremental_seq);
+ wal_begin_seq_ = wal_incremental_seq;
+ epoch++;
+ }
+
+ setForbiddenSlot(migrating_slot_);
+
+ wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
+ if (wal_incremental_seq > wal_begin_seq_) {
+ auto s = migrateIncrementalDataByRawKV(wal_incremental_seq, &batch_sender);
+ if (!s.IsOK()) {
+ return {Status::NotOK, fmt::format("migrate last incremental data failed, {}", s.Msg())};
+ }
+ LOG(INFO) << fmt::format("[migrate] Migrated last incremental data after set forbidden slot, seq from {} to {}",
+ wal_begin_seq_, wal_incremental_seq);
+ }
+
+ auto elapsed = util::GetTimeStampMS() - start_ts;
+ LOG(INFO) << fmt::format(
+ "[migrate] Succeed to migrate incremental data, slot: {}, elapsed: {} ms, "
+ "sent: {} bytes, rate: {:.2f} kb/s, batches: {}, entries: {}",
+ migrating_slot_.load(), elapsed, batch_sender.GetSentBytes(), batch_sender.GetRate(start_ts),
+ batch_sender.GetSentBatchesNum(), batch_sender.GetEntriesNum());
+
+ return Status::OK();
+}
+
+bool SlotMigrator::catchUpIncrementalWAL() {
+ uint64_t gap = storage_->GetDB()->GetLatestSequenceNumber() - wal_begin_seq_;
+ if (gap <= seq_gap_limit_) {
+ LOG(INFO) << fmt::format("[migrate] Incremental data sequence gap: {}, less than limit: {}, set forbidden slot: {}",
+ gap, seq_gap_limit_, migrating_slot_.load());
+ return true;
+ }
+ return false;
+}
+
+Status SlotMigrator::migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender *batch_sender) {
+ engine::WALIterator wal_iter(storage_, migrating_slot_);
+ uint64_t start_seq = wal_begin_seq_ + 1;
+ for (wal_iter.Seek(start_seq); wal_iter.Valid(); wal_iter.Next()) {
+ if (wal_iter.NextSequenceNumber() > end_seq + 1) {
+ break;
+ }
+ auto item = wal_iter.Item();
+ switch (item.type) {
+ case engine::WALItem::Type::kTypeLogData: {
+ GET_OR_RET(batch_sender->PutLogData(item.key));
+ break;
+ }
+ case engine::WALItem::Type::kTypePut: {
+ GET_OR_RET(batch_sender->Put(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)),
+ item.key, item.value));
+ break;
+ }
+ case engine::WALItem::Type::kTypeDelete: {
+ GET_OR_RET(
+ batch_sender->Delete(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)), item.key));
+ break;
+ }
+ default:
Review Comment:
Should we care the kTypeDeleteRange write batch?
##########
src/cluster/slot_migrate.cc:
##########
@@ -1130,3 +1148,164 @@ void SlotMigrator::resumeSyncCtx(const Status &migrate_result) {
blocking_context_ = nullptr;
}
}
+
+Status SlotMigrator::sendMigrationBatch(BatchSender *batch) {
+ // user may dynamically change some configs, apply it when send data
+ batch->SetMaxBytes(migrate_batch_size_bytes_);
+ batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
Review Comment:
```suggestion
// Those configurations may be changed dynamically, so apply them here before sending.
batch->SetMaxBytes(migrate_batch_size_bytes_);
batch->SetBytesPerSecond(migrate_batch_bytes_per_sec_);
```
@caipengbo I cannot see those configurations will be changed except for constructing the migrator. Did I miss anything?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org