1495 lines
40 KiB
C++
1495 lines
40 KiB
C++
/**
|
|
* Copyright (c) 2021 OceanBase
|
|
* OceanBase CE is licensed under Mulan PubL v2.
|
|
* You can use this software according to the terms and conditions of the Mulan PubL v2.
|
|
* You may obtain a copy of Mulan PubL v2 at:
|
|
* http://license.coscl.org.cn/MulanPubL-2.0
|
|
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
|
|
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
|
|
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
|
* See the Mulan PubL v2 for more details.
|
|
*/
|
|
|
|
#include <gtest/gtest.h>
|
|
#include <vector>
|
|
#include <algorithm>
|
|
|
|
#include "share/ob_define.h"
|
|
#include "storage/ob_storage_log_type.h"
|
|
#include "storage/transaction/ob_trans_log.h"
|
|
|
|
#include "liboblog/src/ob_log_instance.h"
|
|
#include "liboblog/src/ob_log_fetcher_stream.h"
|
|
#include "liboblog/src/ob_log_fetcher_part_stream.h"
|
|
|
|
#include "ob_log_utils.h" // get_timestamp
|
|
|
|
using namespace oceanbase;
|
|
using namespace common;
|
|
using namespace liboblog;
|
|
using namespace fetcher;
|
|
using namespace transaction;
|
|
using namespace storage;
|
|
using namespace clog;
|
|
|
|
namespace oceanbase
|
|
{
|
|
namespace unittest
|
|
{
|
|
|
|
/*
|
|
* Utils.
|
|
*/
|
|
typedef std::vector<ObAddr> Svrs;
|
|
typedef std::vector<ObPartitionKey> PKeys;
|
|
typedef std::vector<uint64_t> LogIds;
|
|
typedef std::vector<int64_t> Tstamps;
|
|
|
|
/*
|
|
* Mock Rpc Interface 1.
|
|
* It owns N partitions, each has M log entries.
|
|
* It returns L log entries in each fetch_log() call.
|
|
* Log entry contains nothing.
|
|
* Used to test:
|
|
* - add partition into Stream.
|
|
* - update file id & offset.
|
|
* - fetch log.
|
|
* - kick out offline partitions.
|
|
* - discard partitions.
|
|
*/
|
|
class MockRpcInterface1 : public IFetcherRpcInterface
|
|
{
|
|
struct Entry
|
|
{
|
|
file_id_t file_id_;
|
|
offset_t offset_;
|
|
uint64_t log_id_;
|
|
ObPartitionKey pkey_;
|
|
};
|
|
// Use offset_ as index of Entry in EntryVec.
|
|
typedef std::vector<Entry> EntryVec;
|
|
public:
|
|
MockRpcInterface1(const PKeys &pkeys,
|
|
const int64_t log_entry_per_p,
|
|
const int64_t log_entry_per_call)
|
|
{
|
|
log_entry_per_call_ = log_entry_per_call;
|
|
log_entry_per_p_ = log_entry_per_p;
|
|
addr_ = ObAddr(ObAddr::IPV4, "127.0.0.1", 5999);
|
|
// Gen entries.
|
|
int64_t log_entry_cnt = 0;
|
|
for (int64_t log_id = 1; log_id < log_entry_per_p + 1; ++log_id) {
|
|
for (int64_t pidx = 0, cnt = pkeys.size(); pidx < cnt; ++pidx) {
|
|
// Gen entry.
|
|
Entry entry;
|
|
entry.pkey_ = pkeys.at(pidx);
|
|
entry.file_id_ = 1;
|
|
entry.offset_ = static_cast<offset_t>(log_entry_cnt++);
|
|
entry.log_id_ = log_id;
|
|
// Save it.
|
|
entries_.push_back(entry);
|
|
}
|
|
}
|
|
}
|
|
|
|
virtual ~MockRpcInterface1() { }
|
|
|
|
virtual void set_svr(const common::ObAddr& svr) { UNUSED(svr); }
|
|
|
|
virtual const ObAddr& get_svr() const { static ObAddr svr; return svr; }
|
|
|
|
virtual void set_timeout(const int64_t timeout) { UNUSED(timeout); }
|
|
|
|
virtual int req_start_log_id_by_ts(
|
|
const obrpc::ObLogReqStartLogIdByTsRequest& req,
|
|
obrpc::ObLogReqStartLogIdByTsResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_log_id_by_ts_2(const obrpc::ObLogReqStartLogIdByTsRequestWithBreakpoint &req,
|
|
obrpc::ObLogReqStartLogIdByTsResponseWithBreakpoint &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_pos_by_log_id_2(const obrpc::ObLogReqStartPosByLogIdRequestWithBreakpoint &req,
|
|
obrpc::ObLogReqStartPosByLogIdResponseWithBreakpoint &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_pos_by_log_id(
|
|
const obrpc::ObLogReqStartPosByLogIdRequest& req,
|
|
obrpc::ObLogReqStartPosByLogIdResponse& res)
|
|
{
|
|
/*
|
|
* Err Supportted:
|
|
* - OB_SUCCESS
|
|
* - OB_ENTRY_NOT_EXIST: partition exists without any log
|
|
* - OB_ERR_OUT_OF_LOWER_BOUND: log id beyond lower bound
|
|
* - OB_ERR_OUT_OF_UPPER_BOUND: log id beyond upper bound
|
|
*/
|
|
typedef obrpc::ObLogReqStartPosByLogIdRequest::Param Param;
|
|
typedef obrpc::ObLogReqStartPosByLogIdResponse::Result Result;
|
|
for (int64_t idx = 0, cnt = req.get_params().count(); idx < cnt; ++idx) {
|
|
// Locate for a partition.
|
|
const Param ¶m = req.get_params().at(idx);
|
|
const ObPartitionKey &pkey = param.pkey_;
|
|
const uint64_t start_log_id = param.start_log_id_;
|
|
Result result;
|
|
result.reset();
|
|
// Search.
|
|
bool done = false;
|
|
bool partition_exist = false;
|
|
for (int64_t entry_idx = 0, entry_cnt = entries_.size();
|
|
_SUCC_(result.err_) && !done && entry_idx < entry_cnt;
|
|
++entry_idx) {
|
|
const Entry &entry = entries_.at(entry_idx);
|
|
if (entry.pkey_ == pkey) {
|
|
partition_exist = true;
|
|
// Got it.
|
|
if (start_log_id == entry.log_id_) {
|
|
result.err_ = OB_SUCCESS;
|
|
result.file_id_ = 1;
|
|
result.offset_ = entry.offset_;
|
|
done = true;
|
|
}
|
|
// Too small log id.
|
|
else if (start_log_id < entry.log_id_) {
|
|
result.err_ = OB_ERR_OUT_OF_LOWER_BOUND;
|
|
}
|
|
}
|
|
}
|
|
if (!done && _SUCC_(result.err_)) {
|
|
// No log entry.
|
|
if (!partition_exist) {
|
|
result.err_ = OB_ENTRY_NOT_EXIST;
|
|
}
|
|
// Too large log id.
|
|
else {
|
|
result.err_ = OB_ERR_OUT_OF_UPPER_BOUND;
|
|
}
|
|
}
|
|
res.append_result(result);
|
|
}
|
|
_D_("mock rpc 1 req pos", K(req), K(res));
|
|
return OB_SUCCESS;
|
|
}
|
|
|
|
virtual int fetch_log(
|
|
const obrpc::ObLogExternalFetchLogRequest& req,
|
|
obrpc::ObLogExternalFetchLogResponse& res)
|
|
{
|
|
typedef obrpc::ObLogExternalFetchLogRequest::Param Param;
|
|
typedef obrpc::ObLogExternalFetchLogResponse::OfflinePartition OP;
|
|
|
|
// Fetch log.
|
|
const offset_t offset = req.get_offset();
|
|
if (offset < 0) {
|
|
return OB_INVALID_ARGUMENT;
|
|
}
|
|
offset_t ret_offset = offset;
|
|
// Scan.
|
|
for (int64_t idx = static_cast<int64_t>(offset), cnt = entries_.size();
|
|
idx < cnt && res.get_log_num() < log_entry_per_call_;
|
|
++idx) {
|
|
const Entry &entry = entries_.at(idx);
|
|
bool fetch = false;
|
|
for (int64_t pidx = 0, pcnt = req.get_params().count();
|
|
!fetch && pidx < pcnt;
|
|
++pidx) {
|
|
const Param ¶m = req.get_params().at(pidx);
|
|
if (entry.pkey_ == param.pkey_
|
|
&& param.start_log_id_ <= entry.log_id_
|
|
&& entry.log_id_ <= param.last_log_id_) {
|
|
fetch = true;
|
|
}
|
|
}
|
|
if (fetch) {
|
|
ret_offset = static_cast<offset_t>(entry.offset_);
|
|
// Gen header.
|
|
int64_t ts = get_timestamp();
|
|
ObProposalID proposal_id;
|
|
proposal_id.addr_ = addr_;
|
|
proposal_id.ts_ = ts;
|
|
ObLogEntryHeader header;
|
|
header.generate_header(OB_LOG_SUBMIT, entry.pkey_,
|
|
entry.log_id_, mock_load_, mock_load_len_,
|
|
ts, ts, proposal_id, ts, ObVersion(1));
|
|
ObLogEntry log_entry;
|
|
log_entry.generate_entry(header, mock_load_);
|
|
res.append_log(log_entry);
|
|
}
|
|
}
|
|
res.set_file_id_offset(1, ret_offset + 1);
|
|
|
|
// Handle offline partition.
|
|
// Here, if a partition reaches its last log, it is offline.
|
|
for (int64_t idx = 0, cnt = req.get_params().count(); idx < cnt; ++idx) {
|
|
const Param ¶m = req.get_params().at(idx);
|
|
const uint64_t last_log_id = log_entry_per_p_;
|
|
if (last_log_id < param.start_log_id_) {
|
|
OP op;
|
|
op.pkey_ = param.pkey_;
|
|
// op.last_log_id_ = last_log_id;
|
|
op.sync_ts_ = last_log_id;
|
|
res.append_offline_partition(op);
|
|
}
|
|
}
|
|
|
|
_D_("mock rpc 1 fetch log", K(req), K(res));
|
|
|
|
return OB_SUCCESS;
|
|
}
|
|
|
|
virtual int req_heartbeat_info(
|
|
const obrpc::ObLogReqHeartbeatInfoRequest& req,
|
|
obrpc::ObLogReqHeartbeatInfoResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_leader_heartbeat(
|
|
const obrpc::ObLogLeaderHeartbeatReq &req,
|
|
obrpc::ObLogLeaderHeartbeatResp &res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int open_stream(const obrpc::ObLogOpenStreamReq &req,
|
|
obrpc::ObLogOpenStreamResp &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int fetch_stream_log(const obrpc::ObLogStreamFetchLogReq &req,
|
|
obrpc::ObLogStreamFetchLogResp &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_svr_feedback(const ReqLogSvrFeedback &feedback)
|
|
{
|
|
UNUSED(feedback);
|
|
return OB_SUCCESS;
|
|
}
|
|
private:
|
|
ObAddr addr_;
|
|
int64_t log_entry_per_call_;
|
|
int64_t log_entry_per_p_;
|
|
EntryVec entries_;
|
|
static const int64_t mock_load_len_ = 8;
|
|
char mock_load_[mock_load_len_];
|
|
};
|
|
|
|
/*
|
|
* Factory.
|
|
*/
|
|
class MockRpcInterface1Factory : public IFetcherRpcInterfaceFactory
|
|
{
|
|
public:
|
|
MockRpcInterface1Factory(const PKeys &pkeys,
|
|
const int64_t log_entry_per_p,
|
|
const int64_t log_entry_per_call)
|
|
: pkeys_(pkeys),
|
|
log_entry_per_p_(log_entry_per_p),
|
|
log_entry_per_call_(log_entry_per_call)
|
|
{ }
|
|
virtual int new_fetcher_rpc_interface(IFetcherRpcInterface*& rpc)
|
|
{
|
|
rpc = new MockRpcInterface1(pkeys_, log_entry_per_p_, log_entry_per_call_);
|
|
return OB_SUCCESS;
|
|
}
|
|
virtual int delete_fetcher_rpc_interface(IFetcherRpcInterface* rpc)
|
|
{
|
|
delete rpc;
|
|
return OB_SUCCESS;
|
|
}
|
|
private:
|
|
PKeys pkeys_;
|
|
int64_t log_entry_per_p_;
|
|
int64_t log_entry_per_call_;
|
|
};
|
|
|
|
/*
|
|
* Mock Rpc Interface 2.
|
|
* It owns N servers, each hold some partitions, one of them is
|
|
* the leader. When request start log id, a preseted value is returned.
|
|
* Used to test:
|
|
* - fetch partition
|
|
* - locate start log id
|
|
* - activate partition stream
|
|
* - discard partition stream
|
|
*/
|
|
class MockRpcInterface2 : public IFetcherRpcInterface
|
|
{
|
|
struct Partition
|
|
{
|
|
ObPartitionKey pkey_;
|
|
uint64_t start_log_id_;
|
|
bool is_leader_;
|
|
};
|
|
struct Svr
|
|
{
|
|
ObAddr svr_;
|
|
std::vector<Partition> partitions_;
|
|
bool operator==(const Svr &other) const
|
|
{
|
|
return svr_ == other.svr_;
|
|
}
|
|
bool operator<(const Svr &other) const
|
|
{
|
|
return svr_ < other.svr_;
|
|
}
|
|
};
|
|
public:
|
|
/*
|
|
* Set static result set. The first partition in svrs is the leader.
|
|
*/
|
|
static void add_partition(const ObPartitionKey &pkey,
|
|
uint64_t start_log_id,
|
|
std::vector<ObAddr> svrs)
|
|
{
|
|
EXPECT_NE(0, svrs.size());
|
|
Partition pt = { pkey, start_log_id, false };
|
|
for (int64_t idx = 0, cnt = svrs.size();
|
|
idx < cnt;
|
|
++idx) {
|
|
pt.is_leader_ = (0 == idx);
|
|
Svr target;
|
|
target.svr_ = svrs.at(idx);
|
|
std::vector<Svr>::iterator itor =
|
|
std::find(svrs_.begin(), svrs_.end(), target);
|
|
if (svrs_.end() == itor) {
|
|
target.partitions_.push_back(pt);
|
|
svrs_.push_back(target);
|
|
std::sort(svrs_.begin(), svrs_.end());
|
|
}
|
|
else {
|
|
(*itor).partitions_.push_back(pt);
|
|
}
|
|
}
|
|
}
|
|
/*
|
|
* Clear static result set.
|
|
*/
|
|
static void clear_result_set()
|
|
{
|
|
svrs_.clear();
|
|
}
|
|
public:
|
|
virtual void set_svr(const common::ObAddr& svr) { svr_ = svr; }
|
|
|
|
virtual const ObAddr& get_svr() const { return svr_; }
|
|
|
|
virtual void set_timeout(const int64_t timeout) { UNUSED(timeout); }
|
|
|
|
virtual int req_start_log_id_by_ts(
|
|
const obrpc::ObLogReqStartLogIdByTsRequest& req,
|
|
obrpc::ObLogReqStartLogIdByTsResponse& res)
|
|
{
|
|
bool svr_exist = false;
|
|
for (int64_t idx = 0, cnt = req.get_params().count(); idx < cnt; ++idx) {
|
|
const ObPartitionKey &pkey = req.get_params().at(idx).pkey_;
|
|
bool done = false;
|
|
for (int64_t svr_idx = 0, svr_cnt = svrs_.size();
|
|
svr_idx < svr_cnt;
|
|
++svr_idx) {
|
|
// Simulating sending rpc to svr.
|
|
if (svr_ == svrs_.at(svr_idx).svr_) {
|
|
svr_exist = true;
|
|
const Svr &svr = svrs_.at(svr_idx);
|
|
for (int64_t pidx = 0, pcnt = svr.partitions_.size();
|
|
pidx < pcnt;
|
|
++pidx) {
|
|
const Partition &p = svr.partitions_.at(pidx);
|
|
if (pkey == p.pkey_) {
|
|
done = true;
|
|
typedef obrpc::ObLogReqStartLogIdByTsResponse::Result Result;
|
|
Result result = { OB_SUCCESS, p.start_log_id_, false};
|
|
res.append_result(result);
|
|
}
|
|
}
|
|
if (!done) {
|
|
res.set_err(OB_PARTITION_NOT_EXIST);
|
|
}
|
|
}
|
|
} // End for.
|
|
}
|
|
|
|
_D_("mock rpc req start log id", K(req), K(res));
|
|
|
|
return (svr_exist) ? OB_SUCCESS : OB_TIMEOUT;
|
|
}
|
|
|
|
virtual int req_start_log_id_by_ts_2(const obrpc::ObLogReqStartLogIdByTsRequestWithBreakpoint &req,
|
|
obrpc::ObLogReqStartLogIdByTsResponseWithBreakpoint &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_pos_by_log_id_2(const obrpc::ObLogReqStartPosByLogIdRequestWithBreakpoint &req,
|
|
obrpc::ObLogReqStartPosByLogIdResponseWithBreakpoint &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_pos_by_log_id(
|
|
const obrpc::ObLogReqStartPosByLogIdRequest& req,
|
|
obrpc::ObLogReqStartPosByLogIdResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
// Timeout.
|
|
_D_("mock rpc req pos by log id", K(req), K(res));
|
|
return OB_TIMEOUT;
|
|
}
|
|
|
|
virtual int fetch_log(
|
|
const obrpc::ObLogExternalFetchLogRequest& req,
|
|
obrpc::ObLogExternalFetchLogResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
// Timeout.
|
|
_D_("mock rpc req pos by log id", K(req), K(res));
|
|
return OB_TIMEOUT;
|
|
}
|
|
|
|
virtual int req_heartbeat_info(
|
|
const obrpc::ObLogReqHeartbeatInfoRequest& req,
|
|
obrpc::ObLogReqHeartbeatInfoResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_leader_heartbeat(
|
|
const obrpc::ObLogLeaderHeartbeatReq &req,
|
|
obrpc::ObLogLeaderHeartbeatResp &res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int open_stream(const obrpc::ObLogOpenStreamReq &req,
|
|
obrpc::ObLogOpenStreamResp &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int fetch_stream_log(const obrpc::ObLogStreamFetchLogReq &req,
|
|
obrpc::ObLogStreamFetchLogResp &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
virtual int get_log_svr(const ObPartitionKey& pkey, const uint64_t log_id,
|
|
ObSvrs& svrs, int& leader_cnt)
|
|
{
|
|
UNUSED(log_id);
|
|
leader_cnt = 0;
|
|
for (int64_t svr_idx = 0, svr_cnt = svrs_.size();
|
|
svr_idx < svr_cnt;
|
|
++svr_idx) {
|
|
Svr &svr = svrs_.at(svr_idx);
|
|
for (int64_t pidx = 0, pcnt = svr.partitions_.size();
|
|
pidx < pcnt;
|
|
++pidx) {
|
|
const Partition &p = svr.partitions_.at(pidx);
|
|
if (pkey == p.pkey_) {
|
|
svrs.push_back(svr.svr_);
|
|
if (p.is_leader_) {
|
|
std::swap(svrs.at(leader_cnt), svrs.at(svrs.count() - 1));
|
|
leader_cnt += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
_D_("mock rpc req log servers", K(pkey), K(svrs), K(leader_cnt));
|
|
return OB_SUCCESS;
|
|
}
|
|
virtual int req_svr_feedback(const ReqLogSvrFeedback &feedback)
|
|
{
|
|
UNUSED(feedback);
|
|
return OB_SUCCESS;
|
|
}
|
|
private:
|
|
// Target svr.
|
|
ObAddr svr_;
|
|
// Data set.
|
|
static std::vector<Svr> svrs_;
|
|
};
|
|
// Static data set. So all instances could access it.
|
|
std::vector<MockRpcInterface2::Svr> MockRpcInterface2::svrs_;
|
|
|
|
/*
|
|
* Factory.
|
|
*/
|
|
class MockRpcInterface2Factory : public IFetcherRpcInterfaceFactory
|
|
{
|
|
public:
|
|
virtual int new_fetcher_rpc_interface(IFetcherRpcInterface*& rpc)
|
|
{
|
|
rpc = new MockRpcInterface2();
|
|
return OB_SUCCESS;
|
|
}
|
|
virtual int delete_fetcher_rpc_interface(IFetcherRpcInterface* rpc)
|
|
{
|
|
delete rpc;
|
|
return OB_SUCCESS;
|
|
}
|
|
};
|
|
|
|
/*
|
|
* Mock Rpc Interface 3.
|
|
* It owns some servers and partitions, can return
|
|
* svr addr and heartbeat timestamps.
|
|
* Notice: user set <svr, partition key, log id, tstamp> tuples
|
|
* as results.
|
|
* Used to test:
|
|
* - Heartbeat facilities.
|
|
*/
|
|
class MockRpcInterface3 : public IFetcherRpcInterface
|
|
{
|
|
struct Entry
|
|
{
|
|
ObAddr svr_;
|
|
ObPartitionKey pkey_;
|
|
uint64_t log_id_;
|
|
int64_t tstamp_;
|
|
bool operator<(const Entry &other)
|
|
{
|
|
return log_id_ < other.log_id_;
|
|
}
|
|
};
|
|
typedef std::vector<Entry> EntryVec;
|
|
public:
|
|
static void clear_result()
|
|
{
|
|
entry_vec_.clear();
|
|
}
|
|
static void add_result(const ObAddr &svr, const ObPartitionKey &pkey,
|
|
const uint64_t log_id, const int64_t tstamp)
|
|
{
|
|
Entry entry = { svr, pkey, log_id, tstamp };
|
|
entry_vec_.push_back(entry);
|
|
}
|
|
public:
|
|
virtual void set_svr(const common::ObAddr& svr) { UNUSED(svr); }
|
|
|
|
virtual const ObAddr& get_svr() const { static ObAddr svr; return svr; }
|
|
|
|
virtual void set_timeout(const int64_t timeout) { UNUSED(timeout); }
|
|
|
|
virtual int req_start_log_id_by_ts(
|
|
const obrpc::ObLogReqStartLogIdByTsRequest& req,
|
|
obrpc::ObLogReqStartLogIdByTsResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_log_id_by_ts_2(const obrpc::ObLogReqStartLogIdByTsRequestWithBreakpoint &req,
|
|
obrpc::ObLogReqStartLogIdByTsResponseWithBreakpoint &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_pos_by_log_id_2(const obrpc::ObLogReqStartPosByLogIdRequestWithBreakpoint &req,
|
|
obrpc::ObLogReqStartPosByLogIdResponseWithBreakpoint &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_start_pos_by_log_id(
|
|
const obrpc::ObLogReqStartPosByLogIdRequest& req,
|
|
obrpc::ObLogReqStartPosByLogIdResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int fetch_log(
|
|
const obrpc::ObLogExternalFetchLogRequest& req,
|
|
obrpc::ObLogExternalFetchLogResponse& res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int req_heartbeat_info(
|
|
const obrpc::ObLogReqHeartbeatInfoRequest& req,
|
|
obrpc::ObLogReqHeartbeatInfoResponse& res)
|
|
{
|
|
typedef obrpc::ObLogReqHeartbeatInfoRequest::Param Param;
|
|
typedef obrpc::ObLogReqHeartbeatInfoResponse::Result Result;
|
|
// Itor params.
|
|
for (int64_t idx = 0, cnt = req.get_params().count(); idx < cnt; ++idx) {
|
|
const Param ¶m = req.get_params().at(idx);
|
|
// Find result.
|
|
bool done = false;
|
|
for (int64_t idx2 = 0, cnt2 = entry_vec_.size();
|
|
!done && idx2 < cnt2;
|
|
++idx2) {
|
|
const Entry &entry = entry_vec_[idx2];
|
|
if (param.pkey_ == entry.pkey_
|
|
&& param.log_id_ == entry.log_id_) {
|
|
done = true;
|
|
Result result;
|
|
result.err_ = OB_SUCCESS;
|
|
result.tstamp_ = entry.tstamp_;
|
|
res.append_result(result);
|
|
}
|
|
}
|
|
if (!done) {
|
|
Result result;
|
|
result.err_ = OB_NEED_RETRY;
|
|
res.append_result(result);
|
|
}
|
|
}
|
|
|
|
_D_("mock rpc: req heartbeat", K(req), K(res));
|
|
return OB_SUCCESS;
|
|
}
|
|
|
|
virtual int req_leader_heartbeat(
|
|
const obrpc::ObLogLeaderHeartbeatReq &req,
|
|
obrpc::ObLogLeaderHeartbeatResp &res)
|
|
{
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int open_stream(const obrpc::ObLogOpenStreamReq &req,
|
|
obrpc::ObLogOpenStreamResp &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int fetch_stream_log(const obrpc::ObLogStreamFetchLogReq &req,
|
|
obrpc::ObLogStreamFetchLogResp &res) {
|
|
UNUSED(req);
|
|
UNUSED(res);
|
|
return OB_NOT_IMPLEMENT;
|
|
}
|
|
|
|
virtual int get_log_svr(const ObPartitionKey& pkey, const uint64_t log_id,
|
|
ObSvrs& svrs, int& leader_cnt)
|
|
{
|
|
// Todo. In this version, only one result is enough, log_id is not used.
|
|
UNUSED(log_id);
|
|
UNUSED(leader_cnt);
|
|
|
|
for (int64_t idx = 0, cnt = entry_vec_.size(); idx < cnt; ++idx) {
|
|
const Entry &entry = entry_vec_[idx];
|
|
if (pkey == entry.pkey_) {
|
|
svrs.push_back(entry.svr_);
|
|
break;
|
|
}
|
|
}
|
|
return OB_SUCCESS;
|
|
}
|
|
|
|
virtual int req_svr_feedback(const ReqLogSvrFeedback &feedback)
|
|
{
|
|
UNUSED(feedback);
|
|
return OB_SUCCESS;
|
|
}
|
|
private:
|
|
static EntryVec entry_vec_;
|
|
};
|
|
|
|
MockRpcInterface3::EntryVec MockRpcInterface3::entry_vec_;
|
|
|
|
/*
|
|
* Factory.
|
|
*/
|
|
class MockRpcInterface3Factory : public IFetcherRpcInterfaceFactory
|
|
{
|
|
public:
|
|
virtual int new_fetcher_rpc_interface(IFetcherRpcInterface*& rpc)
|
|
{
|
|
rpc = new MockRpcInterface3();
|
|
return OB_SUCCESS;
|
|
}
|
|
virtual int delete_fetcher_rpc_interface(IFetcherRpcInterface* rpc)
|
|
{
|
|
delete rpc;
|
|
return OB_SUCCESS;
|
|
}
|
|
};
|
|
|
|
/*
|
|
* TransLog Generator 1.
|
|
* Generate single partition transaction logs.
|
|
* Support get trans logs in CORRECT order.
|
|
* Use:
|
|
* - Call next_trans(), specify trans params.
|
|
* - Get logs in correct order: redo, redo, ..., prepare, commit/abort.
|
|
*/
|
|
struct TransParam1
|
|
{
|
|
// Params used in trans log.
|
|
ObPartitionKey pkey_;
|
|
ObTransID trans_id_;
|
|
ObAddr scheduler_;
|
|
ObPartitionKey coordinator_;
|
|
ObPartitionArray participants_;
|
|
ObStartTransParam trans_param_;
|
|
};
|
|
|
|
class TransLogGenerator1
|
|
{
|
|
public:
|
|
TransLogGenerator1()
|
|
: param_(),
|
|
redo_(),
|
|
prepare_(),
|
|
commit_(),
|
|
abort_()
|
|
{ }
|
|
virtual ~TransLogGenerator1() { }
|
|
public:
|
|
void next_trans(const TransParam1 ¶m)
|
|
{
|
|
param_ = param;
|
|
}
|
|
const ObTransRedoLog& next_redo(const uint64_t log_id)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
uint64_t tenant_id = 100;
|
|
const uint64_t cluster_id = 1000;
|
|
redo_.reset();
|
|
ObVersion active_memstore_version(1);
|
|
err = redo_.init(OB_LOG_TRANS_REDO, param_.pkey_, param_.trans_id_,
|
|
tenant_id, log_id, param_.scheduler_, param_.coordinator_,
|
|
param_.participants_, param_.trans_param_, cluster_id, active_memstore_version);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
ObTransMutator &mutator = redo_.get_mutator();
|
|
if (NULL == mutator.get_mutator_buf()) {
|
|
mutator.init(true);
|
|
}
|
|
const char *data = "fly";
|
|
char *buf = static_cast<char*>(mutator.alloc(strlen(data)));
|
|
strcpy(buf, data);
|
|
return redo_;
|
|
}
|
|
const ObTransPrepareLog& next_prepare(const ObRedoLogIdArray &all_redos)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
uint64_t tenant_id = 100;
|
|
const uint64_t cluster_id = 1000;
|
|
prepare_.reset();
|
|
ObVersion active_memstore_version(1);
|
|
err = prepare_.init(OB_LOG_TRANS_PREPARE, param_.pkey_, param_.trans_id_,
|
|
tenant_id, param_.scheduler_, param_.coordinator_,
|
|
param_.participants_, param_.trans_param_,
|
|
OB_SUCCESS, all_redos, 0, cluster_id, active_memstore_version);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
return prepare_;
|
|
}
|
|
const ObTransCommitLog& next_commit(const uint64_t prepare_log_id)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
const uint64_t cluster_id = 1000;
|
|
ObPartitionLogInfo ptl_id(param_.pkey_, prepare_log_id, get_timestamp());
|
|
PartitionLogInfoArray ptl_ids;
|
|
if (OB_INVALID_ID == prepare_log_id) {
|
|
// Pass. For prepare-commit trans log.
|
|
}
|
|
else {
|
|
err = ptl_ids.push_back(ptl_id);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
}
|
|
commit_.reset();
|
|
err = commit_.init(OB_LOG_TRANS_COMMIT, param_.pkey_, param_.trans_id_,
|
|
ptl_ids, 1, 0, cluster_id);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
return commit_;
|
|
}
|
|
const ObTransAbortLog& next_abort()
|
|
{
|
|
int err = OB_SUCCESS;
|
|
const uint64_t cluster_id = 1000;
|
|
PartitionLogInfoArray array;
|
|
abort_.reset();
|
|
err = abort_.init(OB_LOG_TRANS_ABORT, param_.pkey_, param_.trans_id_, array, cluster_id);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
return abort_;
|
|
}
|
|
private:
|
|
TransParam1 param_;
|
|
ObTransRedoLog redo_;
|
|
ObTransPrepareLog prepare_;
|
|
ObTransCommitLog commit_;
|
|
ObTransAbortLog abort_;
|
|
};
|
|
|
|
/*
|
|
* Transaction Log Entry Generator 1.
|
|
* Generate log entries of transactions.
|
|
*/
|
|
class TransLogEntryGenerator1
|
|
{
|
|
public:
|
|
TransLogEntryGenerator1(const ObPartitionKey &pkey)
|
|
: pkey_(pkey),
|
|
log_id_(0),
|
|
remain_log_cnt_(0),
|
|
commit_(false),
|
|
param_(),
|
|
trans_log_gen_(),
|
|
prepare_id_(0),
|
|
redos_(),
|
|
data_len_(0)
|
|
{
|
|
ObAddr addr(ObAddr::IPV4, "127.0.0.1", 5566);
|
|
param_.pkey_ = pkey_;
|
|
param_.trans_id_ = ObTransID(addr);
|
|
param_.scheduler_ = addr;
|
|
param_.coordinator_ = pkey_;
|
|
int err = param_.participants_.push_back(pkey_);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
param_.trans_param_.set_access_mode(ObTransAccessMode::READ_WRITE);
|
|
param_.trans_param_.set_isolation(ObTransIsolation::READ_COMMITED);
|
|
param_.trans_param_.set_type(ObTransType::TRANS_NORMAL);
|
|
|
|
buf_ = new char[buf_len_];
|
|
EXPECT_TRUE(NULL != buf_);
|
|
}
|
|
virtual ~TransLogEntryGenerator1()
|
|
{
|
|
delete[] buf_;
|
|
}
|
|
// Generate normal trans.
|
|
// Start a new trans.
|
|
void next_trans(const int64_t redo_cnt, bool commit)
|
|
{
|
|
remain_log_cnt_ = 2 + redo_cnt;
|
|
commit_ = commit;
|
|
redos_.reset();
|
|
trans_log_gen_.next_trans(param_);
|
|
}
|
|
// Get next log entry.
|
|
int next_log_entry(ObLogEntry &log_entry)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (2 < remain_log_cnt_) {
|
|
next_redo_(log_entry);
|
|
// Store redo id.
|
|
int err = redos_.push_back(log_id_);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else if (2 == remain_log_cnt_) {
|
|
next_prepare_(log_entry);
|
|
prepare_id_ = log_id_;
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else if (1 == remain_log_cnt_ && commit_) {
|
|
next_commit_(log_entry);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else if (1 == remain_log_cnt_ && !commit_) {
|
|
next_abort_(log_entry);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else {
|
|
ret = OB_ITER_END;
|
|
}
|
|
return ret;
|
|
}
|
|
// Generate: redo, redo, redo, prepare-commit.
|
|
int next_log_entry_2(ObLogEntry &log_entry)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (2 < remain_log_cnt_) {
|
|
next_redo_(log_entry);
|
|
// Store redo id.
|
|
int err = redos_.push_back(log_id_);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else if (2 == remain_log_cnt_ && commit_) {
|
|
next_prepare_with_commit(log_entry);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 2;
|
|
}
|
|
else if (2 == remain_log_cnt_ && !commit_) {
|
|
next_prepare_(log_entry);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else if (1 == remain_log_cnt_ && !commit_) {
|
|
next_abort_(log_entry);
|
|
log_id_ += 1;
|
|
remain_log_cnt_ -= 1;
|
|
}
|
|
else {
|
|
ret = OB_ITER_END;
|
|
}
|
|
return ret;
|
|
}
|
|
private:
|
|
void next_redo_(ObLogEntry &log_entry)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
// Gen trans log.
|
|
const ObTransRedoLog &redo = trans_log_gen_.next_redo(log_id_);
|
|
int64_t pos = 0;
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, OB_LOG_TRANS_REDO);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, 0);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = redo.serialize(buf_, buf_len_, pos);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
data_len_ = pos;
|
|
// Gen entry header.
|
|
ObLogEntryHeader header;
|
|
header.generate_header(OB_LOG_SUBMIT, pkey_, log_id_, buf_,
|
|
data_len_, get_timestamp(), get_timestamp(),
|
|
ObProposalID(), get_timestamp(), ObVersion(0));
|
|
// Gen log entry.
|
|
log_entry.generate_entry(header, buf_);
|
|
}
|
|
void next_prepare_(ObLogEntry &log_entry)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
// Gen trans log.
|
|
const ObTransPrepareLog &prepare= trans_log_gen_.next_prepare(redos_);
|
|
int64_t pos = 0;
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, OB_LOG_TRANS_PREPARE);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, 0);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = prepare.serialize(buf_, buf_len_, pos);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
data_len_ = pos;
|
|
// Gen entry header.
|
|
ObLogEntryHeader header;
|
|
header.generate_header(OB_LOG_SUBMIT, pkey_, log_id_, buf_,
|
|
data_len_, get_timestamp(), get_timestamp(),
|
|
ObProposalID(), get_timestamp(), ObVersion(0));
|
|
// Gen log entry.
|
|
log_entry.generate_entry(header, buf_);
|
|
}
|
|
void next_commit_(ObLogEntry &log_entry)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
// Gen trans log.
|
|
const ObTransCommitLog &commit = trans_log_gen_.next_commit(prepare_id_);
|
|
int64_t pos = 0;
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, OB_LOG_TRANS_COMMIT);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, 0);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = commit.serialize(buf_, buf_len_, pos);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
data_len_ = pos;
|
|
// Gen entry header.
|
|
ObLogEntryHeader header;
|
|
header.generate_header(OB_LOG_SUBMIT, pkey_, log_id_, buf_,
|
|
data_len_, get_timestamp(), get_timestamp(),
|
|
ObProposalID(), get_timestamp(), ObVersion(0));
|
|
// Gen log entry.
|
|
log_entry.generate_entry(header, buf_);
|
|
}
|
|
void next_abort_(ObLogEntry &log_entry)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
// Gen trans log.
|
|
const ObTransAbortLog &abort = trans_log_gen_.next_abort();
|
|
int64_t pos = 0;
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, OB_LOG_TRANS_ABORT);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, 0);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = abort.serialize(buf_, buf_len_, pos);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
data_len_ = pos;
|
|
// Gen entry header.
|
|
ObLogEntryHeader header;
|
|
header.generate_header(OB_LOG_SUBMIT, pkey_, log_id_, buf_,
|
|
data_len_, get_timestamp(), get_timestamp(),
|
|
ObProposalID(), get_timestamp(), ObVersion(0));
|
|
// Gen log entry.
|
|
log_entry.generate_entry(header, buf_);
|
|
}
|
|
void next_prepare_with_commit(ObLogEntry &log_entry)
|
|
{
|
|
int err = OB_SUCCESS;
|
|
// Gen trans log.
|
|
const ObTransPrepareLog &prepare= trans_log_gen_.next_prepare(redos_);
|
|
const ObTransCommitLog &commit = trans_log_gen_.next_commit(OB_INVALID_ID);
|
|
int64_t pos = 0;
|
|
err = serialization::encode_i64(buf_, buf_len_,
|
|
pos, OB_LOG_TRANS_PREPARE_WITH_COMMIT);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = serialization::encode_i64(buf_, buf_len_, pos, 0);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = prepare.serialize(buf_, buf_len_, pos);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
err = commit.serialize(buf_, buf_len_, pos);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
data_len_ = pos;
|
|
// Gen entry header.
|
|
ObLogEntryHeader header;
|
|
header.generate_header(OB_LOG_SUBMIT, pkey_, log_id_, buf_,
|
|
data_len_, get_timestamp(), get_timestamp(),
|
|
ObProposalID(), get_timestamp(), ObVersion(0));
|
|
// Gen log entry.
|
|
log_entry.generate_entry(header, buf_);
|
|
}
|
|
private:
|
|
// Params.
|
|
ObPartitionKey pkey_;
|
|
uint64_t log_id_;
|
|
int64_t remain_log_cnt_;
|
|
bool commit_;
|
|
// Gen.
|
|
TransParam1 param_;
|
|
TransLogGenerator1 trans_log_gen_;
|
|
uint64_t prepare_id_;
|
|
ObRedoLogIdArray redos_;
|
|
// Buf.
|
|
int64_t data_len_;
|
|
static const int64_t buf_len_ = 2 * _M_;
|
|
char *buf_;
|
|
};
|
|
|
|
/*
|
|
* Mock Parser 1.
|
|
* Read Task, revert it immediately, and count Task number.
|
|
*/
|
|
class MockParser1 : public IObLogParser
|
|
{
|
|
public:
|
|
MockParser1() : trans_cnt_(0) { }
|
|
virtual ~MockParser1() { }
|
|
virtual int start() { return OB_SUCCESS; }
|
|
virtual void stop() { }
|
|
virtual void mark_stop_flag() { }
|
|
virtual int push(PartTransTask* task, const int64_t timeout)
|
|
{
|
|
UNUSED(timeout);
|
|
if (NULL != task && task->is_normal_trans()) {
|
|
task->revert();
|
|
trans_cnt_ += 1;
|
|
// Debug.
|
|
// _I_(">>> push parser", "req", task->get_seq());
|
|
}
|
|
return OB_SUCCESS;
|
|
}
|
|
int64_t get_trans_cnt() const { return trans_cnt_; }
|
|
private:
|
|
int64_t trans_cnt_;
|
|
};
|
|
|
|
/*
|
|
* Mock Fetcher Error Handler.
|
|
*/
|
|
class MockFetcherErrHandler1 : public IErrHandler
|
|
{
|
|
public:
|
|
virtual ~MockFetcherErrHandler1() { }
|
|
public:
|
|
virtual void handle_err(int err_no, const char* fmt, ...)
|
|
{
|
|
UNUSED(err_no);
|
|
va_list ap;
|
|
va_start(ap, fmt);
|
|
__E__(fmt, ap);
|
|
va_end(ap);
|
|
abort();
|
|
}
|
|
};
|
|
|
|
/*
|
|
* Mock Liboblog Error Handler.
|
|
*/
|
|
class MockLiboblogErrHandler1 : public IObLogErrHandler
|
|
{
|
|
public:
|
|
virtual void handle_error(int err_no, const char* fmt, ...)
|
|
{
|
|
UNUSED(err_no);
|
|
va_list ap;
|
|
va_start(ap, fmt);
|
|
__E__(fmt, ap);
|
|
va_end(ap);
|
|
}
|
|
};
|
|
|
|
/*
|
|
* Mock SvrProvider.
|
|
* User set svrs into it.
|
|
*/
|
|
class MockSvrProvider1 : public sqlclient::ObMySQLServerProvider
|
|
{
|
|
public:
|
|
virtual ~MockSvrProvider1() { }
|
|
void add_svr(const ObAddr &svr) { svrs_.push_back(svr); }
|
|
public:
|
|
virtual int get_cluster_list(common::ObIArray<int64_t> &cluster_list)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (svrs_.size() > 0) {
|
|
if (OB_FAIL(cluster_list.push_back(common::OB_INVALID_ID))) {
|
|
LOG_WARN("fail to push back cluster_id", K(ret));
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
virtual int get_server(const int64_t cluster_id, const int64_t svr_idx, ObAddr& server)
|
|
{
|
|
UNUSED(cluster_id);
|
|
int ret = OB_SUCCESS;
|
|
if (0 <= svr_idx && svr_idx < static_cast<int64_t>(svrs_.size())) {
|
|
server = svrs_[svr_idx];
|
|
}
|
|
else {
|
|
ret = OB_ERR_UNEXPECTED;
|
|
}
|
|
return ret;
|
|
}
|
|
virtual int64_t get_cluster_count() const
|
|
{
|
|
return svrs_.size() > 0 ? 1 : 0;
|
|
}
|
|
virtual int64_t get_server_count(const int64_t cluster_id) const
|
|
{
|
|
UNUSED(cluster_id)
|
|
return static_cast<int64_t>(svrs_.size());
|
|
}
|
|
virtual int refresh_server_list() { return OB_SUCCESS; }
|
|
private:
|
|
std::vector<ObAddr> svrs_;
|
|
};
|
|
|
|
/*
|
|
* Test Dataset Generator.
|
|
*/
|
|
|
|
/*
|
|
* Svr Config.
|
|
* Set svr address and mysql port.
|
|
*/
|
|
struct SvrCfg
|
|
{
|
|
// Svr.
|
|
const char *svr_addr_;
|
|
int internal_port_;
|
|
// Mysql.
|
|
int mysql_port_;
|
|
const char *mysql_user_;
|
|
const char *mysql_password_;
|
|
const char *mysql_db_;
|
|
int64_t mysql_timeout_;
|
|
};
|
|
|
|
/*
|
|
* Configuration for mysql connector.
|
|
*/
|
|
inline ConnectorConfig prepare_cfg_1(const SvrCfg &svr_cfg)
|
|
{
|
|
ConnectorConfig cfg;
|
|
cfg.mysql_addr_ = svr_cfg.svr_addr_;
|
|
cfg.mysql_port_ = svr_cfg.mysql_port_;
|
|
cfg.mysql_user_ = svr_cfg.mysql_user_;
|
|
cfg.mysql_password_ = svr_cfg.mysql_password_;
|
|
cfg.mysql_db_ = svr_cfg.mysql_db_;
|
|
cfg.mysql_timeout_ = svr_cfg.mysql_timeout_;
|
|
return cfg;
|
|
}
|
|
|
|
/*
|
|
* Build table names.
|
|
*/
|
|
inline const char** prepare_table_name_1()
|
|
{
|
|
static const char* tnames[] = {
|
|
"table1",
|
|
"table2",
|
|
"table3",
|
|
"table4",
|
|
"table5",
|
|
"table6",
|
|
"table7",
|
|
"table8",
|
|
"table9",
|
|
"table10",
|
|
"table11",
|
|
"table12",
|
|
"table13",
|
|
"table14",
|
|
"table15",
|
|
"table16"
|
|
};
|
|
return tnames;
|
|
}
|
|
|
|
/*
|
|
* Build table schema.
|
|
*/
|
|
inline const char* prepare_table_schema_1()
|
|
{
|
|
return "c1 int primary key";
|
|
}
|
|
|
|
/*
|
|
* Create table.
|
|
*/
|
|
class CreateTable : public MySQLQueryBase
|
|
{
|
|
public:
|
|
CreateTable(const char *tname, const char *schema)
|
|
{
|
|
snprintf(buf_, 512, "create table %s(%s)", tname, schema);
|
|
sql_ = buf_;
|
|
sql_len_ = strlen(sql_);
|
|
}
|
|
private:
|
|
char buf_[512];
|
|
};
|
|
|
|
/*
|
|
* Drop table.
|
|
*/
|
|
class DropTable : public MySQLQueryBase
|
|
{
|
|
public:
|
|
DropTable(const char *tname)
|
|
{
|
|
snprintf(buf_, 512, "drop table if exists %s", tname);
|
|
sql_ = buf_;
|
|
sql_len_ = strlen(sql_);
|
|
}
|
|
private:
|
|
char buf_[512];
|
|
};
|
|
|
|
/*
|
|
* Get table id.
|
|
*/
|
|
class GetTableId : public MySQLQueryBase
|
|
{
|
|
public:
|
|
GetTableId(const char *tname)
|
|
{
|
|
snprintf(buf_, 512, "select table_id "
|
|
"from __all_table where table_name='%s'", tname);
|
|
sql_ = buf_;
|
|
sql_len_ = strlen(sql_);
|
|
}
|
|
int get_tid(uint64_t &tid)
|
|
{
|
|
int ret = common::OB_SUCCESS;
|
|
while (common::OB_SUCCESS == (ret = next_row())) {
|
|
uint64_t table_id = 0;
|
|
if (OB_SUCC(ret)) {
|
|
if (common::OB_SUCCESS != (ret = get_uint(0, table_id))) {
|
|
OBLOG_LOG(WARN, "err get uint", K(ret));
|
|
}
|
|
}
|
|
tid = table_id;
|
|
}
|
|
ret = (common::OB_ITER_END == ret) ? common::OB_SUCCESS : ret;
|
|
return ret;
|
|
}
|
|
private:
|
|
char buf_[512];
|
|
};
|
|
|
|
/*
|
|
* Get partition key by table id from system table.
|
|
*/
|
|
class GetPartitionKey : public MySQLQueryBase
|
|
{
|
|
public:
|
|
GetPartitionKey(const uint64_t tid)
|
|
{
|
|
snprintf(buf_, 512, "select table_id, partition_id, partition_cnt "
|
|
"from __all_meta_table where table_id=%lu", tid);
|
|
sql_ = buf_;
|
|
sql_len_ = strlen(sql_);
|
|
}
|
|
int get_pkeys(ObArray<ObPartitionKey> &pkeys)
|
|
{
|
|
int ret = common::OB_SUCCESS;
|
|
while (common::OB_SUCCESS == (ret = next_row())) {
|
|
uint64_t table_id = 0;
|
|
int32_t partition_id = 0;
|
|
int32_t partition_cnt = 0;
|
|
if (OB_SUCC(ret)) {
|
|
if (common::OB_SUCCESS != (ret = get_uint(0, table_id))) {
|
|
OBLOG_LOG(WARN, "err get uint", K(ret));
|
|
}
|
|
}
|
|
if (OB_SUCC(ret)) {
|
|
int64_t val = 0;
|
|
if (common::OB_SUCCESS != (ret = get_int(1, val))) {
|
|
OBLOG_LOG(WARN, "err get int", K(ret));
|
|
} else {
|
|
partition_id = static_cast<int32_t>(val);
|
|
}
|
|
}
|
|
if (OB_SUCC(ret)) {
|
|
int64_t val = 0;
|
|
if (common::OB_SUCCESS != (ret = get_int(2, val))) {
|
|
OBLOG_LOG(WARN, "err get int", K(ret));
|
|
} else {
|
|
partition_cnt = static_cast<int32_t>(val);
|
|
}
|
|
}
|
|
ObPartitionKey pkey;
|
|
pkey.init(table_id, partition_id, partition_cnt);
|
|
pkeys.push_back(pkey);
|
|
}
|
|
ret = (common::OB_ITER_END == ret) ? common::OB_SUCCESS : ret;
|
|
return ret;
|
|
}
|
|
private:
|
|
char buf_[512];
|
|
};
|
|
|
|
/*
|
|
* Create table and return their partition keys.
|
|
*/
|
|
inline void prepare_table_1(const SvrCfg& svr_cfg,
|
|
const char** tnames,
|
|
const int64_t tcnt,
|
|
const char* schema,
|
|
ObArray<ObPartitionKey>& pkeys)
|
|
{
|
|
ObLogMySQLConnector conn;
|
|
ConnectorConfig cfg = prepare_cfg_1(svr_cfg);
|
|
|
|
int ret = conn.init(cfg);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
|
|
// Prepare tables.
|
|
|
|
for (int64_t idx = 0; idx < tcnt; ++idx) {
|
|
// Drop.
|
|
DropTable drop_table(tnames[idx]);
|
|
ret = conn.exec(drop_table);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
// Create.
|
|
CreateTable create_table(tnames[idx], schema);
|
|
ret = conn.exec(create_table);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
// Get tid.
|
|
GetTableId get_tid(tnames[idx]);
|
|
ret = conn.query(get_tid);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
uint64_t tid = OB_INVALID_ID;
|
|
ret = get_tid.get_tid(tid);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
// Get pkeys.
|
|
GetPartitionKey get_pkey(tid);
|
|
ret = conn.query(get_pkey);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
ret = get_pkey.get_pkeys(pkeys);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
}
|
|
|
|
ret = conn.destroy();
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
}
|
|
|
|
/*
|
|
* Data generator.
|
|
* Insert a bunch of data in server.
|
|
* For schema 1.
|
|
*/
|
|
class DataGenerator1 : public Runnable
|
|
{
|
|
class Inserter : public MySQLQueryBase
|
|
{
|
|
public:
|
|
void set_data(const char *tname, const int64_t data)
|
|
{
|
|
reuse();
|
|
snprintf(buf_, 512, "insert into %s (c1) values (%ld)", tname, data);
|
|
sql_ = buf_;
|
|
sql_len_ = strlen(sql_);
|
|
}
|
|
private:
|
|
char buf_[512];
|
|
};
|
|
public:
|
|
DataGenerator1(const ConnectorConfig &cfg) :
|
|
conn_(), cfg_(cfg), tname_(NULL), start_(0), end_(0)
|
|
{
|
|
int err = conn_.init(cfg_);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
}
|
|
~DataGenerator1()
|
|
{
|
|
conn_.destroy();
|
|
}
|
|
void insert(const char *tname, const int64_t start, const int64_t end)
|
|
{
|
|
tname_ = tname;
|
|
start_ = start;
|
|
end_ = end;
|
|
create();
|
|
}
|
|
private:
|
|
int routine()
|
|
{
|
|
Inserter inserter;
|
|
int err = OB_SUCCESS;
|
|
for (int64_t cur = start_;
|
|
OB_SUCCESS == err && cur < end_;
|
|
cur++) {
|
|
inserter.set_data(tname_, cur);
|
|
err = conn_.exec(inserter);
|
|
EXPECT_EQ(OB_SUCCESS, err);
|
|
}
|
|
return OB_SUCCESS;
|
|
}
|
|
private:
|
|
ObLogMySQLConnector conn_;
|
|
ConnectorConfig cfg_;
|
|
const char *tname_;
|
|
int64_t start_;
|
|
int64_t end_;
|
|
};
|
|
|
|
/*
|
|
* End of Test Dataset Generator.
|
|
*/
|
|
|
|
}
|
|
}
|