hcq/oceanbase
hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[FEAT MERGE] load local files

Browse Source
This commit is contained in:
hnwyllmm 2024-02-07 20:25:14 +00:00 committed by ob-robot
parent c7fe4c3f69
commit acd0ec6efd
45 changed files with 1671 additions and 503 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
deps/oblib/src/rpc/CMakeLists.txt vendored
View File

@ -48,6 +48,7 @@ ob_set_subtarget(oblib_rpc obmysql_packet
obmysql/packet/ompk_field.cpp
obmysql/packet/ompk_handshake.cpp
obmysql/packet/ompk_handshake_response.cpp
obmysql/packet/ompk_local_infile.cpp
obmysql/packet/ompk_ok.cpp
obmysql/packet/ompk_piece.cpp
obmysql/packet/ompk_prepare.cpp

73
deps/oblib/src/rpc/ob_sql_request_operator.h vendored
View File

@ -19,9 +19,17 @@
namespace oceanbase
{
namespace obmysql
{
class ObICSMemPool;
class ObSqlSockProcessor;
} // namespace obmysql
namespace rpc
{
class ObRequest;
class ObPacket;
struct ObSqlSockDesc
{
ObSqlSockDesc(): type_(0), sock_desc_(NULL) {}
@ -52,6 +60,49 @@ public:
virtual common::ObAddr get_peer(const ObRequest* req) = 0;
virtual void disconnect_sql_conn(ObRequest* req) = 0;
virtual void finish_sql_request(ObRequest* req) = 0;
virtual int create_read_handle(ObRequest* req, void *& read_handle) { return OB_NOT_SUPPORTED; }
/**
* read a packet from socket channel
*
* @param req the `req` received before
* @param mem_pool The memory manager to hold the packet returned
* @param read_handle Read data from socket by this handle
* @param[out] pkt The packet received or null if no message
* @return OB_SUCCESS success to read a packet
* Other failed
*
* read_packet only supported by nio now.
* In NIO, the message reader will cache the message data and the packet in
* it's memory buffer when receiving a request. The request would be hold in
* the memory until the reuqest done.
* We can not read new packet during the processing of request if using the
* normal method. We need to create a new buffer to hold new packet and keep
* the request live.
* We should `create_read_handle` before we read new packet and
* `release_read_handle` after things done.
* You cannot read new packet until the last packet received is dead and you
* can use `release_packet` to kill it.
*
* The whole flow likes below:
* 1. get a request from client. (`load data local infile`)
* 2. call `create_read_handle`
* 3. `read_packet`
* 4. `release_packet`
* 5. goto 3 if we should read more packets
* 6. call `release_read_handle`
* 7. finish_request. request (`load data local`) will be released.
*/
virtual int read_packet(ObRequest* req,
obmysql::ObICSMemPool& mem_pool,
void* read_handle,
obmysql::ObSqlSockProcessor& sock_processor,
ObPacket*& pkt)
{
return OB_NOT_SUPPORTED;
}
virtual int release_packet(ObRequest* req, void* read_handle, ObPacket* pkt) { return OB_NOT_SUPPORTED; }
virtual int release_read_handle(ObRequest* req, void* read_handle) { return OB_NOT_SUPPORTED; }
virtual int write_response(ObRequest* req, const char* buf, int64_t sz) = 0;
virtual int async_write_response(ObRequest* req, const char* buf, int64_t sz) = 0;
virtual void get_sock_desc(ObRequest* req, ObSqlSockDesc& desc) = 0;
@ -84,6 +135,21 @@ public:
void finish_sql_request(ObRequest* req) {
return get_operator(req).finish_sql_request(req);
}
int create_read_handle(ObRequest* req, void *& read_handle) {
return get_operator(req).create_read_handle(req, read_handle);
}
int release_read_handle(ObRequest* req, void * read_handle) {
return get_operator(req).release_read_handle(req, read_handle);
}
int read_packet(ObRequest* req, obmysql::ObICSMemPool& mem_pool, void *read_handle, ObPacket*& pkt) {
return get_operator(req).read_packet(req, mem_pool, read_handle, *sock_processor_, pkt);
}
int release_packet(ObRequest* req, void* read_handle, ObPacket* pkt) {
return get_operator(req).release_packet(req, read_handle, pkt);
}
int write_response(ObRequest* req, const char* buf, int64_t sz) {
return get_operator(req).write_response(req, buf, sz);
}
@ -102,9 +168,16 @@ public:
void set_sql_session_to_sock_desc(ObRequest* req, void* sess) {
return get_operator(req).set_sql_session_to_sock_desc(req, sess);
}
void set_sql_sock_processor(obmysql::ObSqlSockProcessor& sock_processor) {
sock_processor_ = &sock_processor;
}
private:
ObISqlRequestOperator& get_operator(const ObRequest* req);
ObISqlRequestOperator& get_operator(const ObSqlSockDesc& desc);
private:
obmysql::ObSqlSockProcessor *sock_processor_;
};
extern ObSqlRequestOperator global_sql_req_operator;
#define SQL_REQ_OP (oceanbase::rpc::global_sql_req_operator)

8
deps/oblib/src/rpc/obmysql/ob_2_0_protocol_processor.cpp vendored
View File

@ -410,7 +410,9 @@ inline int Ob20ProtocolProcessor::process_ob20_packet(ObProto20PktContext& conte
&& pkt20->get_extra_info().exist_extra_info()) {
char* tmp_buffer = NULL;
int64_t total_len = pkt20->get_extra_info().get_total_len();
if (OB_ISNULL(tmp_buffer = reinterpret_cast<char *>(context.arena_.alloc(total_len)))) {
if (total_len <= 0) {
// empty extra info, do nothing
} else if (OB_ISNULL(tmp_buffer = reinterpret_cast<char *>(context.arena_.alloc(total_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_ERROR("no memory available", "alloc_size", total_len, K(ret));
} else if (OB_FAIL(context.extra_info_.assign(pkt20->get_extra_info(), tmp_buffer, total_len))) {
@ -440,7 +442,9 @@ inline int Ob20ProtocolProcessor::process_ob20_packet(ObProto20PktContext& conte
input_packet->set_proxy_switch_route(pkt20->get_flags().proxy_switch_route());
const int64_t t_len = context.extra_info_.get_total_len();
char *t_buffer = NULL;
if (OB_ISNULL(t_buffer = reinterpret_cast<char *>(pool.alloc(t_len)))) {
if (t_len <= 0) {
// empty extra info, do nothing
} else if (OB_ISNULL(t_buffer = reinterpret_cast<char *>(pool.alloc(t_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_ERROR("no memory available", "alloc_size", t_len, K(ret));
} else if (OB_FAIL(input_packet->extra_info_.assign(context.extra_info_, t_buffer, t_len))) {

3
deps/oblib/src/rpc/obmysql/ob_2_0_protocol_utils.cpp vendored
View File

@ -556,7 +556,8 @@ inline int ObProto20Utils::fill_proto20_header(ObProtoEncodeParam &param) {
Ob20ProtocolFlags flag;
flag.st_flags_.OB_EXTRA_INFO_EXIST = param.proto20_context_->has_extra_info_;
flag.st_flags_.OB_IS_LAST_PACKET = (ObProto20Utils::is_the_last_packet(param) ? 1 : 0);
flag.st_flags_.OB_IS_LAST_PACKET = (ObProto20Utils::is_the_last_packet(param)
|| OB_UNLIKELY(param.proto20_context_->is_filename_packet_) ? 1 : 0);
flag.st_flags_.OB_IS_NEW_EXTRA_INFO = proto20_context.is_new_extra_info_;
flag.st_flags_.OB_TXN_FREE_ROUTE = proto20_context.txn_free_route_ ? 1 : 0;
uint16_t reserved = 0;

9
deps/oblib/src/rpc/obmysql/ob_2_0_protocol_utils.h vendored
View File

@ -67,7 +67,8 @@ public:
tailer_len_(0), next_step_(START_TO_FILL_STEP),
is_proto20_used_(false), is_checksum_off_(false),
has_extra_info_(false), is_new_extra_info_(false),
curr_proto20_packet_start_pos_(0), txn_free_route_(false) {}
curr_proto20_packet_start_pos_(0), txn_free_route_(false),
is_filename_packet_(false) {}
~ObProto20Context() {}
inline void reset() { MEMSET(this, 0, sizeof(ObProto20Context)); }
@ -84,7 +85,8 @@ public:
K_(has_extra_info),
K_(is_new_extra_info),
K_(txn_free_route),
K_(curr_proto20_packet_start_pos));
K_(curr_proto20_packet_start_pos),
K_(is_filename_packet));
public:
uint8_t comp_seq_;
@ -99,6 +101,9 @@ public:
bool is_new_extra_info_;
int64_t curr_proto20_packet_start_pos_;
bool txn_free_route_;
// used in local local.
// We should set `is_filename_packet_` when sending PKT_FILENAME packet.
bool is_filename_packet_;
private:
DISALLOW_COPY_AND_ASSIGN(ObProto20Context);
};

3
deps/oblib/src/rpc/obmysql/ob_mysql_compress_protocol_processor.cpp vendored
View File

@ -198,7 +198,8 @@ inline int ObMysqlCompressProtocolProcessor::process_compressed_packet(
pkt_rec_wrapper.record_recieve_comp_packet(*iraw_pkt, *raw_pkt);
}
}
context.reset();
// reset all context except the compress packet sequence
context.reuse();
}
}
}

19
deps/oblib/src/rpc/obmysql/ob_mysql_packet.h vendored
View File

@ -105,7 +105,8 @@ enum class ObMySQLPacketType
PKT_PREPARE, // 9 -> prepare packet;
PKT_RESHEAD, // 10 -> result header packet
PKT_PREXEC, // 11 -> prepare execute packet;
PKT_END // 12 -> end of packet type
PKT_FILENAME, // 12 -> send file name to client(load local infile)
PKT_END // 13 -> end of packet type
};
union ObServerStatusFlags
@ -161,6 +162,7 @@ union ObProxyCapabilityFlags
bool is_weak_stale_feedback() const { return 1 == cap_flags_.OB_CAP_PROXY_WEAK_STALE_FEEDBACK; }
bool is_flt_show_trace_support() const { return 1 == cap_flags_.OB_CAP_PROXY_FULL_LINK_TRACING_EXT
&& is_ob_protocol_v2_support(); }
bool is_load_local_support() const { return 1 == cap_flags_.OB_CAP_LOCAL_FILES; }
bool is_client_sessid_support() const { return 1 == cap_flags_.OB_CAP_PROXY_CLIENT_SESSION_ID; }
uint64_t capability_;
@ -522,6 +524,7 @@ public:
is_weak_read_(false),
txn_free_route_(false),
proxy_switch_route_(false),
consume_size_(0),
extra_info_()
{}
@ -551,6 +554,9 @@ public:
const common::ObString &get_trace_info() const { return extra_info_.trace_info_; }
virtual int64_t get_serialize_size() const;
void set_consume_size(int64_t consume_size) { consume_size_ = consume_size; }
int64_t get_consume_size() const { return consume_size_; }
virtual void reset() {
ObMySQLPacket::reset();
cmd_ = COM_MAX_NUM;
@ -559,6 +565,7 @@ public:
txn_free_route_ = false;
proxy_switch_route_ = false;
extra_info_.reset();
consume_size_ = 0;
}
virtual void assign(const ObMySQLRawPacket &other)
@ -570,10 +577,12 @@ public:
txn_free_route_ = other.txn_free_route_;
extra_info_ = other.extra_info_;
proxy_switch_route_ = other.proxy_switch_route_;
consume_size_ = other.consume_size_;
}
TO_STRING_KV("header", hdr_, "can_reroute", can_reroute_pkt_, "weak_read", is_weak_read_,
"txn_free_route_", txn_free_route_, "proxy_switch_route", proxy_switch_route_);
"txn_free_route_", txn_free_route_, "proxy_switch_route", proxy_switch_route_,
"consume_size", consume_size_);
protected:
virtual int serialize(char*, const int64_t, int64_t&) const;
@ -585,6 +594,12 @@ private:
bool is_weak_read_;
bool txn_free_route_;
bool proxy_switch_route_;
// In load local scenario, we should tell the NIO to consume specific size data.
// The size is a packet size in usually. But the mysql packet size if not equal
// to the packet that we received if we use ob20 or compress protocol.
// NOTE: one ob20 or compress packet has only one mysql packet in request message.
int64_t consume_size_;
public:
Ob20ExtraInfo extra_info_;
};

6
deps/oblib/src/rpc/obmysql/ob_mysql_request_utils.h vendored
View File

@ -123,6 +123,12 @@ public:
is_multi_pkt_ = false;
}
void reuse()
{
// keep the last_pkt_seq_ here
is_multi_pkt_ = false;
}
TO_STRING_KV(K_(last_pkt_seq),
K_(is_multi_pkt));

68
deps/oblib/src/rpc/obmysql/ob_packet_record.cpp vendored
View File

@ -27,7 +27,7 @@ bool __attribute__((weak)) enable_proto_dia()
}
namespace obmysql
{
static const char* pkt_type_name[13] =
static const char* pkt_type_name[14] =
{
"INVALID_PKT",
"PKT_MYSQL", // 1 -> mysql packet;
@ -41,9 +41,70 @@ static const char* pkt_type_name[13] =
"PKT_PREPARE", // 9 -> prepare packet;
"PKT_RESHEAD", // 10 -> result header packet
"PKT_PREXEC", // 11 -> prepare execute packet;
"PKT_END" // 12 -> end of packet type
"PKT_FILENAME", // 12 -> file name packet(load local infile)
"PKT_END" // 13 -> end of packet type
};
static const char *get_receive_pkt_type_name(const ObpMysqHeader &header)
{
#define PKT_TYPE_NAME(command) case command: { name = #command; } break
const char *name = "UNKNOWN_PKT";
if (header.is_file_content_) {
name = "FILE_CONTENT";
} else {
uint8_t type = header.type_;
switch (type) {
PKT_TYPE_NAME(COM_SLEEP);
PKT_TYPE_NAME(COM_QUIT);
PKT_TYPE_NAME(COM_INIT_DB);
PKT_TYPE_NAME(COM_QUERY);
PKT_TYPE_NAME(COM_FIELD_LIST);
PKT_TYPE_NAME(COM_CREATE_DB);
PKT_TYPE_NAME(COM_DROP_DB);
PKT_TYPE_NAME(COM_REFRESH);
PKT_TYPE_NAME(COM_SHUTDOWN);
PKT_TYPE_NAME(COM_STATISTICS);
PKT_TYPE_NAME(COM_PROCESS_INFO);
PKT_TYPE_NAME(COM_CONNECT);
PKT_TYPE_NAME(COM_PROCESS_KILL);
PKT_TYPE_NAME(COM_DEBUG);
PKT_TYPE_NAME(COM_PING);
PKT_TYPE_NAME(COM_TIME);
PKT_TYPE_NAME(COM_DELAYED_INSERT);
PKT_TYPE_NAME(COM_CHANGE_USER);
PKT_TYPE_NAME(COM_BINLOG_DUMP);
PKT_TYPE_NAME(COM_TABLE_DUMP);
PKT_TYPE_NAME(COM_CONNECT_OUT);
PKT_TYPE_NAME(COM_REGISTER_SLAVE);
PKT_TYPE_NAME(COM_STMT_PREPARE);
PKT_TYPE_NAME(COM_STMT_EXECUTE);
PKT_TYPE_NAME(COM_STMT_SEND_LONG_DATA);
PKT_TYPE_NAME(COM_STMT_CLOSE);
PKT_TYPE_NAME(COM_STMT_RESET);
PKT_TYPE_NAME(COM_SET_OPTION);
PKT_TYPE_NAME(COM_STMT_FETCH);
PKT_TYPE_NAME(COM_DAEMON);
PKT_TYPE_NAME(COM_BINLOG_DUMP_GTID);
PKT_TYPE_NAME(COM_RESET_CONNECTION);
PKT_TYPE_NAME(COM_END);
PKT_TYPE_NAME(COM_DELETE_SESSION);
PKT_TYPE_NAME(COM_HANDSHAKE);
PKT_TYPE_NAME(COM_LOGIN);
PKT_TYPE_NAME(COM_STMT_PREXECUTE);
PKT_TYPE_NAME(COM_STMT_SEND_PIECE_DATA);
PKT_TYPE_NAME(COM_STMT_GET_PIECE_DATA);
default: {
} break;
}
}
#undef PKT_TYPE_NAME
return name;
}
int64_t ObPacketRecord::to_string(char *buf, const int64_t buf_len) const
{
int64_t pos = 0;
@ -91,7 +152,8 @@ int64_t ObPacketRecord::to_string(char *buf, const int64_t buf_len) const
"rec_", obp_mysql_header_.rec_, "seq_", obp_mysql_header_.seq_);
J_OBJ_END();
J_COMMA();
J_KV(K(obp_mysql_header_.type_), K(obp_mysql_header_.is_send_));
J_KV("pkt_name", get_receive_pkt_type_name(obp_mysql_header_),
K(obp_mysql_header_.type_), K(obp_mysql_header_.is_send_));
}
J_OBJ_END();
return pos;

38
deps/oblib/src/rpc/obmysql/ob_packet_record.h vendored
View File

@ -72,11 +72,13 @@ struct ObpMysqHeader {
uint8_t seq_;
uint8_t type_;
uint8_t com_seq_; // compress head sequence
uint8_t is_send_;
uint8_t is_send_:1;
uint8_t is_file_content_:1;
ObpMysqHeader() {
rec_ = 0;
seq_ = 0;
mysql_header_.len_ = 0;
is_file_content_ = 0;
}
~ObpMysqHeader() {}
@ -168,6 +170,10 @@ public:
obp_mysql_header_.type_ = static_cast<uint8_t>(type);
}
inline void set_file_content() __restrict__ {
obp_mysql_header_.is_file_content_ = 1;
}
int64_t to_string(char *buf, const int64_t buf_len) const;
Obp20Header obp20_header_; // 16 byte
ObpMysqHeader obp_mysql_header_; // 16 byte
@ -182,6 +188,7 @@ class ObPacketRecordWrapper {
cur_pkt_pos_ = 0;
last_type_ = obmysql::ObMySQLPacketType::INVALID_PKT;
enable_proto_dia_ = false;
receiving_file_contents_ = false;
}
~ObPacketRecordWrapper() {}
void init() {
@ -189,6 +196,7 @@ class ObPacketRecordWrapper {
cur_pkt_pos_ = 0;
last_type_ = obmysql::ObMySQLPacketType::INVALID_PKT;
enable_proto_dia_ = observer::enable_proto_dia();
receiving_file_contents_ = false;
}
int64_t to_string(char *buf, int64_t buf_len) const;
@ -214,6 +222,13 @@ class ObPacketRecordWrapper {
rec.record_recieve_obp20_packet(obp20_pkt);
rec.record_recieve_mysql_packet(pkt);
cur_pkt_pos_++;
if (OB_UNLIKELY(receiving_file_contents_)) {
pkt_rec_[idx].set_file_content();
if (0 == pkt.get_clen()) {
receiving_file_contents_ = false;
}
}
}
// for 20 protocol end
@ -236,6 +251,13 @@ class ObPacketRecordWrapper {
rec.record_recieve_comp_packet(com_pkt);
rec.record_recieve_mysql_packet(pkt);
cur_pkt_pos_++;
if (OB_UNLIKELY(receiving_file_contents_)) {
pkt_rec_[idx].set_file_content();
if (0 == pkt.get_clen()) {
receiving_file_contents_ = false;
}
}
}
// for compress protocol end
@ -251,12 +273,23 @@ class ObPacketRecordWrapper {
int64_t idx = (cur_pkt_pos_-1) % ObPacketRecordWrapper::REC_BUF_SIZE;
pkt_rec_[idx].record_send_mysql_packet(pkt, len);
last_type_ = pkt.get_mysql_packet_type();
if (OB_UNLIKELY(pkt.get_mysql_packet_type() == ObMySQLPacketType::PKT_FILENAME)) {
receiving_file_contents_ = true;
}
}
inline void record_recieve_mysql_packet(obmysql::ObMySQLRawPacket &__restrict__ pkt) __restrict__
{
int64_t idx = cur_pkt_pos_ % ObPacketRecordWrapper::REC_BUF_SIZE;
pkt_rec_[idx].record_recieve_mysql_packet(pkt);
cur_pkt_pos_++;
if (OB_UNLIKELY(receiving_file_contents_)) {
pkt_rec_[idx].set_file_content();
if (0 == pkt.get_clen()) {
receiving_file_contents_ = false;
}
}
}
inline void record_recieve_mysql_pkt_fragment(int32_t recive) __restrict__
{
@ -274,6 +307,9 @@ class ObPacketRecordWrapper {
uint32_t cur_pkt_pos_;
obmysql::ObMySQLPacketType last_type_;
bool enable_proto_dia_;
// in load local infile, we will receive some file content packets and there is no `cmd` in the packet.
// so we use a flag to mark the context.
bool receiving_file_contents_;
};

30
deps/oblib/src/rpc/obmysql/ob_poc_sql_request_operator.cpp vendored
View File

@ -13,6 +13,7 @@
#include "rpc/obmysql/ob_poc_sql_request_operator.h"
#include "rpc/obmysql/ob_sql_sock_session.h"
#include "rpc/obrpc/ob_rpc_opts.h"
#include "rpc/obmysql/ob_sql_sock_processor.h"
namespace oceanbase
{
@ -72,6 +73,35 @@ void ObPocSqlRequestOperator::finish_sql_request(ObRequest* req)
obmysql::request_finish_callback();
}
int ObPocSqlRequestOperator::create_read_handle(ObRequest* req, void*& read_handle)
{
ObSqlSockSession* sess = (ObSqlSockSession*)req->get_server_handle_context();
return sess->create_read_handle(read_handle);
}
int ObPocSqlRequestOperator::release_read_handle(ObRequest* req, void* read_handle)
{
ObSqlSockSession* sess = (ObSqlSockSession*)req->get_server_handle_context();
return sess->release_read_handle(read_handle);
}
int ObPocSqlRequestOperator::read_packet(ObRequest* req,
ObICSMemPool& mem_pool,
void* read_handle,
ObSqlSockProcessor &sock_processor,
ObPacket*& pkt)
{
ObSqlSockSession* sess = (ObSqlSockSession*)req->get_server_handle_context();
return sock_processor.decode_sql_packet(mem_pool, *sess, read_handle, pkt);
}
int ObPocSqlRequestOperator::release_packet(ObRequest* req, void* read_handle, ObPacket* pkt)
{
ObSqlSockSession* sess = (ObSqlSockSession*)req->get_server_handle_context();
obmysql::ObMySQLRawPacket* mysql_packet = static_cast<obmysql::ObMySQLRawPacket*>(pkt);
return sess->consume_data(read_handle, mysql_packet->get_consume_size());
}
int ObPocSqlRequestOperator::write_response(ObRequest* req, const char* buf, int64_t sz)
{
ObSqlSockSession* sess = (ObSqlSockSession*)req->get_server_handle_context();

8
deps/oblib/src/rpc/obmysql/ob_poc_sql_request_operator.h vendored
View File

@ -31,6 +31,14 @@ public:
virtual common::ObAddr get_peer(const rpc::ObRequest* req) override;
virtual void disconnect_sql_conn(rpc::ObRequest* req) override;
virtual void finish_sql_request(rpc::ObRequest* req) override;
virtual int create_read_handle(rpc::ObRequest* req, void*& read_handle) override;
virtual int release_read_handle(rpc::ObRequest* req, void* read_handle) override;
virtual int read_packet(rpc::ObRequest* req,
ObICSMemPool& mem_pool,
void* read_handle,
ObSqlSockProcessor& sock_processor,
rpc::ObPacket*& pkt) override; // TODO change to mysql packet
virtual int release_packet(rpc::ObRequest* req, void* read_handle, rpc::ObPacket* pkt) override;
virtual int write_response(rpc::ObRequest* req, const char* buf, int64_t sz) override;
virtual int async_write_response(rpc::ObRequest* req, const char* buf, int64_t sz) override;
virtual void get_sock_desc(rpc::ObRequest* req, rpc::ObSqlSockDesc& desc) override;

174
deps/oblib/src/rpc/obmysql/ob_sql_nio.cpp vendored
View File

@ -48,6 +48,8 @@ namespace oceanbase
{
namespace obmysql
{
static const char *MEMORY_MODEL_NAME = "SqlNio";
class ObDList
{
@ -135,13 +137,63 @@ private:
int32_t ready_ CACHE_ALIGNED;
};
class SocketReader
{
public:
SocketReader(int fd)
: fd_(fd),
has_EAGAIN_(false)
{}
~SocketReader()
{
}
TO_STRING_KV(K_(fd));
void set_fd(int fd) { fd_ = fd; }
int get_fd() const { return fd_; }
bool has_EAGAIN() const { return has_EAGAIN_; }
void clear_EAGAIN() { has_EAGAIN_ = false; }
int read(char* buf, int64_t buf_size, int64_t& read_size)
{
int ret = OB_SUCCESS;
int64_t read_ret = ob_read_regard_ssl(fd_, buf, buf_size);
if (read_ret > 0) {
read_size = read_ret;
} else if (0 == read_ret) {
LOG_INFO("read fd return EOF", K_(fd));
has_EAGAIN_ = true;
ret = OB_IO_ERROR;
} else if (EAGAIN == errno || EWOULDBLOCK == errno) {
has_EAGAIN_ = true;
} else if (EINTR == errno) {
// pass
} else {
ret = OB_IO_ERROR;
LOG_WARN("read fd has error", K_(fd), K(errno));
}
return ret;
}
private:
int fd_;
bool has_EAGAIN_;
};
class ReadBuffer
{
public:
enum { IO_BUFFER_SIZE = (1<<15) - 128};
ReadBuffer(int fd): fd_(fd), has_EAGAIN_(false), request_more_data_(false),
alloc_buf_(NULL), buf_end_(NULL), cur_buf_(NULL), data_end_(NULL),
consume_sz_(0)
ReadBuffer(SocketReader& reader)
: reader_(reader),
request_more_data_(false),
alloc_buf_(NULL),
buf_end_(NULL),
cur_buf_(NULL),
data_end_(NULL),
consume_sz_(0)
{}
~ReadBuffer()
{
@ -150,18 +202,18 @@ public:
}
}
int64_t get_remain_sz() const { return remain(); }
void set_fd(int fd) { fd_ = fd; }
int peek_data(int64_t limit, const char*& buf, int64_t& sz) {
int ret = OB_SUCCESS;
if (OB_FAIL(try_read_fd(limit))) {
LOG_WARN("read fail", K(ret), K_(fd), K(limit));
LOG_WARN("read fail", K(ret), K_(reader), K(limit));
} else {
buf = cur_buf_;
sz = remain();
if (sz < limit) {
request_more_data_ = true;
}
LOG_DEBUG("peek data", K_(fd), K(limit), K(sz));
LOG_DEBUG("peek data", K_(reader), K(limit), K(sz));
}
return ret;
}
@ -170,16 +222,16 @@ public:
if (sz > 0 && sz <= remain()) {
cur_buf_ += sz;
consume_sz_ += sz;
LOG_DEBUG("consume data", K_(fd), K(sz));
LOG_DEBUG("consume data", K_(reader), K(sz));
} else {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("consume data, invalid argument", K_(fd), K(sz));
LOG_WARN("consume data, invalid argument", K_(reader), K(sz));
}
return ret;
}
bool clear_EAGAIN() {
bool ret = (has_EAGAIN_ && (remain() <= 0 || request_more_data_));
has_EAGAIN_ = false;
bool ret = (reader_.has_EAGAIN() && (remain() <= 0 || request_more_data_));
reader_.clear_EAGAIN();
request_more_data_ = false;
return ret;
}
@ -192,9 +244,9 @@ private:
} else if (remain() >= limit) {
} else if (cur_buf_ + limit > buf_end_ && OB_FAIL(switch_buffer(limit))) {
LOG_ERROR("alloc read buffer fail", K_(fd), K(ret));
LOG_ERROR("alloc read buffer fail", K_(reader), K(ret));
} else if (OB_FAIL(do_read_fd(limit))) {
LOG_WARN("do_read_fd fail", K(ret), K_(fd), K(limit));
LOG_WARN("do_read_fd fail", K(ret), K_(reader), K(limit));
}
return ret;
}
@ -210,7 +262,7 @@ private:
data_end_ = cur_buf_ + rsz;
} else if (NULL == (new_buf = (char*)alloc_io_buffer(alloc_size))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("alloc buffer fail", K(ret), K_(fd), K(alloc_size));
LOG_WARN("alloc buffer fail", K(ret), K(alloc_size), K_(reader), K(alloc_size));
} else {
char* old_buffer = alloc_buf_;
int64_t rsz = remain();
@ -228,10 +280,10 @@ private:
int do_read_fd(int64_t sz) {
int ret = OB_SUCCESS;
const int MAX_SSL_REQ_PKT_SIZE = 36;
while(remain() < sz && OB_SUCCESS == ret) {
while(remain() < sz && OB_SUCCESS == ret && !reader_.has_EAGAIN()) {
int64_t rbytes = 0;
size_t read_size = 0;
if (OB_UNLIKELY(0 == consume_sz_)) {
if (OB_UNLIKELY(0 == consume_sz_) && data_end_ == alloc_buf_) {
/*
set read size for ssl, when client want to open ssl, it will send a 36 bytes
incomplete Login Request packet and then do SSL_connect, the data flow will be
@ -242,21 +294,9 @@ private:
} else {
read_size = buf_end_ - data_end_;
}
if ((rbytes = ob_read_regard_ssl(fd_, data_end_, read_size)) > 0) {
ret = reader_.read(data_end_, read_size, rbytes);
if (OB_SUCC(ret)) {
data_end_ += rbytes;
} else if (0 == rbytes) {
LOG_INFO("read fd return EOF", K_(fd));
has_EAGAIN_ = true;
ret = OB_IO_ERROR; // for mysql protocol, it is not prossible
break;
} else if (EAGAIN == errno || EWOULDBLOCK == errno) {
has_EAGAIN_ = true;
break;
} else if (EINTR == errno) {
// pass
} else {
ret = OB_IO_ERROR;
LOG_WARN("read fd has error", K_(fd), K(errno));
}
}
return ret;
@ -268,8 +308,7 @@ private:
}
static void direct_free(void* p) { ob_free(p); }
private:
int fd_;
bool has_EAGAIN_;
SocketReader& reader_;
bool request_more_data_;
char* alloc_buf_;
char* buf_end_;
@ -341,8 +380,8 @@ private:
class ObSqlSock: public ObLink
{
public:
ObSqlSock(ObSqlNioImpl *nio, int fd): dlink_(), all_list_link_(), write_task_link_(), nio_impl_(nio),
fd_(fd), err_(0), read_buffer_(fd), need_epoll_trigger_write_(false), may_handling_(true),
ObSqlSock(ObSqlNioImpl *nio, int fd): dlink_(), all_list_link_(), write_task_link_(), nio_impl_(nio), fd_(fd),
err_(0), reader_(fd), read_buffer_(reader_), need_epoll_trigger_write_(false), may_handling_(true),
handler_close_flag_(false), need_shutdown_(false), last_decode_time_(0), last_write_time_(0),
sql_session_info_(NULL), tls_verion_option_(SSL_OP_NO_SSLv2|SSL_OP_NO_SSLv3) {
memset(sess_, 0, sizeof(sess_));
@ -350,7 +389,7 @@ public:
~ObSqlSock() {}
int64_t get_remain_sz() const { return read_buffer_.get_remain_sz(); }
TO_STRING_KV(KP(this), "session_id", get_sql_session_id(), "trace_id", get_trace_id(), "sql_handling_stage", get_sql_request_execute_state(), "sql_initiative_shutdown", need_shutdown_,
K_(fd), K_(err), K_(last_decode_time), K_(last_write_time), K_(pending_write_task), K_(need_epoll_trigger_write),
K_(reader), K_(err), K_(last_decode_time), K_(last_write_time), K_(pending_write_task), K_(need_epoll_trigger_write),
"consume_size", read_buffer_.get_consume_sz(), "pending_flag", get_pending_flag(), "may_handling_flag", get_may_handling_flag(), K_(handler_close_flag));
ObSqlNioImpl *get_nio_impl() { return nio_impl_; }
void set_nio_impl(ObSqlNioImpl *impl) { nio_impl_ = impl; }
@ -361,21 +400,58 @@ public:
if (fd_ >= 0) {
ob_fd_disable_ssl(fd_);
close(fd_);
read_buffer_.set_fd(-1);
reader_.set_fd(-1);
fd_ = -1;
}
}
void set_last_decode_succ_time(int64_t time) { last_decode_time_ = time; }
int64_t get_consume_sz() { return read_buffer_.get_consume_sz(); }
int peek_data(int64_t limit, const char*& buf, int64_t& sz) {
return read_buffer_.peek_data(limit ,buf, sz);
int peek_data(void* read_handle, int64_t limit, const char*& buf, int64_t& sz)
{
ReadBuffer *read_buffer = &read_buffer_;
if (OB_UNLIKELY(OB_NOT_NULL(read_handle))) {
read_buffer = static_cast<ReadBuffer *>(read_handle);
// The application layer try to read packet but it may failed as the EAGAIN flag is set.
read_buffer->clear_EAGAIN();
}
return read_buffer->peek_data(limit, buf, sz);
}
int consume_data(int64_t sz) { return read_buffer_.consume_data(sz); }
int consume_data(void* read_handle, int64_t sz)
{
ReadBuffer *read_buffer = (OB_ISNULL(read_handle)) ? &read_buffer_ : static_cast<ReadBuffer *>(read_handle);
return read_buffer->consume_data(sz);
}
void init_write_task(const char* buf, int64_t sz) {
pending_write_task_.init(buf, sz);
}
int create_read_handle(void *& handle)
{
int ret = OB_SUCCESS;
ReadBuffer * buffer = OB_NEW(ReadBuffer, MEMORY_MODEL_NAME, reader_);
if (OB_ISNULL(buffer)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
} else {
handle = static_cast<void *>(buffer);
}
return ret;
}
int release_read_handle(void * handle)
{
int ret = OB_SUCCESS;
if (OB_ISNULL(handle)) {
ret = OB_INVALID_ARGUMENT;
} else {
ReadBuffer* read_handle = static_cast<ReadBuffer*>(handle);
OB_DELETE(ReadBuffer, MEMORY_MODEL_NAME, read_handle);
}
return ret;
}
bool is_need_epoll_trigger_write() const { return need_epoll_trigger_write_; }
int do_pending_write(bool& become_clean) {
int ret = OB_SUCCESS;
@ -453,6 +529,7 @@ private:
ObSqlNioImpl *nio_impl_;
int fd_;
int err_;
SocketReader reader_;
ReadBuffer read_buffer_;
ReadyFlag ready_flag_;
SingleWaitCond write_cond_;
@ -505,7 +582,7 @@ int ObSqlSock::set_ssl_enabled()
{
int ret = OB_SUCCESS;
if (OB_FAIL(ob_fd_enable_ssl_for_server(fd_, OB_SSL_CTX_ID_SQL_NIO, tls_verion_option_))) {
LOG_WARN("sqlnio enable ssl for server failed", K(ret), K(fd_));
LOG_WARN("sqlnio enable ssl for server failed", K(ret), K(reader_));
}
return ret;
}
@ -529,7 +606,7 @@ int ObSqlSock::write_handshake_packet(const char* buf, int64_t sz) {
//will be treated as error
IGNORE_RETURN(set_error(EIO));
ret = OB_IO_ERROR;
LOG_WARN("write data error", K_(fd), K(errno));
LOG_WARN("write data error", K(fd_), K(errno));
}
}
last_write_time_ = ObTimeUtility::current_time();
@ -1159,14 +1236,23 @@ bool ObSqlNio::has_error(void* sess)
return sess2sock(sess)->has_error();
}
int ObSqlNio::peek_data(void* sess, int64_t limit, const char*& buf, int64_t& sz)
int ObSqlNio::create_read_handle(void* sess, void*& read_handle)
{
return sess2sock(sess)->peek_data(limit, buf, sz);
return sess2sock(sess)->create_read_handle(read_handle);
}
int ObSqlNio::release_read_handle(void* sess, void* read_handle)
{
return sess2sock(sess)->release_read_handle(read_handle);
}
int ObSqlNio::consume_data(void* sess, int64_t sz)
int ObSqlNio::peek_data(void* sess, void* read_handle, int64_t limit, const char*& buf, int64_t& sz)
{
return sess2sock(sess)->consume_data(sz);
return sess2sock(sess)->peek_data(read_handle, limit, buf, sz);
}
int ObSqlNio::consume_data(void* sess, void* read_handle, int64_t sz)
{
return sess2sock(sess)->consume_data(read_handle, sz);
}
int ObSqlNio::write_data(void* sess, const char* buf, int64_t sz)

6
deps/oblib/src/rpc/obmysql/ob_sql_nio.h vendored
View File

@ -37,8 +37,10 @@ public:
bool has_error(void* sess);
void destroy_sock(void* sess);
void revert_sock(void* sess);
int peek_data(void* sess, int64_t limit, const char*& buf, int64_t& sz);
int consume_data(void* sess, int64_t sz);
int create_read_handle(void* sess, void*& read_handle);
int release_read_handle(void* sess, void* read_handle);
int peek_data(void* sess, void* read_handle, int64_t limit, const char*& buf, int64_t& sz);
int consume_data(void* sess, void* read_handle, int64_t sz);
int write_data(void* sess, const char* buf, int64_t sz);
void async_write_data(void* sess, const char* buf, int64_t sz);
void stop();

5
deps/oblib/src/rpc/obmysql/ob_sql_nio_server.cpp vendored
View File

@ -59,6 +59,11 @@ void ObSqlNioServer::update_tcp_keepalive_params(int keepalive_enabled, uint32_t
nio_.update_tcp_keepalive_params(keepalive_enabled, tcp_keepidle, tcp_keepintvl, tcp_keepcnt);
}
ObSqlSockProcessor& ObSqlNioServer::get_sql_sock_processor()
{
return thread_processor_;
}
ObSqlNioServer* global_sql_nio_server = NULL;
}; // end namespace obmysql
}; // end namespace oceanbase

3
deps/oblib/src/rpc/obmysql/ob_sql_nio_server.h vendored
View File

@ -40,6 +40,9 @@ public:
void wait();
void destroy();
void update_tcp_keepalive_params(int keepalive_enabled, uint32_t tcp_keepidle, uint32_t tcp_keepintvl, uint32_t tcp_keepcnt);
ObSqlSockProcessor& get_sql_sock_processor();
private:
ObSqlSockProcessor thread_processor_; // for tenant worker
ObSqlSockHandler io_handler_; // for io thread

2
deps/oblib/src/rpc/obmysql/ob_sql_sock_handler.cpp vendored
View File

@ -130,7 +130,7 @@ int ObSqlSockHandler::on_readable(void* udata)
if (NULL == sess) {
ret = OB_INVALID_ARGUMENT;
LOG_ERROR("sess is null!", K(ret));
} else if (OB_FAIL(sock_processor_.decode_sql_packet(*sess, pkt))) {
} else if (OB_FAIL(sock_processor_.decode_sql_packet(sess->pool_, *sess, NULL, pkt))) {
LOG_WARN("decode sql req fail", K(ret));
} else if (NULL == pkt) {
sess->revert_sock();

22
deps/oblib/src/rpc/obmysql/ob_sql_sock_processor.cpp vendored
View File

@ -37,7 +37,10 @@ static int processor_do_decode(ObVirtualCSProtocolProcessor& processor, ObSMConn
return ret;
}
int ObSqlSockProcessor::decode_sql_packet(ObSqlSockSession& sess, rpc::ObPacket*& ret_pkt)
int ObSqlSockProcessor::decode_sql_packet(ObICSMemPool& mem_pool,
ObSqlSockSession& sess,
void* read_handle,
rpc::ObPacket*& ret_pkt)
{
int ret = OB_SUCCESS;
ObSMConnection& conn = sess.conn_;
@ -46,17 +49,17 @@ int ObSqlSockProcessor::decode_sql_packet(ObSqlSockSession& sess, rpc::ObPacket*
const char* start = NULL;
int64_t limit = 1;
int64_t read_sz = 0;
char* buf = NULL;
ret_pkt = NULL;
while(OB_SUCCESS == ret && NULL == ret_pkt) {
bool need_read_more = false;
int64_t consume_sz = 0;
if (OB_FAIL(sess.peek_data(limit, start, read_sz))) {
if (OB_FAIL(sess.peek_data(read_handle, limit, start, read_sz))) {
LOG_WARN("peed data fail", K(ret));
} else if (read_sz < limit) {
break;
} else if (OB_FAIL(processor_do_decode(*processor, conn, sess.pool_, start, read_sz, pkt, limit, consume_sz))) {
} else if (OB_FAIL(processor_do_decode(*processor, conn, mem_pool, start, read_sz, pkt, limit, consume_sz))) {
LOG_WARN("do_decode fail", K(ret));
} else if (NULL == pkt) {
// try read more
@ -70,13 +73,18 @@ int ObSqlSockProcessor::decode_sql_packet(ObSqlSockSession& sess, rpc::ObPacket*
} else if (!conn.is_in_authed_phase()) {
ret_pkt = pkt;
sess.set_last_pkt_sz(consume_sz);
} else if (OB_FAIL(processor->do_splice(conn, sess.pool_, (void*&)pkt, need_read_more))) {
} else if (OB_FAIL(processor->do_splice(conn, mem_pool, (void*&)pkt, need_read_more))) {
LOG_WARN("do_splice fail");
} else if (!need_read_more) {
ret_pkt = pkt;
sess.set_last_pkt_sz(consume_sz);
if (NULL == read_handle) {
sess.set_last_pkt_sz(consume_sz);
} else if (OB_LIKELY(ret_pkt != NULL)) {
ObMySQLRawPacket *raw_packet = static_cast<ObMySQLRawPacket *>(ret_pkt);
raw_packet->set_consume_size(consume_sz);
}
} else {
sess.consume_data(consume_sz);
sess.consume_data(read_handle, consume_sz); // assert read_handle == NULL
limit = 1; // new pkt need more data
}
}

2
deps/oblib/src/rpc/obmysql/ob_sql_sock_processor.h vendored
View File

@ -32,7 +32,7 @@ public:
ObSqlSockProcessor(ObMySQLHandler& handler):
mysql_processor_(), compress_processor_(), ob_2_0_processor_() {}
~ObSqlSockProcessor() {}
int decode_sql_packet(ObSqlSockSession& sess, rpc::ObPacket*& pkt);
int decode_sql_packet(ObICSMemPool& mem_pool, ObSqlSockSession& sess, void* read_handle, rpc::ObPacket*& pkt);
int build_sql_req(ObSqlSockSession& sess, rpc::ObPacket* pkt, rpc::ObRequest*& sql_req);
private:
ObVirtualCSProtocolProcessor *get_protocol_processor(common::ObCSProtocolType type);

20
deps/oblib/src/rpc/obmysql/ob_sql_sock_session.cpp vendored
View File

@ -71,7 +71,7 @@ void ObSqlSockSession::shutdown()
void ObSqlSockSession::revert_sock()
{
if (last_pkt_sz_ > 0) {
nio_->consume_data((void*)this, last_pkt_sz_);
nio_->consume_data((void*)this, NULL, last_pkt_sz_);
last_pkt_sz_ = 0;
}
sql_req_.reset_trace_id();
@ -100,13 +100,23 @@ bool ObSqlSockSession::has_error()
return nio_->has_error((void*)this);
}
int ObSqlSockSession::peek_data(int64_t limit, const char*& buf, int64_t& sz)
int ObSqlSockSession::create_read_handle(void *& read_handle)
{
return nio_->create_read_handle((void*)this, read_handle);
}
int ObSqlSockSession::release_read_handle(void * read_handle)
{
return nio_->release_read_handle((void*)this, read_handle);
}
int ObSqlSockSession::peek_data(void *handle, int64_t limit, const char*& buf, int64_t& sz)
{
int ret = OB_SUCCESS;
if (has_error()) {
ret = OB_IO_ERROR;
LOG_WARN("sock has error", K(ret));
} else if (OB_FAIL(nio_->peek_data((void*)this, limit, buf, sz))) {
} else if (OB_FAIL(nio_->peek_data((void*)this, handle, limit, buf, sz))) {
destroy_sock();
}
return ret;
@ -117,13 +127,13 @@ void ObSqlSockSession::clear_sql_session_info()
nio_->reset_sql_session_info(this);
}
int ObSqlSockSession::consume_data(int64_t sz)
int ObSqlSockSession::consume_data(void* read_handle, int64_t sz)
{
int ret = OB_SUCCESS;
if (has_error()) {
ret = OB_IO_ERROR;
LOG_WARN("sock has error", K(ret));
} else if (OB_FAIL(nio_->consume_data((void*)this, sz))) {
} else if (OB_FAIL(nio_->consume_data((void*)this, read_handle, sz))) {
destroy_sock();
}
return ret;

6
deps/oblib/src/rpc/obmysql/ob_sql_sock_session.h vendored
View File

@ -46,8 +46,10 @@ public:
void destroy();
void destroy_sock();
bool has_error();
int peek_data(int64_t limit, const char*& buf, int64_t& sz);
int consume_data(int64_t sz);
int create_read_handle(void*& read_handle);
int release_read_handle(void* read_handle);
int peek_data(void* read_handle, int64_t limit, const char*& buf, int64_t& sz);
int consume_data(void* read_handle, int64_t sz);
int write_data(const char* buf, int64_t sz);
int async_write_data(const char* buf, int64_t sz);
void on_flushed();

72
deps/oblib/src/rpc/obmysql/packet/ompk_local_infile.cpp vendored Normal file
View File

@ -0,0 +1,72 @@
/**
* 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.
*/
#define USING_LOG_PREFIX RPC_OBMYSQL
#include "rpc/obmysql/packet/ompk_local_infile.h"
#include "rpc/obmysql/ob_mysql_util.h"
namespace oceanbase
{
namespace obmysql
{
OMPKLocalInfile::OMPKLocalInfile() :
packet_type_(0xFB)
{}
OMPKLocalInfile::~OMPKLocalInfile()
{}
int OMPKLocalInfile::serialize(char *buffer, int64_t len, int64_t &pos) const
{
int ret = OB_SUCCESS;
if (OB_ISNULL(buffer) || OB_UNLIKELY(len - pos < 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", KP(buffer), K(len), K(pos), K(ret));
} else if (OB_UNLIKELY(len - pos < static_cast<int64_t>(get_serialize_size()))) {
ret = OB_SIZE_OVERFLOW;
LOG_WARN("invalid argument", K(len), K(pos), "need_size", get_serialize_size());
} else if (OB_FAIL(ObMySQLUtil::store_int1(buffer, len, packet_type_, pos))) {
LOG_WARN("store fail", KP(buffer), K(len), K(pos), K(ret));
} else if (OB_FAIL(ObMySQLUtil::store_obstr_nzt(buffer, len, filename_, pos))) {
LOG_WARN("store fail", KP(buffer), K(len), K(pos), K(ret));
}
return ret;
}
int64_t OMPKLocalInfile::get_serialize_size() const
{
int64_t len = 1/* packet type*/ + filename_.length();
return len;
}
int64_t OMPKLocalInfile::to_string(char *buf, const int64_t buf_len) const
{
int64_t pos = 0;
J_OBJ_START();
J_KV("header", hdr_,
"packet_type", "LOCAL INFILE",
K_(filename));
J_OBJ_END();
return pos;
}
void OMPKLocalInfile::set_filename(const ObString &filename)
{
filename_ = filename;
}
} // namespace obmysql
} // namespace oceanbase

54
deps/oblib/src/rpc/obmysql/packet/ompk_local_infile.h vendored Normal file
View File

@ -0,0 +1,54 @@
/**
* 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.
*/
#ifndef _OMPK_LOCAL_INFILE_H_
#define _OMPK_LOCAL_INFILE_H_
#include "rpc/obmysql/ob_mysql_packet.h"
#include "lib/ob_define.h"
namespace oceanbase
{
namespace obmysql
{
// In the MySQL client/server protocol, server send a `local infile` message to client
// to specific the file name to load.
// format:
// int<1> | packet type | 0xFB: LOCAL INFILE
// string<EOF> | filename | the path to the file the client shall send
// The notation is "string<EOF>" Strings whose length will be calculated by the packet remaining length.
class OMPKLocalInfile : public ObMySQLRawPacket
{
public:
OMPKLocalInfile();
virtual ~OMPKLocalInfile();
virtual int serialize(char *buffer, int64_t len, int64_t &pos) const override;
virtual int64_t get_serialize_size() const override;
virtual int64_t to_string(char *buf, const int64_t buf_len) const override;
void set_filename(const ObString &filename);
inline ObMySQLPacketType get_mysql_packet_type() override { return ObMySQLPacketType::PKT_FILENAME; }
private:
DISALLOW_COPY_AND_ASSIGN(OMPKLocalInfile);
int8_t packet_type_;
ObString filename_;
};
} // end of namespace obmysql
} // end of namespace oceanbase
#endif /* _OMPK_LOCAL_INFILE_H_ */

1
src/observer/mysql/ob_query_driver.h
View File

@ -69,6 +69,7 @@ public:
bool has_more_result,
bool &can_retry,
int64_t fetch_limit = common::OB_INVALID_COUNT);
ObIMPPacketSender& get_packet_sender() { return sender_; }
int response_query_header(const ColumnsFieldIArray &fields,
bool has_more_result = false,
bool need_set_ps_out = false,

9
src/observer/mysql/obmp_base.cpp
View File

@ -203,6 +203,15 @@ int ObMPBase::get_conn_id(uint32_t &sessid) const
return packet_sender_.get_conn_id(sessid);
}
int ObMPBase::read_packet(obmysql::ObICSMemPool& mem_pool, obmysql::ObMySQLPacket *&pkt)
{
return packet_sender_.read_packet(mem_pool, pkt);
}
int ObMPBase::release_packet(obmysql::ObMySQLPacket* pkt)
{
return packet_sender_.release_packet(pkt);
}
int ObMPBase::send_error_packet(int err,
const char* errmsg,
bool is_partition_hit /* = true */,

3
src/observer/mysql/obmp_base.h
View File

@ -55,6 +55,9 @@ protected:
virtual int update_last_pkt_pos() { return packet_sender_.update_last_pkt_pos(); }
virtual bool need_send_extra_ok_packet() { return packet_sender_.need_send_extra_ok_packet(); }
virtual int read_packet(obmysql::ObICSMemPool& mem_pool, obmysql::ObMySQLPacket*& pkt) override;
virtual int release_packet(obmysql::ObMySQLPacket* pkt) override;
// Response a packet to client peer.
//

1
src/observer/mysql/obmp_connect.cpp
View File

@ -1689,6 +1689,7 @@ int ObMPConnect::check_update_proxy_capability(ObSMConnection &conn) const
server_proxy_cap_flag.cap_flags_.OB_CAP_PROXY_SESSION_VAR_SYNC = 1;
server_proxy_cap_flag.cap_flags_.OB_CAP_PROXY_FULL_LINK_TRACING_EXT = 1;
server_proxy_cap_flag.cap_flags_.OB_CAP_SERVER_DUP_SESS_INFO_SYNC = 1;
server_proxy_cap_flag.cap_flags_.OB_CAP_LOCAL_FILES = 1;
if (GET_MIN_CLUSTER_VERSION() >= CLUSTER_VERSION_4_3_0_0) {
server_proxy_cap_flag.cap_flags_.OB_CAP_PROXY_CLIENT_SESSION_ID = 1;
} else {

90
src/observer/mysql/obmp_packet_sender.cpp
View File

@ -66,6 +66,7 @@ int64_t ObOKPParam::to_string(char *buf, const int64_t buf_len) const
ObMPPacketSender::ObMPPacketSender()
: req_(NULL),
seq_(0),
read_handle_(NULL),
comp_context_(),
proto20_context_(),
ez_buf_(NULL),
@ -85,6 +86,8 @@ ObMPPacketSender::~ObMPPacketSender()
void ObMPPacketSender::reset()
{
(void)release_read_handle();
req_ = NULL;
seq_ = 0;
comp_context_.reset();
@ -194,6 +197,52 @@ int ObMPPacketSender::alloc_ezbuf()
return ret;
}
int ObMPPacketSender::read_packet(obmysql::ObICSMemPool &mem_pool, obmysql::ObMySQLPacket *&mysql_pkt)
{
int ret = OB_SUCCESS;
mysql_pkt = NULL;
if (OB_ISNULL(read_handle_)) {
ret = init_read_handle();
}
if (OB_SUCC(ret)) {
rpc::ObPacket *pkt = NULL;
ret = SQL_REQ_OP.read_packet(req_, mem_pool, read_handle_, pkt);
if (pkt != NULL) {
mysql_pkt = static_cast<obmysql::ObMySQLPacket *>(pkt);
const uint8_t seq = mysql_pkt->get_seq();
seq_ = seq + 1;
if (comp_context_.use_compress()) {
comp_context_.seq_ = comp_context_.conn_->compressed_pkt_context_.last_pkt_seq_ + 1;
}
if (proto20_context_.is_proto20_used_) {
ObSMConnection *conn = get_conn();
if (OB_NOT_NULL(conn)) {
proto20_context_.comp_seq_ = seq_;
proto20_context_.request_id_ = conn->proto20_pkt_context_.proto20_last_request_id_;
proto20_context_.proto20_seq_ = static_cast<uint8_t>(conn->proto20_pkt_context_.proto20_last_pkt_seq_ + 1);
proto20_context_.is_new_extra_info_ = conn->proxy_cap_flags_.is_new_extra_info_support();
}
}
}
}
return ret;
}
int ObMPPacketSender::release_packet(obmysql::ObMySQLPacket* pkt)
{
int ret = OB_SUCCESS;
if (OB_ISNULL(read_handle_)) {
ret = OB_NOT_INIT;
} else {
ret = SQL_REQ_OP.release_packet(req_, read_handle_, static_cast<rpc::ObPacket*>(pkt));
}
return ret;
}
int ObMPPacketSender::response_compose_packet(obmysql::ObMySQLPacket &pkt,
obmysql::ObMySQLPacket &okp,
sql::ObSQLSessionInfo* session,
@ -854,6 +903,20 @@ int ObMPPacketSender::try_encode_with(ObMySQLPacket &pkt,
}
__builtin_prefetch(&conn_->pkt_rec_wrapper_.pkt_rec_[conn_->pkt_rec_wrapper_.cur_pkt_pos_
% ObPacketRecordWrapper::REC_BUF_SIZE]);
if (OB_UNLIKELY(pkt.get_mysql_packet_type() == ObMySQLPacketType::PKT_FILENAME)) {
proto20_context_.is_filename_packet_ = true;
if (comp_context_.use_compress() && comp_context_.is_proxy_compress_based()) {
// The compression protocol between in observer and obproxy requires that the sequence id should use
// last packet's sequence id if there is the last response packet to obproxy. And PKT_FILENAME is
// the "last packet" reponse to obproxy this time.
// `seq_` means observer will not send more response and client can resolve the response packets now.
// NOTE: This action is not compatible with MySQL protocol
comp_context_.seq_--;
}
}
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_UNLIKELY(!conn_valid_)) {
@ -946,6 +1009,10 @@ int ObMPPacketSender::flush_buffer(const bool is_last)
LOG_ERROR("fail to fill ob20 protocol header and tailer", K(ret));
} else {
}
// `filename packet` is used in load local protocol.
// We use the field to indicate if we should set `is last packet flag`
proto20_context_.is_filename_packet_ = false;
}
int64_t buf_sz = ez_buf_->last - ez_buf_->pos;
@ -1105,6 +1172,7 @@ int ObMPPacketSender::update_transmission_checksum_flag(const ObSQLSessionInfo &
void ObMPPacketSender::finish_sql_request()
{
if (conn_valid_ && !req_has_wokenup_) {
(void)release_read_handle();
SQL_REQ_OP.finish_sql_request(req_);
req_has_wokenup_ = true;
ez_buf_ = NULL;
@ -1141,6 +1209,28 @@ bool ObMPPacketSender::has_pl()
int64_t ObMPPacketSender::TRY_EZ_BUF_SIZES[] = {64*1024, 128*1024, 2*1024*1024 - 1024, 4*1024*1024 - 1024,
64*1024*1024 - 1024, 128*1024*1024, 512*1024*1024, 1024*1024*1024};
int ObMPPacketSender::init_read_handle()
{
int ret = OB_SUCCESS;
if (OB_NOT_NULL(read_handle_)) {
ret = OB_INVALID_ARGUMENT;
} else {
ret = SQL_REQ_OP.create_read_handle(req_, read_handle_);
LOG_DEBUG("create read handle", KP(req_), KP_(read_handle));
}
return ret;
}
int ObMPPacketSender::release_read_handle()
{
int ret = OB_SUCCESS;
if (OB_NOT_NULL(read_handle_)) {
LOG_DEBUG("release read handle", KP(req_), KP_(read_handle));
ret = SQL_REQ_OP.release_read_handle(req_, read_handle_);
read_handle_ = NULL;
}
return ret;
}
}; // end namespace observer
}; // end namespace oceanbase

21
src/observer/mysql/obmp_packet_sender.h
View File

@ -93,6 +93,20 @@ public:
virtual void disconnect() = 0;
virtual void force_disconnect() = 0;
virtual int update_last_pkt_pos() = 0;
/**
* read a mysql packet from the socket channel
*
* Error returned if no message. You need to determine the exact cause of the error to decide
* whether to continue to wait for the message or interrupt the service.
*
* @param mem_pool The memory manager to hold the `pkt`
* @param pkt The mysql packet received or nullptr if not message came in
* @return OB_SUCCESS The socket channel is OK. The `pk` returned may be null is no data came in.
* OB_IO_ERROR Something wrong with the channel, such as it was closed.
*/
virtual int read_packet(obmysql::ObICSMemPool& mem_pool, obmysql::ObMySQLPacket *&pkt) = 0;
virtual int release_packet(obmysql::ObMySQLPacket* pkt) = 0;
virtual int response_packet(obmysql::ObMySQLPacket &pkt, sql::ObSQLSessionInfo* session) = 0;
virtual int send_error_packet(int err,
const char* errmsg,
@ -117,6 +131,8 @@ public:
virtual void disconnect() override;
virtual void force_disconnect() override;
virtual int update_last_pkt_pos() override;
virtual int read_packet(obmysql::ObICSMemPool& mem_pool, obmysql::ObMySQLPacket *&pkt) override;
virtual int release_packet(obmysql::ObMySQLPacket* pkt) override;
virtual int response_packet(obmysql::ObMySQLPacket &pkt, sql::ObSQLSessionInfo* session) override;
// when connect with proxy, need to append extra ok packet to last statu packet
int response_compose_packet(obmysql::ObMySQLPacket &pkt,
@ -154,6 +170,10 @@ public:
bool has_pl();
int alloc_ezbuf();
private:
int init_read_handle();
int release_read_handle();
private:
static const int64_t MAX_TRY_STEPS = 8;
static int64_t TRY_EZ_BUF_SIZES[MAX_TRY_STEPS];
@ -170,6 +190,7 @@ private:
protected:
rpc::ObRequest *req_;
uint8_t seq_;
void * read_handle_;
obmysql::ObCompressionContext comp_context_;
obmysql::ObProto20Context proto20_context_;
easy_buf_t *ez_buf_;

1
src/observer/ob_srv_network_frame.cpp
View File

@ -207,6 +207,7 @@ int ObSrvNetworkFrame::start()
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_ERROR("allocate memory for global_sql_nio_server failed", K(ret));
} else {
SQL_REQ_OP.set_sql_sock_processor(obmysql::global_sql_nio_server->get_sql_sock_processor());
int sql_net_thread_count = (int)GCONF.sql_net_thread_count;
if (sql_net_thread_count == 0) {
if (GCONF.net_thread_count == 0) {

3
src/observer/virtual_table/ob_all_virtual_load_data_stat.cpp
View File

@ -137,7 +137,8 @@ int ObAllVirtualLoadDataStat::inner_get_next_row(ObNewRow *&row)
case ESTIMATED_REMAINING_TIME: {
int64_t load_time = current_time - job_status->start_time_;
int64_t estimated_remaining_time = 0;
if ((job_status->parsed_bytes_ != 0) && OB_LIKELY(load_time != 0)) {
// in load data local infile, the total_bytes_ is 0 or -1
if ((job_status->parsed_bytes_ != 0) && OB_LIKELY(load_time != 0) && job_status->total_bytes_ > 0) {
double speed = (double)job_status->parsed_bytes_ / load_time;
if (OB_LIKELY(speed != 0)) {
int64_t remain_bytes = job_status->total_bytes_ - job_status->parsed_bytes_;

0
src/share/ob_errno.cpp Executable file → Normal file
View File

0
src/share/ob_errno.h Executable file → Normal file
View File

1
src/sql/CMakeLists.txt
View File

@ -183,6 +183,7 @@ ob_set_subtarget(ob_sql engine_cmd
engine/cmd/ob_kill_executor.cpp
engine/cmd/ob_kill_session_arg.cpp
engine/cmd/ob_load_data_direct_impl.cpp
engine/cmd/ob_load_data_file_reader.cpp
engine/cmd/ob_load_data_executor.cpp
engine/cmd/ob_load_data_impl.cpp
engine/cmd/ob_load_data_parser.cpp

378
src/sql/engine/cmd/ob_load_data_direct_impl.cpp
View File

@ -9,7 +9,6 @@
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#define USING_LOG_PREFIX SQL_ENG
#include "sql/engine/cmd/ob_load_data_direct_impl.h"
@ -20,6 +19,7 @@
#include "observer/table_load/ob_table_load_table_ctx.h"
#include "observer/table_load/ob_table_load_task.h"
#include "observer/table_load/ob_table_load_task_scheduler.h"
#include "observer/mysql/ob_query_driver.h"
#include "share/schema/ob_schema_getter_guard.h"
#include "share/ob_device_manager.h"
#include "share/backup/ob_backup_io_adapter.h"
@ -226,208 +226,6 @@ int ObLoadDataDirectImpl::Logger::log_error_line(const ObString &file_name, int6
return ret;
}
/**
* RandomFileReader
*/
ObLoadDataDirectImpl::RandomFileReader::RandomFileReader() : is_inited_(false)
{
}
ObLoadDataDirectImpl::RandomFileReader::~RandomFileReader()
{
}
int ObLoadDataDirectImpl::RandomFileReader::open(const DataAccessParam &data_access_param, const ObString &filename)
{
int ret = OB_SUCCESS;
UNUSED(data_access_param);
if (IS_INIT) {
ret = OB_INIT_TWICE;
LOG_WARN("RandomFileReader init twice", KR(ret), KP(this));
} else if (OB_FAIL(file_reader_.open(filename.ptr(), false))) {
LOG_WARN("fail to open file", KR(ret), K(filename));
} else {
filename_ = filename;
is_inited_ = true;
}
return ret;
}
int ObLoadDataDirectImpl::RandomFileReader::pread(char *buf, int64_t count, int64_t offset, int64_t &read_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("RandomFileReader not init", KR(ret), KP(this));
} else if (OB_FAIL(file_reader_.pread(buf, count, offset, read_size))) {
LOG_WARN("fail to pread file buf", KR(ret), K(count), K(offset), K(read_size));
}
return ret;
}
int ObLoadDataDirectImpl::RandomFileReader::get_file_size(int64_t &file_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("RandomFileReader not init", KR(ret), KP(this));
} else {
file_size = ::get_file_size(filename_.ptr());
}
return ret;
}
/**
* RandomOSSReader
*/
ObLoadDataDirectImpl::RandomOSSReader::RandomOSSReader() : device_handle_(nullptr), is_inited_(false)
{
}
ObLoadDataDirectImpl::RandomOSSReader::~RandomOSSReader()
{
if (fd_.is_valid()) {
device_handle_->close(fd_);
fd_.reset();
}
if (nullptr != device_handle_) {
common::ObDeviceManager::get_instance().release_device(device_handle_);
device_handle_ = nullptr;
}
}
int ObLoadDataDirectImpl::RandomOSSReader::open(const DataAccessParam &data_access_param,
const ObString &filename)
{
int ret = OB_SUCCESS;
ObIODOpt opt;
ObIODOpts iod_opts;
ObBackupIoAdapter util;
iod_opts.opts_ = &opt;
iod_opts.opt_cnt_ = 0;
if (IS_INIT) {
ret = OB_INIT_TWICE;
LOG_WARN("RandomOSSReader init twice", KR(ret), KP(this));
} else if (OB_FAIL(
util.get_and_init_device(device_handle_, &data_access_param.access_info_, filename))) {
LOG_WARN("fail to get device manager", KR(ret), K(filename));
} else if (OB_FAIL(util.set_access_type(&iod_opts, false, 1))) {
LOG_WARN("fail to set access type", KR(ret));
} else if (OB_FAIL(device_handle_->open(to_cstring(filename), -1, 0, fd_, &iod_opts))) {
LOG_WARN("fail to open oss file", KR(ret), K(filename));
} else {
is_inited_ = true;
}
return ret;
}
int ObLoadDataDirectImpl::RandomOSSReader::pread(char *buf, int64_t count, int64_t offset,
int64_t &read_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("RandomOSSReader not init", KR(ret), KP(this));
} else if (OB_FAIL(device_handle_->pread(fd_, offset, count, buf, read_size))) {
LOG_WARN("fail to pread oss buf", KR(ret), K(offset), K(count), K(read_size));
}
return ret;
}
int ObLoadDataDirectImpl::RandomOSSReader::get_file_size(int64_t &file_size)
{
int ret = OB_SUCCESS;
ObBackupIoAdapter util;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("RandomOSSReader not init", KR(ret), KP(this));
} else if (OB_FAIL(util.get_file_size(device_handle_, fd_, file_size))) {
LOG_WARN("fail to get oss file size", KR(ret), K(file_size));
}
return ret;
}
/**
* SequentialDataAccessor
*/
ObLoadDataDirectImpl::SequentialDataAccessor::SequentialDataAccessor()
: random_io_device_(nullptr), offset_(0), is_inited_(false)
{
}
ObLoadDataDirectImpl::SequentialDataAccessor::~SequentialDataAccessor()
{
}
int ObLoadDataDirectImpl::SequentialDataAccessor::init(const DataAccessParam &data_access_param,
const ObString &filename)
{
int ret = OB_SUCCESS;
if (IS_INIT) {
ret = OB_INIT_TWICE;
LOG_WARN("ObLoadDataDirectImpl::SequentialDataAccessor init twice", KR(ret), KP(this));
} else if (OB_UNLIKELY(!data_access_param.is_valid() || filename.empty())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid args", KR(ret), K(data_access_param), K(filename));
} else {
if (data_access_param.file_location_ == ObLoadFileLocation::SERVER_DISK) {
if (OB_FAIL(random_file_reader_.open(data_access_param, filename))) {
LOG_WARN("fail to open random file reader", KR(ret), K(filename));
} else {
random_io_device_ = &random_file_reader_;
}
} else if (data_access_param.file_location_ == ObLoadFileLocation::OSS) {
if (OB_FAIL(random_oss_reader_.open(data_access_param, filename))) {
LOG_WARN("fail to open random oss reader", KR(ret), K(filename));
} else {
random_io_device_ = &random_oss_reader_;
}
} else {
ret = OB_NOT_SUPPORTED;
LOG_WARN("not supported load file location", KR(ret), K(data_access_param.file_location_));
FORWARD_USER_ERROR_MSG(ret, "not supported load file location");
}
if (OB_SUCC(ret)) {
is_inited_ = true;
}
}
return ret;
}
int ObLoadDataDirectImpl::SequentialDataAccessor::read(char *buf, int64_t count, int64_t &read_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("ObLoadDataDirectImpl::SequentialDataAccessor not init", KR(ret), KP(this));
} else if (OB_UNLIKELY(nullptr == buf || count <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid args", KR(ret), KP(buf), K(count));
} else {
if (OB_FAIL(random_io_device_->pread(buf, count, offset_, read_size))) {
LOG_WARN("fail to do pread", KR(ret), K(offset_));
} else {
offset_ += read_size;
}
}
return ret;
}
int ObLoadDataDirectImpl::SequentialDataAccessor::get_file_size(int64_t &file_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("ObLoadDataDirectImpl::SequentialDataAccessor not init", KR(ret), KP(this));
} else if (OB_FAIL(random_io_device_->get_file_size(file_size))) {
LOG_WARN("fail to get random io device file size", KR(ret), K(file_size));
}
return ret;
}
/**
* DataDescIterator
*/
@ -637,10 +435,23 @@ void ObLoadDataDirectImpl::DataBuffer::swap(DataBuffer &other)
*/
ObLoadDataDirectImpl::DataReader::DataReader()
: execute_ctx_(nullptr), end_offset_(0), read_raw_(false), is_iter_end_(false), is_inited_(false)
: execute_ctx_(nullptr),
file_reader_(nullptr),
end_offset_(0),
read_raw_(false),
is_iter_end_(false),
is_inited_(false)
{
}
ObLoadDataDirectImpl::DataReader::~DataReader()
{
if (OB_NOT_NULL(file_reader_)) {
file_reader_->~ObFileReader();
file_reader_ = nullptr;
}
}
int ObLoadDataDirectImpl::DataReader::init(const DataAccessParam &data_access_param,
LoadExecuteContext &execute_ctx,
const DataDesc &data_desc, bool read_raw)
@ -665,13 +476,28 @@ int ObLoadDataDirectImpl::DataReader::init(const DataAccessParam &data_access_pa
}
if (OB_SUCC(ret)) {
end_offset_ = data_desc.end_;
if (OB_FAIL(io_accessor_.init(data_access_param, data_desc.filename_))) {
LOG_WARN("fail to init io device", KR(ret), K(data_desc));
} else if (end_offset_ == -1 && OB_FAIL(io_accessor_.get_file_size(end_offset_))) {
LOG_WARN("fail to get file size", KR(ret), K(data_desc));
} else {
io_accessor_.seek(data_desc.start_);
ATOMIC_AAF(&execute_ctx_->job_stat_->total_bytes_, (end_offset_ - data_desc.start_));
ObFileReadParam file_read_param;
file_read_param.file_location_ = data_access_param.file_location_;
file_read_param.filename_ = data_desc.filename_;
file_read_param.access_info_ = data_access_param.access_info_;
file_read_param.packet_handle_ = &execute_ctx.exec_ctx_.get_session_info()->get_pl_query_sender()->get_packet_sender();
file_read_param.session_ = execute_ctx.exec_ctx_.get_session_info();
file_read_param.timeout_ts_ = THIS_WORKER.get_timeout_ts();
if (OB_FAIL(ObFileReader::open(file_read_param, *execute_ctx_->allocator_, file_reader_))) {
LOG_WARN("failed to open file", KR(ret), K(data_desc));
} else if (file_reader_->seekable()) {
if (end_offset_ == -1 && OB_FAIL(file_reader_->get_file_size(end_offset_))) {
LOG_WARN("fail to get file size", KR(ret), K(data_desc));
} else {
file_reader_->seek(data_desc.start_);
ATOMIC_AAF(&execute_ctx_->job_stat_->total_bytes_, (end_offset_ - data_desc.start_));
}
} else if (data_desc.start_ != 0) {
ret = OB_NOT_SUPPORTED; // should not happen
LOG_WARN("file reader asked to seek while not supported by unseekable file", KR(ret), K(file_read_param));
}
}
if (OB_SUCC(ret)) {
@ -681,6 +507,32 @@ int ObLoadDataDirectImpl::DataReader::init(const DataAccessParam &data_access_pa
return ret;
}
int ObLoadDataDirectImpl::DataReader::read_buffer(ObLoadFileBuffer &file_buffer)
{
int ret = OB_SUCCESS;
int64_t read_count = file_buffer.get_remain_len();
if (0 == read_count) {
ret = OB_BUF_NOT_ENOUGH;
LOG_WARN("cannot read more data as buffer is full", KR(ret));
} else if (!is_end_file()) {
int64_t read_size = 0;
if (end_offset_ > 0 && read_count > (end_offset_ - file_reader_->get_offset())) {
read_count = end_offset_ - file_reader_->get_offset();
}
if (OB_FAIL(file_reader_->readn(file_buffer.current_ptr(), read_count, read_size))) {
LOG_WARN("fail to read file", KR(ret));
} else if (0 == read_size) {
LOG_TRACE("read nothing", K(is_end_file()));
} else {
file_buffer.update_pos(read_size); // 更新buffer中数据长度
LOG_TRACE("read file sucess", K(read_size));
ATOMIC_AAF(&execute_ctx_->job_stat_->read_bytes_, read_size);
}
}
return ret;
}
int ObLoadDataDirectImpl::DataReader::get_next_buffer(ObLoadFileBuffer &file_buffer,
int64_t &line_count, int64_t limit)
{
@ -704,25 +556,14 @@ int ObLoadDataDirectImpl::DataReader::get_next_buffer(ObLoadFileBuffer &file_buf
if (OB_FAIL(data_trimer_.recover_incomplate_data(file_buffer))) {
LOG_WARN("fail to recover incomplate data", KR(ret));
}
// 2. 从文件里读取后续的数据
else if (!is_end_file()) {
int64_t read_count = 0;
int64_t read_size = 0;
if (FALSE_IT(read_count =
MIN(file_buffer.get_remain_len(), end_offset_ - io_accessor_.get_offset()))) {
} else if (OB_FAIL(io_accessor_.read(file_buffer.current_ptr(), read_count, read_size))) {
LOG_WARN("fail to read file", KR(ret));
} else if (OB_UNLIKELY(read_count != read_size)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected read size", KR(ret), K(read_count), K(read_size), K(end_offset_));
} else {
file_buffer.update_pos(read_size); // 更新buffer中数据长度
ATOMIC_AAF(&execute_ctx_->job_stat_->read_bytes_, read_size);
}
}
// 3. 从buffer中找出完整的行，剩下的数据缓存到data_trimer
// 2. 读取数据，然后从buffer中找出完整的行，剩下的数据缓存到data_trimer
if (OB_SUCC(ret)) {
if (!file_buffer.is_valid()) {
int64_t complete_cnt = limit;
int64_t complete_len = 0;
if (OB_FAIL(read_buffer(file_buffer))) {
LOG_WARN("failed to read buffer as there is not enough data to parse", KR(ret));
} else if (!file_buffer.is_valid()) {
is_iter_end_ = true;
ret = OB_ITER_END;
} else {
@ -760,19 +601,37 @@ int ObLoadDataDirectImpl::DataReader::get_next_raw_buffer(DataBuffer &data_buffe
} else if (is_end_file()) {
ret = OB_ITER_END;
} else if (data_buffer.get_remain_length() > 0) {
const int64_t read_count =
MIN(data_buffer.get_remain_length(), end_offset_ - io_accessor_.get_offset());
int64_t read_count = data_buffer.get_remain_length();
if (file_reader_->seekable() && read_count > end_offset_ - file_reader_->get_offset()) {
read_count = end_offset_ - file_reader_->get_offset();
}
int64_t read_size = 0;
if (OB_FAIL(io_accessor_.read(data_buffer.data() + data_buffer.get_data_length(), read_count,
if (OB_FAIL(file_reader_->readn(data_buffer.data() + data_buffer.get_data_length(), read_count,
read_size))) {
LOG_WARN("fail to read file", KR(ret));
} else if (OB_UNLIKELY(read_count != read_size)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected read size", KR(ret), K(read_count), K(read_size), K(end_offset_));
} else {
} else if (read_size > 0) {
data_buffer.update_data_length(read_size);
ATOMIC_AAF(&execute_ctx_->job_stat_->read_bytes_, read_size);
}
} else {
// read_size == 0
if (is_end_file()) {
ret = OB_ITER_END;
} else {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("read buffer got unexpected bytes.", K(read_size), K(read_count), K(ret));
}
}
}
return ret;
}
bool ObLoadDataDirectImpl::DataReader::is_end_file() const
{
bool ret = false;
if (file_reader_->eof()) {
ret = true;
} else if (end_offset_ > 0) {
ret = file_reader_->get_offset() >= end_offset_;
}
return ret;
}
@ -931,16 +790,32 @@ int ObLoadDataDirectImpl::SimpleDataSplitUtils::split(const DataAccessParam &dat
data_access_param.file_cs_type_))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected data format", KR(ret), K(data_access_param));
} else if (1 == count) {
} else if (1 == count || (ObLoadFileLocation::CLIENT_DISK == data_access_param.file_location_)) {
if (OB_FAIL(data_desc_iter.add_data_desc(data_desc))) {
LOG_WARN("fail to push back", KR(ret));
}
} else {
ObArenaAllocator allocator;
allocator.set_tenant_id(MTL_ID());
int64_t end_offset = data_desc.end_;
SequentialDataAccessor io_device;
if (OB_FAIL(io_device.init(data_access_param, data_desc.filename_))) {
LOG_WARN("fail to init io device", KR(ret), K(data_desc.filename_));
} else if (-1 == end_offset && OB_FAIL(io_device.get_file_size(end_offset))) {
ObFileReadParam file_read_param;
file_read_param.file_location_ = data_access_param.file_location_;
file_read_param.filename_ = data_desc.filename_;
file_read_param.access_info_ = data_access_param.access_info_;
file_read_param.packet_handle_ = NULL;
file_read_param.session_ = NULL;
file_read_param.timeout_ts_ = THIS_WORKER.get_timeout_ts();
ObFileReader *file_reader = NULL;
if (OB_FAIL(ObFileReader::open(file_read_param, allocator, file_reader))) {
LOG_WARN("failed to open file.", KR(ret), K(data_desc));
} else if (!file_reader->seekable()) {
if (OB_FAIL(data_desc_iter.add_data_desc(data_desc))) {
LOG_WARN("fail to push back", KR(ret));
}
} else if (-1 == end_offset && OB_FAIL(file_reader->get_file_size(end_offset))) {
LOG_WARN("fail to get io device file size", KR(ret), K(end_offset));
} else {
const int64_t file_size = end_offset - data_desc.start_;
@ -953,26 +828,24 @@ int ObLoadDataDirectImpl::SimpleDataSplitUtils::split(const DataAccessParam &dat
const char line_term_char = data_access_param.file_format_.line_term_str_.ptr()[0];
const int64_t buf_size = (128LL << 10) + 1;
const int64_t split_size = file_size / count;
ObArenaAllocator allocator;
char *buf = nullptr;
int64_t read_size = 0;
DataDesc data_desc_ret;
data_desc_ret.file_idx_ = data_desc.file_idx_;
data_desc_ret.filename_ = data_desc.filename_;
data_desc_ret.start_ = data_desc.start_;
allocator.set_tenant_id(MTL_ID());
if (OB_ISNULL(buf = static_cast<char *>(allocator.alloc(buf_size)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("fail to alloc memory", KR(ret));
}
for (int64_t i = 0; OB_SUCC(ret) && i < count - 1; ++i) {
int64_t read_offset = data_desc.start_ + split_size * (i + 1);
io_device.seek(read_offset);
file_reader->seek(read_offset);
char *found = nullptr;
while (OB_SUCC(ret) && end_offset > io_device.get_offset() && nullptr == found) {
read_offset = io_device.get_offset();
while (OB_SUCC(ret) && end_offset > file_reader->get_offset() && nullptr == found) {
read_offset = file_reader->get_offset();
const int64_t read_count = MIN(end_offset - read_offset, buf_size - 1);
if (OB_FAIL(io_device.read(buf, read_count, read_size))) {
if (OB_FAIL(file_reader->readn(buf, read_count, read_size))) {
LOG_WARN("fail to do read", KR(ret), K(read_offset), K(read_count));
} else if (OB_UNLIKELY(read_count != read_size)) {
ret = OB_ERR_UNEXPECTED;
@ -1004,6 +877,11 @@ int ObLoadDataDirectImpl::SimpleDataSplitUtils::split(const DataAccessParam &dat
}
}
}
if (OB_NOT_NULL(file_reader)) {
file_reader->~ObFileReader();
allocator.free(file_reader);
}
}
return ret;
}
@ -1139,6 +1017,7 @@ int ObLoadDataDirectImpl::FileLoadExecutor::execute()
LOG_WARN("fail to prepare execute", KR(ret));
}
LOG_TRACE("file load executor prepare execute done", K(ret));
while (OB_SUCC(ret) && OB_SUCC(execute_ctx_->exec_ctx_.check_status())) {
TaskHandle *handle = nullptr;
if (OB_FAIL(get_next_task_handle(handle))) {
@ -1177,8 +1056,9 @@ int ObLoadDataDirectImpl::FileLoadExecutor::execute()
LOG_WARN("fail to handle all task result", KR(ret));
}
}
}
LOG_TRACE("large file load executor init done", K(ret));
return ret;
}

68
src/sql/engine/cmd/ob_load_data_direct_impl.h
View File

@ -9,7 +9,6 @@
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#pragma once
#include "lib/allocator/page_arena.h"
@ -20,6 +19,7 @@
#include "share/table/ob_table_load_define.h"
#include "sql/engine/cmd/ob_load_data_impl.h"
#include "sql/engine/cmd/ob_load_data_parser.h"
#include "sql/engine/cmd/ob_load_data_file_reader.h"
#include "common/storage/ob_io_device.h"
#include "observer/table_load/ob_table_load_exec_ctx.h"
#include "observer/table_load/ob_table_load_instance.h"
@ -170,61 +170,6 @@ private:
int64_t pos_;
};
class IRandomIODevice
{
public:
virtual ~IRandomIODevice() = default;
virtual int open(const DataAccessParam &data_access_param, const ObString &filename) = 0;
virtual int pread(char *buf, int64_t count, int64_t offset, int64_t &read_size) = 0;
virtual int get_file_size(int64_t &file_size) = 0;
};
class RandomFileReader : public IRandomIODevice
{
public:
RandomFileReader();
virtual ~RandomFileReader();
int open(const DataAccessParam &data_access_param, const ObString &filename) override;
int pread(char *buf, int64_t count, int64_t offset, int64_t &read_size) override;
int get_file_size(int64_t &file_size) override;
private:
ObString filename_;
ObFileReader file_reader_;
bool is_inited_;
};
class RandomOSSReader : public IRandomIODevice
{
public:
RandomOSSReader();
virtual ~RandomOSSReader();
int open(const DataAccessParam &data_access_param, const ObString &filename) override;
int pread(char *buf, int64_t count, int64_t offset, int64_t &read_size) override;
int get_file_size(int64_t &file_size) override;
private:
ObIODevice *device_handle_;
ObIOFd fd_;
bool is_inited_;
};
class SequentialDataAccessor
{
public:
SequentialDataAccessor();
~SequentialDataAccessor();
int init(const DataAccessParam &data_access_param, const ObString &filename);
int read(char *buf, int64_t count, int64_t &read_size);
int get_file_size(int64_t &file_size);
void seek(int64_t offset) { offset_ = offset; }
int64_t get_offset() const { return offset_; }
private:
RandomFileReader random_file_reader_;
RandomOSSReader random_oss_reader_;
IRandomIODevice *random_io_device_;
int64_t offset_;
bool is_inited_;
};
struct DataBuffer
{
public:
@ -256,20 +201,25 @@ private:
{
public:
DataReader();
~DataReader();
int init(const DataAccessParam &data_access_param, LoadExecuteContext &execute_ctx,
const DataDesc &data_desc, bool read_raw = false);
int get_next_buffer(ObLoadFileBuffer &file_buffer, int64_t &line_count,
int64_t limit = INT64_MAX);
int get_next_raw_buffer(DataBuffer &data_buffer);
bool has_incomplate_data() const { return data_trimer_.has_incomplate_data(); }
bool is_end_file() const { return io_accessor_.get_offset() >= end_offset_; }
bool is_end_file() const;
ObCSVGeneralParser &get_csv_parser() { return csv_parser_; }
private:
int read_buffer(ObLoadFileBuffer &file_buffer);
private:
LoadExecuteContext *execute_ctx_;
ObCSVGeneralParser csv_parser_; // 用来计算完整行
ObLoadFileDataTrimer data_trimer_; // 缓存不完整行的数据
SequentialDataAccessor io_accessor_;
int64_t end_offset_;
ObFileReader *file_reader_;
int64_t end_offset_; // use -1 in stream file such as load data local
bool read_raw_;
bool is_iter_end_;
bool is_inited_;

3
src/sql/engine/cmd/ob_load_data_executor.cpp
View File

@ -38,6 +38,7 @@ int ObLoadDataExecutor::check_is_direct_load(ObTableDirectInsertCtx &ctx, const
} else {
ctx.set_is_direct(false);
}
LOG_INFO("check load data is direct done.", K(ctx.get_is_direct()));
return ret;
}
@ -70,6 +71,8 @@ int ObLoadDataExecutor::execute(ObExecContext &ctx, ObLoadDataStmt &stmt)
if (OB_SUCC(ret)) {
if (OB_FAIL(load_impl->execute(ctx, stmt))) {
LOG_WARN("failed to execute load data stmt", K(ret));
} else {
LOG_TRACE("load data success");
}
load_impl->~ObLoadDataBase();
}

458
src/sql/engine/cmd/ob_load_data_file_reader.cpp Normal file
View File

@ -0,0 +1,458 @@
/**
* 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.
*/
#define USING_LOG_PREFIX SQL_ENG
#include "sql/engine/cmd/ob_load_data_file_reader.h"
#include "share/ob_device_manager.h"
#include "share/backup/ob_backup_io_adapter.h"
#include "rpc/obmysql/ob_i_cs_mem_pool.h"
#include "rpc/obmysql/ob_mysql_packet.h"
#include "rpc/obmysql/packet/ompk_local_infile.h"
namespace oceanbase
{
namespace sql
{
/**
* ObFileReadParam
*/
ObFileReadParam::ObFileReadParam()
: packet_handle_(NULL),
session_(NULL),
timeout_ts_(-1)
{
}
/**
* ObFileReader
*/
int ObFileReader::open(const ObFileReadParam &param, ObIAllocator &allocator, ObFileReader *& file_reader)
{
int ret = OB_SUCCESS;
if (param.file_location_ == ObLoadFileLocation::SERVER_DISK) {
ObRandomFileReader *tmp_reader = OB_NEWx(ObRandomFileReader, &allocator, allocator);
if (OB_ISNULL(tmp_reader)) {
LOG_WARN("failed to create ObRandomFileReader", K(ret));
} else if (OB_FAIL(tmp_reader->open(param.filename_))) {
LOG_WARN("fail to open random file reader", KR(ret), K(param.filename_));
tmp_reader->~ObRandomFileReader();
allocator.free(tmp_reader);
} else {
file_reader = tmp_reader;
}
} else if (param.file_location_ == ObLoadFileLocation::OSS) {
ObRandomOSSReader *tmp_reader = OB_NEWx(ObRandomOSSReader, &allocator, allocator);
if (OB_ISNULL(tmp_reader)) {
LOG_WARN("failed to create RandomOSSReader", K(ret));
} else if (OB_FAIL(tmp_reader->open(param.access_info_, param.filename_))) {
LOG_WARN("fail to open random oss reader", KR(ret), K(param.filename_));
tmp_reader->~ObRandomOSSReader();
allocator.free(tmp_reader);
} else {
file_reader = tmp_reader;
}
} else if (param.file_location_ == ObLoadFileLocation::CLIENT_DISK) {
if (OB_ISNULL(param.packet_handle_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("cannot create packet stream file reader while the packet handle is null", K(ret));
} else {
ObPacketStreamFileReader *tmp_reader = OB_NEWx(ObPacketStreamFileReader, &allocator, allocator);
if (OB_ISNULL(tmp_reader)) {
LOG_WARN("failed to create ObPacketStreamFileReader", K(ret));
} else if (OB_FAIL(tmp_reader->open(param.filename_, *param.packet_handle_, param.session_, param.timeout_ts_))) {
LOG_WARN("failed to open packet stream file reader", KR(ret), K(param.filename_));
tmp_reader->~ObPacketStreamFileReader();
allocator.free(tmp_reader);
} else {
file_reader = tmp_reader;
}
}
} else {
ret = OB_NOT_SUPPORTED;
LOG_WARN("not supported load file location", KR(ret), K(param.file_location_));
}
return ret;
}
int ObFileReader::readn(char *buffer, int64_t count, int64_t &read_size)
{
int ret = OB_SUCCESS;
read_size = 0;
while (OB_SUCC(ret) && !eof() && read_size < count) {
int64_t this_read_size = 0;
ret = this->read(buffer + read_size, count - read_size, this_read_size);
if (OB_SUCC(ret)) {
read_size += this_read_size;
}
}
return ret;
}
/**
* ObRandomFileReader
*/
ObRandomFileReader::ObRandomFileReader(ObIAllocator &allocator)
: ObFileReader(allocator),
offset_(0),
eof_(false),
is_inited_(false)
{
}
ObRandomFileReader::~ObRandomFileReader()
{
}
int ObRandomFileReader::open(const ObString &filename)
{
int ret = OB_SUCCESS;
if (is_inited_) {
ret = OB_INIT_TWICE;
LOG_WARN("ObRandomFileReader init twice", KR(ret), KP(this));
} else if (OB_FAIL(file_reader_.open(filename.ptr(), false))) {
LOG_WARN("fail to open file", KR(ret), K(filename));
} else {
filename_ = filename;
offset_ = 0;
eof_ = false;
is_inited_ = true;
}
return ret;
}
int ObRandomFileReader::read(char *buf, int64_t count, int64_t &read_size)
{
int ret = OB_SUCCESS;
if (!is_inited_) {
ret = OB_NOT_INIT;
LOG_WARN("ObRandomFileReader not init", KR(ret), KP(this));
} else if (OB_FAIL(file_reader_.pread(buf, count, offset_, read_size))) {
LOG_WARN("fail to pread file buf", KR(ret), K(count), K_(offset), K(read_size));
} else if (0 == read_size) {
eof_ = true;
} else {
offset_ += read_size;
}
return ret;
}
int ObRandomFileReader::seek(int64_t offset)
{
offset_ = offset;
return OB_SUCCESS;
}
int ObRandomFileReader::get_file_size(int64_t &file_size)
{
int ret = OB_SUCCESS;
if (!is_inited_) {
ret = OB_NOT_INIT;
LOG_WARN("ObRandomFileReader not init", KR(ret), KP(this));
} else {
file_size = ::get_file_size(filename_.ptr());
}
return ret;
}
/**
* ObRandomOSSReader
*/
ObRandomOSSReader::ObRandomOSSReader(ObIAllocator &allocator)
: ObFileReader(allocator),
device_handle_(nullptr),
offset_(0),
eof_(false),
is_inited_(false)
{
}
ObRandomOSSReader::~ObRandomOSSReader()
{
if (fd_.is_valid()) {
device_handle_->close(fd_);
fd_.reset();
}
if (nullptr != device_handle_) {
common::ObDeviceManager::get_instance().release_device(device_handle_);
device_handle_ = nullptr;
}
is_inited_ = false;
}
int ObRandomOSSReader::open(const share::ObBackupStorageInfo &storage_info, const ObString &filename)
{
int ret = OB_SUCCESS;
ObIODOpt opt;
ObIODOpts iod_opts;
ObBackupIoAdapter util;
iod_opts.opts_ = &opt;
iod_opts.opt_cnt_ = 0;
if (IS_INIT) {
ret = OB_INIT_TWICE;
LOG_WARN("ObRandomOSSReader init twice", KR(ret), KP(this));
} else if (OB_FAIL(
util.get_and_init_device(device_handle_, &storage_info, filename))) {
LOG_WARN("fail to get device manager", KR(ret), K(filename));
} else if (OB_FAIL(util.set_access_type(&iod_opts, false, 1))) {
LOG_WARN("fail to set access type", KR(ret));
} else if (OB_FAIL(device_handle_->open(to_cstring(filename), -1, 0, fd_, &iod_opts))) {
LOG_WARN("fail to open oss file", KR(ret), K(filename));
} else {
offset_ = 0;
eof_ = false;
is_inited_ = true;
}
return ret;
}
int ObRandomOSSReader::read(char *buf, int64_t count, int64_t &read_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("ObRandomOSSReader not init", KR(ret), KP(this));
} else if (OB_FAIL(device_handle_->pread(fd_, offset_, count, buf, read_size))) {
LOG_WARN("fail to pread oss buf", KR(ret), K_(offset), K(count), K(read_size));
} else if (0 == read_size) {
eof_ = true;
} else {
offset_ += read_size;
}
return ret;
}
int ObRandomOSSReader::seek(int64_t offset)
{
offset_ = offset;
return OB_SUCCESS;
}
int ObRandomOSSReader::get_file_size(int64_t &file_size)
{
int ret = OB_SUCCESS;
ObBackupIoAdapter util;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("ObRandomOSSReader not init", KR(ret), KP(this));
} else if (OB_FAIL(util.get_file_size(device_handle_, fd_, file_size))) {
LOG_WARN("fail to get oss file size", KR(ret), K(file_size));
}
return ret;
}
/**
* ObPacketStreamFileReader
*/
class CSMemPoolAdaptor : public obmysql::ObICSMemPool
{
public:
explicit CSMemPoolAdaptor(ObIAllocator *allocator)
: allocator_(allocator)
{}
virtual ~CSMemPoolAdaptor() {}
void *alloc(int64_t size) override
{
return allocator_->alloc(size);
}
private:
ObIAllocator *allocator_;
};
ObPacketStreamFileReader::ObPacketStreamFileReader(ObIAllocator &allocator)
: ObStreamFileReader(allocator),
packet_handle_(NULL),
session_(NULL),
timeout_ts_(INT64_MAX),
arena_allocator_(allocator),
cached_packet_(NULL),
received_size_(0),
read_size_(0),
eof_(false)
{
}
ObPacketStreamFileReader::~ObPacketStreamFileReader()
{
int ret = OB_SUCCESS;
// We read all data from client before close the file.
// We will stop to handle the process while something error.
// But the client must send all file content to us and the
// normal SQL processor cannot handle the packets, so we
// eat all packets with file content.
timeout_ts_ = -1;
while (!eof_ && OB_SUCC(ret)) {
ret = receive_packet();
}
arena_allocator_.reset();
}
int ObPacketStreamFileReader::open(const ObString &filename,
observer::ObIMPPacketSender &packet_handle,
ObSQLSessionInfo *session,
int64_t timeout_ts)
{
int ret = OB_SUCCESS;
if (OB_NOT_NULL(packet_handle_)) {
ret = OB_INIT_TWICE;
} else {
// in `load data local` request, we should send the filename to client
obmysql::OMPKLocalInfile filename_packet;
filename_packet.set_filename(filename);
if (OB_FAIL(packet_handle.response_packet(filename_packet, session))) {
LOG_INFO("failed to send local infile packet to client", K(ret), K(filename));
} else if (OB_FAIL(packet_handle.flush_buffer(false/*is_last*/))) {
LOG_INFO("failed to flush socket buffer while send local infile packet", K(ret), K(filename));
} else {
LOG_TRACE("send filename to client success", K(filename));
observer::ObSMConnection *sm_connection = session->get_sm_connection();
if (OB_NOT_NULL(sm_connection) &&
sm_connection->pkt_rec_wrapper_.enable_proto_dia()) {
sm_connection->pkt_rec_wrapper_.record_send_mysql_pkt(filename_packet,
filename_packet.get_serialize_size() + OB_MYSQL_HEADER_LENGTH);
}
}
packet_handle_ = &packet_handle;
session_ = session;
timeout_ts_ = timeout_ts;
received_size_ = 0;
read_size_ = 0;
eof_ = false;
}
return ret;
}
/**
* As decripted in MySQL/MariaDB document, client send the file content with
* continous packets and `eof` with an empty packet. Every non-empty packet
* has the format:
* -------------------
* MySQL Packet Header
* string<EOF>
* -------------------
* The notation is "string<EOF>" Strings whose length will be calculated by
* the packet remaining length.
*/
int ObPacketStreamFileReader::read(char *buf, int64_t count, int64_t &read_size)
{
int ret = OB_SUCCESS;
if (OB_ISNULL(cached_packet_) || read_size_ == received_size_) {
ret = receive_packet();
}
const int64_t remain_in_packet = received_size_ - read_size_;
if (OB_SUCC(ret) && OB_NOT_NULL(cached_packet_) && (!eof_ || remain_in_packet > 0)) {
read_size = MIN(count, remain_in_packet);
// a MySQL packet contains a header and payload. The payload is the file content here.
// In the mysql_packet code, it use the first byte as MySQL command, but there is no
// MySQL command in the file content packet, so we backward 1 byte.
const int64_t packet_offset = cached_packet_->get_pkt_len() - remain_in_packet;
MEMCPY(buf, cached_packet_->get_cdata() - 1 + packet_offset, read_size);
read_size_ += read_size;
} else {
read_size = 0;
}
if (is_timeout()) {
ret = OB_TIMEOUT;
LOG_WARN("load data reader file timeout", KR(ret));
} else if (session_ != NULL && session_->is_query_killed()) {
ret = OB_ERR_QUERY_INTERRUPTED;
LOG_WARN("load data reader terminated as the query is killed", KR(ret));
} else if (session_ != NULL && session_->is_zombie()) {
ret = OB_SESSION_KILLED;
LOG_WARN("load data reader terminated as the session is killed", KR(ret));
} else if (!eof_ && read_size == 0) {
ret = OB_IO_ERROR;
LOG_WARN("[should not happen] cannot read data but eof is false", KR(ret));
}
return ret;
}
int ObPacketStreamFileReader::receive_packet()
{
int ret = OB_SUCCESS;
ret = release_packet();
if (OB_SUCC(ret)) {
arena_allocator_.reuse();
CSMemPoolAdaptor mem_pool(&arena_allocator_);
// We read packet until we got one or timeout or error occurs
obmysql::ObMySQLPacket *pkt = NULL;
ret = packet_handle_->read_packet(mem_pool, pkt);
cached_packet_ = static_cast<obmysql::ObMySQLRawPacket *>(pkt);
while (OB_SUCC(ret) && OB_ISNULL(cached_packet_) && !is_timeout() && !is_killed()) {
// sleep can reduce cpu usage while the network is not so good.
// We need not worry about the speed while the speed of load data core is lower than
// file receiver's.
usleep(100 * 1000); // 100 ms
ret = packet_handle_->read_packet(mem_pool, pkt);
cached_packet_ = static_cast<obmysql::ObMySQLRawPacket *>(pkt);
}
if (OB_SUCC(ret) && OB_NOT_NULL(cached_packet_)) {
const int pkt_len = cached_packet_->get_pkt_len();
if (0 == pkt_len) { // empty packet
eof_ = true;
(void)release_packet();
} else {
received_size_ += pkt_len;
LOG_TRACE("got a packet", K(pkt_len));
}
}
}
// If anything wrong, we end the reading
if (OB_FAIL(ret)) {
eof_ = true;
}
return ret;
}
int ObPacketStreamFileReader::release_packet()
{
int ret = OB_SUCCESS;
if (OB_NOT_NULL(cached_packet_)) {
ret = packet_handle_->release_packet(cached_packet_);
cached_packet_ = NULL;
}
return ret;
}
bool ObPacketStreamFileReader::is_timeout() const
{
return timeout_ts_ != -1 && ObTimeUtility::current_time() >= timeout_ts_;
}
bool ObPacketStreamFileReader::is_killed() const
{
return NULL != session_ && (session_->is_query_killed() || session_->is_zombie());
}
} // namespace sql
} // namespace oceanbase

205
src/sql/engine/cmd/ob_load_data_file_reader.h Normal file
View File

@ -0,0 +1,205 @@
/**
* 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.
*/
#ifndef OCEANBASE_SQL_LOAD_DATA_FILE_READER_H_
#define OCEANBASE_SQL_LOAD_DATA_FILE_READER_H_
#include "lib/oblog/ob_log_module.h"
#include "lib/string/ob_string.h"
#include "lib/allocator/ob_allocator.h"
#include "lib/file/ob_file.h"
#include "sql/resolver/cmd/ob_load_data_stmt.h"
#include "share/backup/ob_backup_struct.h"
#include "observer/mysql/obmp_packet_sender.h"
namespace oceanbase
{
namespace sql
{
class ObSQLSessionInfo;
struct ObFileReadParam
{
public:
ObFileReadParam();
TO_STRING_KV(K_(file_location), K_(filename), K_(timeout_ts));
public:
ObLoadFileLocation file_location_;
ObString filename_;
share::ObBackupStorageInfo access_info_;
observer::ObIMPPacketSender *packet_handle_;
ObSQLSessionInfo *session_;
int64_t timeout_ts_; // A job always has a deadline and file reading may cost a long time
};
class ObFileReader
{
public:
ObFileReader(ObIAllocator &allocator) : allocator_(allocator) {}
virtual ~ObFileReader() {}
/**
* read data from file into the buffer
*
* @note read_size equals to 0 does not mean end of file.
* You should call `eof` to decide whether end of file.
* This is not the same with the system call `read`.
*/
virtual int read(char *buf, int64_t count, int64_t &read_size) = 0;
/**
* get the file size
*
* Stream files may not support this feature.
*/
virtual int get_file_size(int64_t &file_size) = 0;
/**
* seek to the specific position and the `read` subsequently fetch data from the position
*
* You can use `seekable` to check whether this file can read at random position.
*/
virtual int seek(int64_t offset) = 0;
virtual bool seekable() const { return true; }
virtual int64_t get_offset() const = 0;
virtual bool eof() const = 0;
/**
* read data until we got `count` bytes data or exception occurs
*
* This routine calls `read` repeatly until we got `count` bytes
* data.
* As usual, the normal `read` try to read data once and return.
*/
int readn(char *buffer, int64_t count, int64_t &read_size);
/**
* A file reader factory
*/
static int open(const ObFileReadParam &param, ObIAllocator &allocator, ObFileReader *& file_reader);
protected:
ObIAllocator &allocator_;
};
/**
* Stream file that can read sequential only
*/
class ObStreamFileReader : public ObFileReader
{
public:
ObStreamFileReader(ObIAllocator &allocator): ObFileReader(allocator) {}
virtual ~ObStreamFileReader() {}
int get_file_size(int64_t &file_size) override { return OB_NOT_SUPPORTED; }
int seek(int64_t offset) override { return OB_NOT_SUPPORTED; }
bool seekable() const override { return false; }
};
class ObRandomFileReader : public ObFileReader
{
public:
ObRandomFileReader(ObIAllocator &allocator);
virtual ~ObRandomFileReader();
int read(char *buf, int64_t count, int64_t &read_size) override;
int seek(int64_t offset) override;
int get_file_size(int64_t &file_size) override;
int64_t get_offset() const override { return offset_; }
bool eof() const override { return eof_; }
int open(const ObString &filename);
private:
ObString filename_;
common::ObFileReader file_reader_;
int64_t offset_;
bool eof_;
bool is_inited_;
};
class ObRandomOSSReader : public ObFileReader
{
public:
ObRandomOSSReader(ObIAllocator &allocator);
virtual ~ObRandomOSSReader();
int open(const share::ObBackupStorageInfo &storage_info, const ObString &filename);
int read(char *buf, int64_t count, int64_t &read_size) override;
int seek(int64_t offset) override;
int get_file_size(int64_t &file_size) override;
int64_t get_offset() const override { return offset_; }
bool eof() const override { return eof_; }
private:
ObIODevice *device_handle_;
ObIOFd fd_;
int64_t offset_;
bool eof_;
bool is_inited_;
};
/**
* A strem file reader whose data source is mysql packets
* Refer to LOAD DATA LOCAL INFILE for more detail.
* Read data flow:
* client send file content through mysql packets
* (@see PacketStreamFileReader::read) and end with an
* empty mysql packet.
*/
class ObPacketStreamFileReader : public ObStreamFileReader
{
public:
ObPacketStreamFileReader(ObIAllocator &allocator);
virtual ~ObPacketStreamFileReader();
int open(const ObString &filename,
observer::ObIMPPacketSender &packet_handle,
ObSQLSessionInfo *session,
int64_t timeout_ts);
int read(char *buf, int64_t count, int64_t &read_size) override;
int64_t get_offset() const override { return read_size_; }
bool eof() const override { return eof_; }
private:
int receive_packet();
/// The packet read from NIO is cached, so we must release it explicitly
/// and then we can reuse the resource
int release_packet();
bool is_timeout() const;
bool is_killed() const;
private:
observer::ObIMPPacketSender *packet_handle_; // We use this handle to read packet from client
ObSQLSessionInfo *session_;
int64_t timeout_ts_; // The deadline of job
// As we read a packet from client, the NIO store the data into the NIO buffer
// and allocate an ObPacket by an allocator(arena_allocator_). The ObPacket(cached_packet_)
// is cached in the memory of allocator.
ObArenaAllocator arena_allocator_;
obmysql::ObMySQLRawPacket *cached_packet_;
int64_t received_size_; // All data received in bytes
int64_t read_size_; // All data has been read in bytes
bool eof_;
};
} // namespace sql
} // namespace oceanbase
#endif // OCEANBASE_SQL_LOAD_DATA_FILE_READER_H_

63
src/sql/engine/cmd/ob_load_data_impl.cpp
View File

@ -1388,23 +1388,12 @@ int ObLoadDataSPImpl::next_file_buffer(ObExecContext &ctx,
//从data_trimer中恢复出上次读取剩下的数据
OZ (box.data_trimer.recover_incomplate_data(*handle->data_buffer));
if (ObLoadFileLocation::SERVER_DISK == box.load_file_storage) {
OZ (box.file_reader.pread(handle->data_buffer->current_ptr(),
handle->data_buffer->get_remain_len(),
box.read_cursor.file_offset_,
box.read_cursor.read_size_));
} else {
OZ (box.device_handle_->pread(box.fd_, box.read_cursor.file_offset_,
handle->data_buffer->get_remain_len(),
handle->data_buffer->current_ptr(),
box.read_cursor.read_size_));
}
OZ (box.file_reader->readn(handle->data_buffer->current_ptr(),
handle->data_buffer->get_remain_len(),
box.read_cursor.read_size_));
if (OB_SUCC(ret)) {
if (OB_UNLIKELY(0 == box.read_cursor.read_size_)) {
box.read_cursor.is_end_file_ = true;
LOG_DEBUG("LOAD DATA reach file end", K(box.read_cursor));
} else {
if (OB_LIKELY(box.read_cursor.read_size_ > 0)) {
handle->data_buffer->update_pos(box.read_cursor.read_size_); //更新buffer中数据长度
int64_t last_proccessed_GBs = box.read_cursor.get_total_read_GBs();
box.read_cursor.commit_read();
@ -1414,6 +1403,9 @@ int ObLoadDataSPImpl::next_file_buffer(ObExecContext &ctx,
}
box.job_status->read_bytes_ += box.read_cursor.read_size_;
} else if (box.file_reader->eof()) {
box.read_cursor.is_end_file_ = true;
LOG_DEBUG("LOAD DATA reach file end", K(box.read_cursor));
}
}
@ -2465,12 +2457,10 @@ int ObLoadDataSPImpl::ToolBox::release_resources()
ob_free(expr_buffer);
}
//release fd and device
if (NULL != device_handle_) {
if (fd_.is_valid()) {
device_handle_->close(fd_);
}
common::ObDeviceManager::get_instance().release_device(device_handle_);
//release file reader
if (OB_NOT_NULL(file_reader)) {
file_reader->~ObFileReader();
file_reader = NULL;
}
if (OB_NOT_NULL(temp_handle)) {
@ -2708,12 +2698,6 @@ int ObLoadDataSPImpl::ToolBox::init(ObExecContext &ctx, ObLoadDataStmt &load_stm
LOG_WARN("fail to gen insert column names buff", K(ret));
} else if (OB_FAIL(data_frag_mgr.init(ctx, load_args.table_id_))) {
LOG_WARN("fail to init data frag mgr", K(ret));
} else if (ObLoadFileLocation::SERVER_DISK != load_file_storage) {
if (OB_FAIL(util.get_and_init_device(device_handle_, &load_args.access_info_, load_args.file_name_))) {
LOG_WARN("fail to get device manager", K(ret), K(load_args.access_info_), K(load_args.file_name_));
} else if (OB_FAIL(util.set_access_type(&iod_opts, false, 1))) {
LOG_WARN("fail to set access type", K(ret));
}
}
//init server_info_map
@ -2779,14 +2763,20 @@ int ObLoadDataSPImpl::ToolBox::init(ObExecContext &ctx, ObLoadDataStmt &load_stm
}
if (OB_SUCC(ret)) {
if (ObLoadFileLocation::SERVER_DISK == load_file_storage) {
OZ (file_reader.open(load_args.file_name_, false));
OX (file_size = get_file_size(load_args.file_name_.ptr()));
} else {
int64_t file_length = -1;
OZ (device_handle_->open(load_args.file_name_.ptr(), -1, 0, fd_, &iod_opts));
OZ (util.get_file_size(device_handle_, fd_, file_length));
OX (file_size = file_length);
file_read_param.file_location_ = load_file_storage;
file_read_param.filename_ = load_args.file_name_;
file_read_param.access_info_ = load_args.access_info_;
file_read_param.packet_handle_ = &ctx.get_my_session()->get_pl_query_sender()->get_packet_sender();
file_read_param.session_ = ctx.get_my_session();
file_read_param.timeout_ts_ = THIS_WORKER.get_timeout_ts();
if (OB_FAIL(ObFileReader::open(file_read_param, ctx.get_allocator(), file_reader))) {
LOG_WARN("failed to open file.", KR(ret), K(file_read_param), K(load_args.file_name_));
} else if (!file_reader->seekable()) {
file_size = -1;
} else if (OB_FAIL(file_reader->get_file_size(file_size))) {
LOG_WARN("fail to get io device file size", KR(ret), K(file_size));
}
}
@ -3075,7 +3065,8 @@ int ObLoadDataSPImpl::ToolBox::init(ObExecContext &ctx, ObLoadDataStmt &load_stm
}
if (OB_SUCC(ret)) {
int64_t max_task_count = (file_size / ObLoadFileBuffer::MAX_BUFFER_SIZE + 1) * 2;
const int64_t fake_file_size = (file_size > 0) ? file_size : (2 << 30); // use 2G as default in load local mode
int64_t max_task_count = (fake_file_size / ObLoadFileBuffer::MAX_BUFFER_SIZE + 1) * 2;
if (OB_FAIL(file_buf_row_num.reserve(max_task_count))) {
LOG_WARN("fail to reserve", K(ret));
}

8
src/sql/engine/cmd/ob_load_data_impl.h
View File

@ -31,6 +31,7 @@
#include "sql/engine/cmd/ob_load_data_rpc.h"
#include "sql/engine/ob_des_exec_context.h"
#include "sql/engine/cmd/ob_load_data_parser.h"
#include "sql/engine/cmd/ob_load_data_file_reader.h"
#include "common/storage/ob_io_device.h"
namespace oceanbase
@ -684,7 +685,7 @@ public:
InsertTask,
};
struct ToolBox {
ToolBox() : device_handle_(NULL), fd_(), expr_buffer(nullptr) {}
ToolBox() : expr_buffer(nullptr) {}
int init(ObExecContext &ctx, ObLoadDataStmt &load_stmt);
int build_calc_partid_expr(ObExecContext &ctx,
ObLoadDataStmt &load_stmt,
@ -692,14 +693,13 @@ public:
int release_resources();
//modules
ObFileReader file_reader;
ObIODevice* device_handle_;
ObIOFd fd_;
ObFileReader * file_reader;
ObFileAppender file_appender;
ObFileReadCursor read_cursor;
ObLoadFileDataTrimer data_trimer;
ObInsertValueGenerator generator;
ObDataFragMgr data_frag_mgr;
ObFileReadParam file_read_param;
//running control
ObParallelTaskController shuffle_task_controller;

218
src/sql/resolver/cmd/ob_load_data_resolver.cpp
View File

@ -92,92 +92,31 @@ int ObLoadDataResolver::resolve(const ParseNode &parse_tree)
case T_REMOTE_OSS:
load_args.load_file_storage_ = ObLoadFileLocation::OSS;
break;
case T_LOCAL:
//load_args.load_file_storage_ = ObLoadFileLocation::CLIENT_DISK;
//break;
//not support local
case T_LOCAL: {
bool enabled = false;
if (OB_FAIL(local_infile_enabled(enabled))) {
LOG_WARN("failed to check local_infile_enabled", K(ret));
} else if (!enabled) {
ret = OB_ERR_CLIENT_LOCAL_FILES_DISABLED;
LOG_USER_ERROR(OB_ERR_CLIENT_LOCAL_FILES_DISABLED);
} else {
load_args.load_file_storage_ = ObLoadFileLocation::CLIENT_DISK;
}
}
break;
default:
ret = OB_NOT_SUPPORTED;
LOG_USER_ERROR(OB_NOT_SUPPORTED, "load data local");
LOG_USER_ERROR(OB_NOT_SUPPORTED, "unknown location");
}
} else {
load_args.load_file_storage_ = ObLoadFileLocation::SERVER_DISK;
}
LOG_DEBUG("load data location", K(load_args.load_file_storage_));
}
if (OB_SUCC(ret)) {
/* 1. file name */
ObLoadArgument &load_args = load_stmt->get_load_arguments();
ParseNode *file_name_node = node->children_[ENUM_FILE_NAME];
if (OB_ISNULL(file_name_node)
|| OB_UNLIKELY(T_VARCHAR != file_name_node->type_ && T_CHAR != file_name_node->type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid node", "child", file_name_node);
} else {
ObString file_name(file_name_node->str_len_, file_name_node->str_value_);
if (ObLoadFileLocation::OSS != load_args.load_file_storage_) {
load_args.file_name_ = file_name;
const char *p = nullptr;
ObString sub_file_name;
ObString cstyle_file_name; // ends with '\0'
char *full_path_buf = nullptr;
char *actual_path = nullptr;
if (OB_ISNULL(full_path_buf = static_cast<char *>(allocator_->alloc(MAX_PATH_SIZE)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("fail to allocate memory", K(ret));
}
while (OB_SUCC(ret) && !file_name.empty()) {
p = file_name.find(',');
if (nullptr == p) {
sub_file_name = file_name;
cstyle_file_name = sub_file_name;
file_name.reset();
} else {
sub_file_name = file_name.split_on(p);
cstyle_file_name.reset();
}
if (!sub_file_name.empty()) {
if (cstyle_file_name.empty() &&
OB_FAIL(ob_write_string(*allocator_, sub_file_name, cstyle_file_name, true))) {
LOG_WARN("fail to write string", KR(ret));
} else if (OB_ISNULL(actual_path = realpath(cstyle_file_name.ptr(), full_path_buf))) {
ret = OB_FILE_NOT_EXIST;
LOG_WARN("file not exist", K(ret), K(cstyle_file_name));
}
if (OB_SUCC(ret)) {
ObString secure_file_priv;
if (OB_FAIL(session_info_->get_secure_file_priv(secure_file_priv))) {
LOG_WARN("failed to get secure file priv", K(ret));
} else if (OB_FAIL(
ObResolverUtils::check_secure_path(secure_file_priv, actual_path))) {
LOG_WARN("failed to check secure path", K(ret), K(secure_file_priv),
K(actual_path));
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(load_args.file_iter_.add_files(&cstyle_file_name))) {
LOG_WARN("fail to add files", KR(ret));
}
}
}
}
} else {
ObString temp_file_name = file_name.split_on('?');
ObString storage_info;
if (OB_FAIL(ob_write_string(*allocator_, temp_file_name, load_args.file_name_, true))) {
LOG_WARN("fail to copy string", K(ret));
} else if (OB_FAIL(ob_write_string(*allocator_, file_name, storage_info, true))) {
LOG_WARN("fail to copy string", K(ret));
} else if (temp_file_name.length() <= 0 || storage_info.length() <= 0) {
ret = OB_INVALID_ARGUMENT;
LOG_USER_ERROR(OB_INVALID_ARGUMENT, "file name or access key");
} else if (OB_FAIL(load_args.access_info_.set(load_args.file_name_.ptr(), storage_info.ptr()))) {
LOG_WARN("failed to set access info", K(ret));
} else if (OB_FAIL(load_args.file_iter_.add_files(&load_args.file_name_))) {
LOG_WARN("fail to add files", KR(ret));
}
}
}
ret = resolve_filename(load_stmt, node);
}
if (OB_SUCC(ret)) {
@ -185,7 +124,10 @@ int ObLoadDataResolver::resolve(const ParseNode &parse_tree)
ObLoadArgument &load_args = load_stmt->get_load_arguments();
ObLoadDupActionType dupl_action = ObLoadDupActionType::LOAD_STOP_ON_DUP;
if (NULL == node->children_[ENUM_DUPLICATE_ACTION]) {
if (ObLoadFileLocation::CLIENT_DISK == load_args.load_file_storage_) {
if (ObLoadFileLocation::CLIENT_DISK == load_args.load_file_storage_ &&
lib::is_mysql_mode()) {
// https://dev.mysql.com/doc/refman/8.0/en/load-data.html
// In MySQL, LOCAL modifier has the same effect as the IGNORE modifier.
dupl_action = ObLoadDupActionType::LOAD_IGNORE;
}
} else if (T_IGNORE == node->children_[ENUM_DUPLICATE_ACTION]->type_) {
@ -561,6 +503,92 @@ int ObLoadDataResolver::resolve_hints(const ParseNode &node)
return ret;
}
int ObLoadDataResolver::resolve_filename(ObLoadDataStmt *load_stmt, ParseNode *node)
{
int ret = OB_SUCCESS;
ObLoadArgument &load_args = load_stmt->get_load_arguments();
ParseNode *file_name_node = node->children_[ENUM_FILE_NAME];
if (OB_ISNULL(file_name_node)
|| OB_UNLIKELY(T_VARCHAR != file_name_node->type_ && T_CHAR != file_name_node->type_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid node", "child", file_name_node);
} else {
ObString file_name(file_name_node->str_len_, file_name_node->str_value_);
if (ObLoadFileLocation::OSS != load_args.load_file_storage_) {
load_args.file_name_ = file_name;
const char *p = nullptr;
ObString sub_file_name;
ObString cstyle_file_name; // ends with '\0'
char *full_path_buf = nullptr;
char *actual_path = nullptr;
if (OB_ISNULL(full_path_buf = static_cast<char *>(allocator_->alloc(MAX_PATH_SIZE)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("fail to allocate memory", K(ret));
}
while (OB_SUCC(ret) && !file_name.empty()) {
p = file_name.find(',');
if (nullptr == p) {
sub_file_name = file_name;
cstyle_file_name = sub_file_name;
file_name.reset();
} else {
sub_file_name = file_name.split_on(p);
cstyle_file_name.reset();
}
if (!sub_file_name.empty()) {
if (cstyle_file_name.empty() &&
OB_FAIL(ob_write_string(*allocator_, sub_file_name, cstyle_file_name, true))) {
LOG_WARN("fail to write string", KR(ret));
} else if (ObLoadFileLocation::SERVER_DISK == load_args.load_file_storage_ &&
OB_ISNULL(actual_path = realpath(cstyle_file_name.ptr(), full_path_buf))) {
ret = OB_FILE_NOT_EXIST;
LOG_WARN("file not exist", K(ret), K(cstyle_file_name));
}
//security check for mysql mode
if (OB_SUCC(ret) && lib::is_mysql_mode() && ObLoadFileLocation::SERVER_DISK == load_args.load_file_storage_) {
ObString secure_file_priv;
if (OB_FAIL(session_info_->get_secure_file_priv(secure_file_priv))) {
LOG_WARN("failed to get secure file priv", K(ret));
} else if (OB_FAIL(
ObResolverUtils::check_secure_path(secure_file_priv, actual_path))) {
LOG_WARN("failed to check secure path", K(ret), K(secure_file_priv),
K(actual_path));
}
}
if (OB_SUCC(ret)) {
if (ObLoadFileLocation::CLIENT_DISK == load_args.load_file_storage_ && load_args.file_iter_.count() != 0) {
ret = OB_NOT_SUPPORTED;
LOG_USER_ERROR(OB_NOT_SUPPORTED, "load multi files not supported");
} else if (OB_FAIL(load_args.file_iter_.add_files(&cstyle_file_name))) {
LOG_WARN("fail to add files", KR(ret));
}
}
}
}
} else {
ObString temp_file_name = file_name.split_on('?');
ObString storage_info;
if (OB_FAIL(ob_write_string(*allocator_, temp_file_name, load_args.file_name_, true))) {
LOG_WARN("fail to copy string", K(ret));
} else if (OB_FAIL(ob_write_string(*allocator_, file_name, storage_info, true))) {
LOG_WARN("fail to copy string", K(ret));
} else if (temp_file_name.length() <= 0 || storage_info.length() <= 0) {
ret = OB_INVALID_ARGUMENT;
LOG_USER_ERROR(OB_INVALID_ARGUMENT, "file name or access key");
} else if (OB_FAIL(load_args.access_info_.set(load_args.file_name_.ptr(), storage_info.ptr()))) {
LOG_WARN("failed to set access info", K(ret));
} else if (OB_FAIL(load_args.file_iter_.add_files(&load_args.file_name_))) {
LOG_WARN("fail to add files", KR(ret));
}
}
}
return ret;
}
//validation for loaddata statement obeys the following rules:
//0. in loaddata Ver1, only ascii charset are supported.
//1. according to the defined charset, escaped and enclosed valid char length should <= 1.
@ -1349,6 +1377,42 @@ int ObLoadDataResolver::resolve_char_node(const ParseNode &node, int32_t &single
return ret;
}
int ObLoadDataResolver::local_infile_enabled(bool &enabled) const
{
int ret = OB_SUCCESS;
// 1. let's check the system variable and the capability flag in the mysql handshake
enabled = false;
int64_t local_infile_sys_var = 0;
if (OB_ISNULL(session_info_)) {
} else if (OB_FAIL(session_info_->get_sys_variable(share::SYS_VAR_LOCAL_INFILE, local_infile_sys_var))) {
LOG_WARN("failed to get SYS_VAR_LOCAL_INFILE system variable.", KR(ret));
} else {
const int64_t local_infile_capability_flag = session_info_->get_capability().cap_flags_.OB_CLIENT_LOCAL_FILES;
enabled = (local_infile_sys_var != 0) && (local_infile_capability_flag != 0);
LOG_DEBUG("LOCAL_INFILE enabled by system variable and client capability",
K(enabled), K(local_infile_capability_flag), K(local_infile_sys_var));
}
// 2. let's check the client type.
// The obproxy set the capability flag but it does not support load local
if (OB_SUCC(ret) && enabled) {
if (session_info_->get_client_mode() > common::OB_MIN_CLIENT_MODE &&
session_info_->get_client_mode() < OB_MAX_CLIENT_MODE) {
// this is an ob client, such as obclient 2.x, objdbc, obproxy, obclient 1.x is not included
// check the proxy capability flags
obmysql::ObProxyCapabilityFlags proxy_cap = session_info_->get_proxy_cap_flags();
LOG_DEBUG("load local infile: get proxy capability flag",
K(proxy_cap.capability_), K(proxy_cap.is_load_local_support()));
if (!proxy_cap.is_load_local_support()) {
enabled = false;
LOG_INFO("load data local infile is disabled by client: the obclient proxy capability flag is not set");
}
}
}
return ret;
}
int ObLoadDataResolver::check_trigger_constraint(const ObTableSchema *table_schema)
{
int ret = OB_SUCCESS;

4
src/sql/resolver/cmd/ob_load_data_resolver.h
View File

@ -57,6 +57,10 @@ public:
const common::ObString &table_name, bool cte_table_fisrt, uint64_t& table_id);
int validate_stmt(ObLoadDataStmt* stmt);
int resolve_hints(const ParseNode &node);
int resolve_filename(ObLoadDataStmt *load_stmt, ParseNode *node);
int local_infile_enabled(bool &enabled) const;
int check_trigger_constraint(const ObTableSchema *table_schema);
private:
enum ParameterEnum {

1
src/sql/session/ob_basic_session_info.h
View File

@ -1031,6 +1031,7 @@ public:
// client mode related
void set_client_mode(const common::ObClientMode mode) { client_mode_ = mode; }
common::ObClientMode get_client_mode() const { return client_mode_; }
bool is_java_client_mode() const { return common::OB_JAVA_CLIENT_MODE == client_mode_; }
bool is_obproxy_mode() const { return common::OB_PROXY_CLIENT_MODE == client_mode_; }

6
unittest/sql/engine/load_data/ob_load_data_parser_test.cpp
View File

@ -97,7 +97,7 @@ TEST_F(TestParser, csv_parser)
file_name = f_meta.file_name;
}
ObFileReader reader;
oceanbase::common::ObFileReader reader;
ASSERT_EQ(OB_SUCCESS, reader.open(file_name.c_str(), false));
int64_t file_size = get_file_size(file_name.c_str());
int64_t total_read_bytes = 0;
@ -169,7 +169,7 @@ TEST_F(TestParser, general_parser)
file_name = f_meta.file_name;
}
ObFileReader reader;
oceanbase::common::ObFileReader reader;
ASSERT_EQ(OB_SUCCESS, reader.open(file_name.c_str(), false));
int64_t file_size = get_file_size(file_name.c_str());
int64_t total_read_bytes = 0;
@ -252,7 +252,7 @@ TEST_F(TestParser, general_parser_escape)
file_name = f_meta.file_name;
}
ObFileReader reader;
oceanbase::common::ObFileReader reader;
ASSERT_EQ(OB_SUCCESS, reader.open(file_name.c_str(), false));
int64_t file_size = get_file_size(file_name.c_str());
int64_t total_read_bytes = 0;