[fix](cloud) Avoid write recycle partition key if operation log is written (apache#53828)

Author: w41ter

cloud/src/meta-service/meta_service_partition.cpp

@@ -604,6 +604,7 @@ void MetaServiceImpl::drop_partition(::google::protobuf::RpcController* controll
         pb.SerializeToString(&to_save_val);
     }
     bool need_commit = false;
+    bool is_versioned_write = is_version_write_enabled(instance_id);
     DropPartitionLogPB drop_partition_log;
     drop_partition_log.set_db_id(request->db_id());
     drop_partition_log.set_table_id(request->table_id());
@@ -615,9 +616,12 @@ void MetaServiceImpl::drop_partition(::google::protobuf::RpcController* controll
         std::string val;
         err = txn->get(key, &val);
         if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { // UNKNOWN
-            LOG_INFO("put recycle partition").tag("key", hex(key));
-            txn->put(key, to_save_val);
-            drop_partition_log.add_partition_ids(part_id);
+            if (is_versioned_write) {
+                drop_partition_log.add_partition_ids(part_id);
+            } else {
+                LOG_INFO("put recycle partition").tag("key", hex(key));
+                txn->put(key, to_save_val);
+            }
             need_commit = true;
             continue;
         }
@@ -661,7 +665,7 @@ void MetaServiceImpl::drop_partition(::google::protobuf::RpcController* controll
 
     if ((drop_partition_log.update_table_version() ||
          drop_partition_log.partition_ids_size() > 0) &&
-        is_version_write_enabled(instance_id)) {
+        is_versioned_write) {
         std::string operation_log_key = versioned::log_key({instance_id});
         std::string operation_log_value;
         OperationLogPB operation_log;
@@ -673,6 +677,10 @@ void MetaServiceImpl::drop_partition(::google::protobuf::RpcController* controll
             return;
         }
         versioned_put(txn.get(), operation_log_key, operation_log_value);
+        LOG(INFO) << "put drop partition operation log"
+                  << " instance_id=" << instance_id << " table_id=" << request->table_id()
+                  << " partition_ids=" << drop_partition_log.partition_ids_size()
+                  << " log_size=" << operation_log_value.size();
     }
 
     err = txn->commit();

cloud/test/meta_service_operation_log_test.cpp

@@ -285,6 +285,14 @@ TEST(MetaServiceOperationLogTest, DropPartitionLog) {
         ASSERT_EQ(new_num_logs, num_logs)
                 << "Expected no new operation logs for drop partition 0, but found "
                 << new_num_logs - num_logs << dump_range(txn_kv.get());
+
+        // The recycle partition key must exists.
+        std::string recycle_key = recycle_partition_key({instance_id, partition_id});
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+        std::string value;
+        TxnErrorCode err = txn->get(recycle_key, &value);
+        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
     }
 
     {
@@ -328,6 +336,16 @@ TEST(MetaServiceOperationLogTest, DropPartitionLog) {
                 << "Expected new operation logs for drop partition 3, but found "
                 << new_num_logs - num_logs << dump_range(txn_kv.get());
         num_logs = new_num_logs;
+
+        // The recycle partition key should not be exists.
+        std::string recycle_key = recycle_partition_key({instance_id, partition_id + 3});
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+        std::string value;
+        TxnErrorCode err = txn->get(recycle_key, &value);
+        ASSERT_EQ(err, TxnErrorCode::TXN_KEY_NOT_FOUND)
+                << "Expected recycle partition key to not exist, but found it: " << hex(recycle_key)
+                << " with value: " << escape_hex(value);
     }
 
     Versionstamp version1;