Skip to content

Commit

Permalink
feat(tianmu): Improve the readbility of stonedb. (#11)
Browse files Browse the repository at this point in the history 
[summary]
1 class member variable refactor：exproter directory;
2 class member variable refactor：loader directory;
  • Loading branch information
@konghaiya
konghaiya committed Nov 2, 2022
1 parent 5e86e93 commit 8086fbc
Show file tree
Hide file tree
Showing 10 changed files with 251 additions and 249 deletions.
24 changes: 12 additions & 12 deletions storage/tianmu/exporter/data_exporter.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,23 +27,23 @@ namespace exporter {

void DataExporter::Init(std::shared_ptr<system::LargeBuffer> buffer, std::vector<core::AttributeTypeInfo> source_deas,
fields_t const &fields, std::vector<core::AttributeTypeInfo> &result_deas) {
_fields = fields;
buf = buffer;
fields_ = fields;
data_exporter_buf_ = buffer;

this->source_deas = source_deas;
this->deas = result_deas;
this->no_attrs = int(deas.size());
this->source_attr_infos_ = source_deas;
this->attr_infos_ = result_deas;
this->no_attrs_ = int(attr_infos_.size());

for (size_t i = 0; i < deas.size(); ++i) {
for (size_t i = 0; i < attr_infos_.size(); ++i) {
common::CT f_at = ha_rcengine_->GetCorrespondingType(fields[i]);
if (core::ATI::IsStringType(deas[i].Type()) && !core::ATI::IsStringType(f_at))
this->deas[i] = core::AttributeTypeInfo(f_at, deas[i].NotNull());
if (core::ATI::IsStringType(attr_infos_[i].Type()) && !core::ATI::IsStringType(f_at))
this->attr_infos_[i] = core::AttributeTypeInfo(f_at, attr_infos_[i].NotNull());
}

cur_attr = 0;
row = NULL;
row_ptr = NULL;
nulls_indicator = 0;
cur_attr_ = 0;
row_ = NULL;
row_ptr_ = NULL;
nulls_indicator_ = 0;
}

DataExporter::~DataExporter() {}
Expand Down
28 changes: 14 additions & 14 deletions storage/tianmu/exporter/data_exporter.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,13 +29,13 @@ namespace exporter {

class DataExporter {
public:
DataExporter() : progressout(NULL), row(NULL), row_ptr(NULL), nulls_indicator(NULL) {}
DataExporter() : progress_out_(NULL), row_(NULL), row_ptr_(NULL), nulls_indicator_(NULL) {}
virtual void Init(std::shared_ptr<system::LargeBuffer> buffer, std::vector<core::AttributeTypeInfo> source_deas,
fields_t const &fields, std::vector<core::AttributeTypeInfo> &result_deas);
virtual ~DataExporter();
void FlushBuffer();

void SetProgressOut(system::ChannelOut *po) { progressout = po; }
void SetProgressOut(system::ChannelOut *po) { progress_out_ = po; }
void ShowProgress(int no_eq);

virtual void PutNull() = 0;
Expand All @@ -46,22 +46,22 @@ class DataExporter {
virtual void PutRowEnd() = 0;

protected:
int cur_attr;
std::vector<core::AttributeTypeInfo> deas;
std::vector<core::AttributeTypeInfo> source_deas;
fields_t _fields;
int cur_attr_ = 0;
std::vector<core::AttributeTypeInfo> attr_infos_;
std::vector<core::AttributeTypeInfo> source_attr_infos_;
fields_t fields_;

std::shared_ptr<system::LargeBuffer> buf;
system::ChannelOut *progressout;
std::shared_ptr<system::LargeBuffer> data_exporter_buf_;
system::ChannelOut *progress_out_;

int no_attrs;
int no_attrs_ = 0;

// the fields below should be moved to RCDEforBinIndicator
char *row;
char *row_ptr;
int max_row_size;
int nulls_indicator_len;
char *nulls_indicator;
char *row_;
char *row_ptr_;
int max_row_size_ = 0;
int nulls_indicator_len_ = 0;
char *nulls_indicator_;
};

} // namespace exporter
Expand Down
84 changes: 43 additions & 41 deletions storage/tianmu/exporter/data_exporter_txt.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,31 +23,31 @@ namespace Tianmu {
namespace exporter {

DEforTxt::DEforTxt(const system::IOParameters &iop)
: delim(iop.Delimiter()[0]),
str_q(iop.StringQualifier()),
esc(iop.EscapeCharacter()),
line_terminator(iop.LineTerminator()[0]),
nulls_str(iop.NullsStr()),
destination_cs(iop.CharsetInfoNumber() ? get_charset(iop.CharsetInfoNumber(), 0) : 0) {}
: delimiter_(iop.Delimiter()[0]),
str_qualifier_(iop.StringQualifier()),
escape_character_(iop.EscapeCharacter()),
line_terminator_(iop.LineTerminator()[0]),
nulls_str_(iop.NullsStr()),
destination_charset_(iop.CharsetInfoNumber() ? get_charset(iop.CharsetInfoNumber(), 0) : 0) {}

void DEforTxt::PutText(const types::BString &str) {
WriteStringQualifier();
size_t char_len =
deas[cur_attr].GetCollation().collation->cset->numchars(deas[cur_attr].GetCollation().collation, str.val_,
str.val_ + str.len_); // len in chars
WriteString(str, str.len_); // len in bytes
if ((deas[cur_attr].Type() == common::CT::STRING) && (char_len < deas[cur_attr].CharLen()))
size_t char_len = attr_infos_[cur_attr_].GetCollation().collation->cset->numchars(
attr_infos_[cur_attr_].GetCollation().collation, str.val_,
str.val_ + str.len_); // len in chars
WriteString(str, str.len_); // len in bytes
if ((attr_infos_[cur_attr_].Type() == common::CT::STRING) && (char_len < attr_infos_[cur_attr_].CharLen()))
// it can be necessary to change the WritePad implementation to something like:
// collation->cset->fill(cs, copy->to_ptr+copy->from_length,
// copy->to_length-copy->from_length, '
// ');
// if ' ' (space) can have different codes.
// for export data not fill space
#if 0
if (!destination_cs) { //export as binary
WritePad(deas[cur_attr].Precision() - bytes_written);
if (!destination_cs_) { //export as binary
WritePad(attr_infos_[cur_attr_].Precision() - bytes_written);
} else {
WritePad(deas[cur_attr].CharLen() - char_len);
WritePad(attr_infos_[cur_attr_].CharLen() - char_len);
}
#endif
WriteStringQualifier();
Expand All @@ -56,9 +56,9 @@ void DEforTxt::PutText(const types::BString &str) {

void DEforTxt::PutBin(const types::BString &str) {
int len = str.size();
// if((rcdea[cur_attr].attrt == common::CT::BYTE) && (len <
// rcdea[cur_attr].size))
// len = rcdea[cur_attr].size;
// if((rcdea[cur_attr_].attrt == common::CT::BYTE) && (len <
// rcdea[cur_attr_].size))
// len = rcdea[cur_attr_].size;
if (len > 0) {
char *hex = new char[len * 2];
system::Convert2Hex((const unsigned char *)str.val_, len, hex, len * 2, false);
Expand All @@ -69,58 +69,60 @@ void DEforTxt::PutBin(const types::BString &str) {
}

void DEforTxt::PutNumeric(int64_t num) {
types::RCNum rcn(num, source_deas[cur_attr].Scale(), core::ATI::IsRealType(source_deas[cur_attr].Type()),
source_deas[cur_attr].Type());
types::RCNum rcn(num, source_attr_infos_[cur_attr_].Scale(),
core::ATI::IsRealType(source_attr_infos_[cur_attr_].Type()), source_attr_infos_[cur_attr_].Type());
types::BString rcs = rcn.ToBString();
WriteString(rcs);
WriteValueEnd();
}

void DEforTxt::PutDateTime(int64_t dt) {
types::RCDateTime rcdt(dt, deas[cur_attr].Type());
types::RCDateTime rcdt(dt, attr_infos_[cur_attr_].Type());
types::BString rcs = rcdt.ToBString();
WriteString(rcs);
WriteValueEnd();
}

void DEforTxt::PutRowEnd() {
if (line_terminator == 0)
if (line_terminator_ == 0)
WriteString("\r\n", 2);
else
buf->WriteIfNonzero(line_terminator);
data_exporter_buf_->WriteIfNonzero(line_terminator_);
}

size_t DEforTxt::WriteString(const types::BString &str, int len) {
int res_len = 0;
if (esc) {
escaped.erase();
if (escape_character_) {
escaped_.erase();
for (size_t i = 0; i < str.size(); i++) {
if (str[i] == str_q || (!str_q && str[i] == delim))
escaped.append(1, esc);
escaped.append(1, str[i]);
if (str[i] == str_qualifier_ || (!str_qualifier_ && str[i] == delimiter_))
escaped_.append(1, escape_character_);
escaped_.append(1, str[i]);
}
if (destination_cs) {
int max_res_len =
std::max(destination_cs->mbmaxlen * len + len, deas[cur_attr].GetCollation().collation->mbmaxlen * len + len);
if (destination_charset_) {
int max_res_len = std::max(destination_charset_->mbmaxlen * len + len,
attr_infos_[cur_attr_].GetCollation().collation->mbmaxlen * len + len);
uint errors = 0;
res_len = copy_and_convert(buf->BufAppend(max_res_len), max_res_len, destination_cs, escaped.c_str(),
uint32_t(escaped.length()), deas[cur_attr].GetCollation().collation, &errors);
buf->SeekBack(max_res_len - res_len);
res_len = copy_and_convert(data_exporter_buf_->BufAppend(max_res_len), max_res_len, destination_charset_,
escaped_.c_str(), uint32_t(escaped_.length()),
attr_infos_[cur_attr_].GetCollation().collation, &errors);
data_exporter_buf_->SeekBack(max_res_len - res_len);
} else {
std::strncpy(buf->BufAppend(uint(escaped.length())), escaped.c_str(), escaped.length());
std::strncpy(data_exporter_buf_->BufAppend(uint(escaped_.length())), escaped_.c_str(), escaped_.length());
res_len = len;
}

} else {
if (destination_cs) {
int max_res_len =
std::max(destination_cs->mbmaxlen * len, deas[cur_attr].GetCollation().collation->mbmaxlen * len);
if (destination_charset_) {
int max_res_len = std::max(destination_charset_->mbmaxlen * len,
attr_infos_[cur_attr_].GetCollation().collation->mbmaxlen * len);
uint errors = 0;
res_len = copy_and_convert(buf->BufAppend(max_res_len), max_res_len, destination_cs, str.GetDataBytesPointer(),
len, deas[cur_attr].GetCollation().collation, &errors);
buf->SeekBack(max_res_len - res_len);
res_len =
copy_and_convert(data_exporter_buf_->BufAppend(max_res_len), max_res_len, destination_charset_,
str.GetDataBytesPointer(), len, attr_infos_[cur_attr_].GetCollation().collation, &errors);
data_exporter_buf_->SeekBack(max_res_len - res_len);
} else {
str.CopyTo(buf->BufAppend((uint)len), len);
str.CopyTo(data_exporter_buf_->BufAppend((uint)len), len);
res_len = len;
}
}
Expand Down
22 changes: 11 additions & 11 deletions storage/tianmu/exporter/data_exporter_txt.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,27 +43,27 @@ class DEforTxt : public DataExporter {
void PutRowEnd() override;

protected:
void WriteStringQualifier() { /*buf->WriteIfNonzero(str_q);*/
void WriteStringQualifier() { /*data_exporter_buf_->WriteIfNonzero(str_q);*/
}
void WriteDelimiter() { buf->WriteIfNonzero(delim); }
void WriteNull() { WriteString(nulls_str.c_str(), (int)nulls_str.length()); }
void WriteDelimiter() { data_exporter_buf_->WriteIfNonzero(delimiter_); }
void WriteNull() { WriteString(nulls_str_.c_str(), (int)nulls_str_.length()); }
void WriteString(const types::BString &str) { WriteString(str, str.size()); }
size_t WriteString(const types::BString &str, int len);
void WriteChar(char c, uint repeat = 1) { std::memset(buf->BufAppend(repeat), c, repeat); }
void WriteChar(char c, uint repeat = 1) { std::memset(data_exporter_buf_->BufAppend(repeat), c, repeat); }
void WritePad(uint repeat) { WriteChar(' ', repeat); }
void WriteValueEnd() {
if (cur_attr == no_attrs - 1)
cur_attr = 0;
if (cur_attr_ == no_attrs_ - 1)
cur_attr_ = 0;
else {
WriteDelimiter();
cur_attr++;
cur_attr_++;
}
}

uchar delim, str_q, esc;
uchar line_terminator;
std::string nulls_str, escaped;
CHARSET_INFO *destination_cs;
uchar delimiter_, str_qualifier_, escape_character_;
uchar line_terminator_;
std::string nulls_str_, escaped_;
CHARSET_INFO *destination_charset_;
};

} // namespace exporter
Expand Down
6 changes: 3 additions & 3 deletions storage/tianmu/exporter/export2file.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ namespace Tianmu {
namespace exporter {

select_tianmu_export::select_tianmu_export(Query_result_export *se)
: Query_result_export(se->get_sql_exchange()), se(se), prepared(false) {}
: Query_result_export(se->get_sql_exchange()), select_export_(se), prepared_(false) {}

int select_tianmu_export::prepare(List<Item> &list, SELECT_LEX_UNIT *u) {
bool blob_flag = 0;
Expand Down Expand Up @@ -53,7 +53,7 @@ int select_tianmu_export::prepare(List<Item> &list, SELECT_LEX_UNIT *u) {
exchange->field.opt_enclosed = 1; // A little quicker loop
fixed_row_size = (!field_term_length && !exchange->field.enclosed->length() && !blob_flag);

prepared = true;
prepared_ = true;
return 0;
}

Expand All @@ -63,7 +63,7 @@ void select_tianmu_export::SendOk(THD *thd) { ::my_ok(thd, row_count); }

sql_exchange *select_tianmu_export::SqlExchange() { return exchange; }

bool select_tianmu_export::send_data(List<Item> &items) { return se->send_data(items); }
bool select_tianmu_export::send_data(List<Item> &items) { return select_export_->send_data(items); }

} // namespace exporter
} // namespace Tianmu
6 changes: 3 additions & 3 deletions storage/tianmu/exporter/export2file.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,12 +33,12 @@ class select_tianmu_export : public Query_result_export {
void SetRowCount(ha_rows x);
void SendOk(THD *thd);
sql_exchange *SqlExchange();
bool IsPrepared() const { return prepared; };
bool IsPrepared() const { return prepared_; };
bool send_data(List<Item> &items) override;

private:
Query_result_export *se;
bool prepared;
Query_result_export *select_export_;
bool prepared_;
};

} // namespace exporter
Expand Down
Loading

0 comments on commit 8086fbc

Please sign in to comment.