[GitHub] [incubator-kvrocks] torwig commented on a diff in pull request #1287: Use the RocksDB WriteBatchWithIndex to implement the read-your-own-writes in transaction

["torwig (via GitHub)" <gi...@apache.org>]

torwig commented on code in PR #1287:
URL: https://github.com/apache/incubator-kvrocks/pull/1287#discussion_r1123603242


##########
src/storage/storage.cc:
##########
@@ -500,10 +500,48 @@ Status Storage::GetWALIter(rocksdb::SequenceNumber seq, std::unique_ptr<rocksdb:
 
 rocksdb::SequenceNumber Storage::LatestSeq() { return db_->GetLatestSequenceNumber(); }
 
+rocksdb::Status Storage::Get(const rocksdb::ReadOptions &options, const rocksdb::Slice &key, std::string *value) {
+  return Get(options, db_->DefaultColumnFamily(), key, value);
+}
+
+rocksdb::Status Storage::Get(const rocksdb::ReadOptions &options, rocksdb::ColumnFamilyHandle *column_family,
+                             const rocksdb::Slice &key, std::string *value) {
+  if (is_txn_mode_ && txn_write_batch_->GetWriteBatch()->Count() > 0) {
+    return txn_write_batch_->GetFromBatchAndDB(db_, options, column_family, key, value);
+  }
+  return db_->Get(options, column_family, key, value);
+}
+
+rocksdb::Iterator *Storage::NewIterator(const rocksdb::ReadOptions &options) {
+  return NewIterator(options, db_->DefaultColumnFamily());
+}

Review Comment:
   A little formatting here: add a new line after `}`.



##########
src/storage/storage.cc:
##########
@@ -639,12 +677,41 @@ void Storage::SetIORateLimit(int64_t max_io_mb) {
 
 rocksdb::DB *Storage::GetDB() { return db_; }
 
-Status Storage::WriteToPropagateCF(const std::string &key, const std::string &value) {
-  rocksdb::WriteBatch batch;
+Status Storage::BeginTxn() {
+  if (is_txn_mode_) {
+    return Status{Status::NotOK, "cannot begin a new transaction while already in transaction mode"};
+  }
+  // The EXEC command is exclusive and shouldn't have multi transaction at the same time,
+  // so it's fine to reset the global write batch without any lock.
+  is_txn_mode_ = true;
+  txn_write_batch_ = std::make_unique<rocksdb::WriteBatchWithIndex>();
+  return Status::OK();
+}
 
+Status Storage::CommitTxn() {
+  if (!is_txn_mode_) {
+    return Status{Status::NotOK, "cannot commit while not in transaction mode"};
+  }
+  is_txn_mode_ = false;
+  auto s = Write(write_opts_, txn_write_batch_->GetWriteBatch());

Review Comment:
   Could we reset `txn_write_batch` after the transaction commit? Otherwise, it will be reset only on the next call to `BeginTxn`.



##########
src/storage/storage.cc:
##########
@@ -639,12 +677,41 @@ void Storage::SetIORateLimit(int64_t max_io_mb) {
 
 rocksdb::DB *Storage::GetDB() { return db_; }
 
-Status Storage::WriteToPropagateCF(const std::string &key, const std::string &value) {
-  rocksdb::WriteBatch batch;
+Status Storage::BeginTxn() {
+  if (is_txn_mode_) {
+    return Status{Status::NotOK, "cannot begin a new transaction while already in transaction mode"};
+  }
+  // The EXEC command is exclusive and shouldn't have multi transaction at the same time,
+  // so it's fine to reset the global write batch without any lock.
+  is_txn_mode_ = true;
+  txn_write_batch_ = std::make_unique<rocksdb::WriteBatchWithIndex>();
+  return Status::OK();
+}
 
+Status Storage::CommitTxn() {
+  if (!is_txn_mode_) {
+    return Status{Status::NotOK, "cannot commit while not in transaction mode"};
+  }
+  is_txn_mode_ = false;
+  auto s = Write(write_opts_, txn_write_batch_->GetWriteBatch());
+  if (s.ok()) {
+    return {Status::cOK};

Review Comment:
   You can just return `Status::OK();` here instead of `{Status::cOK}`.



##########
src/storage/storage.cc:
##########
@@ -639,12 +677,41 @@ void Storage::SetIORateLimit(int64_t max_io_mb) {
 
 rocksdb::DB *Storage::GetDB() { return db_; }
 
-Status Storage::WriteToPropagateCF(const std::string &key, const std::string &value) {
-  rocksdb::WriteBatch batch;
+Status Storage::BeginTxn() {
+  if (is_txn_mode_) {
+    return Status{Status::NotOK, "cannot begin a new transaction while already in transaction mode"};
+  }
+  // The EXEC command is exclusive and shouldn't have multi transaction at the same time,
+  // so it's fine to reset the global write batch without any lock.
+  is_txn_mode_ = true;
+  txn_write_batch_ = std::make_unique<rocksdb::WriteBatchWithIndex>();
+  return Status::OK();
+}
 
+Status Storage::CommitTxn() {
+  if (!is_txn_mode_) {
+    return Status{Status::NotOK, "cannot commit while not in transaction mode"};
+  }
+  is_txn_mode_ = false;

Review Comment:
   I think `is_txn_mode_` should be set to `false` only **after** the transaction was committed (**after** the call to `Write`). Otherwise `Write` could be called concurrently. Am I right?



-- 
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