Skip to content

[improve](hive) Refactor csv reader #50379

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Jun 6, 2025
Merged

[improve](hive) Refactor csv reader #50379

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
f695e12
use double quote to escape quote
suxiaogang223 Apr 16, 2025
f5bc8ec
intro TextReader
suxiaogang223 Apr 23, 2025
ee5a3b8
add cases
suxiaogang223 May 5, 2025
663a5fc
fix test_open_csv_serde
suxiaogang223 May 7, 2025
fdf333a
fix
suxiaogang223 Jun 4, 2025
0587de7
fix be ut
suxiaogang223 Jun 5, 2025
fc3e4ad
fix p0 coredump
suxiaogang223 Jun 5, 2025
2c9c389
fix
suxiaogang223 Jun 5, 2025
5d3d010
fix p0
suxiaogang223 Jun 5, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions be/src/http/action/stream_load.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,10 +259,6 @@ Status StreamLoadAction::_on_header(HttpRequest* http_req, std::shared_ptr<Strea
//treat as CSV
format_str = BeConsts::CSV;
}
if (iequal(format_str, "hive_text")) {
ctx->header_type = format_str;
format_str = BeConsts::CSV;
}
LoadUtil::parse_format(format_str, http_req->header(HTTP_COMPRESS_TYPE), &ctx->format,
&ctx->compress_type);
if (ctx->format == TFileFormatType::FORMAT_UNKNOWN) {
Expand Down
17 changes: 10 additions & 7 deletions be/src/service/internal_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@
#include "vec/exec/format/json/new_json_reader.h"
#include "vec/exec/format/orc/vorc_reader.h"
#include "vec/exec/format/parquet/vparquet_reader.h"
#include "vec/exec/format/text/text_reader.h"
#include "vec/functions/dictionary_factory.h"
#include "vec/jsonb/serialize.h"
#include "vec/runtime/vdata_stream_mgr.h"
Expand Down Expand Up @@ -847,6 +848,8 @@ void PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
io::IOContext io_ctx;
io::FileCacheStatistics file_cache_statis;
io_ctx.file_cache_stats = &file_cache_statis;
// file_slots is no use, but the lifetime should be longer than reader
std::vector<SlotDescriptor*> file_slots;
switch (params.format_type) {
case TFileFormatType::FORMAT_CSV_PLAIN:
case TFileFormatType::FORMAT_CSV_GZ:
Expand All @@ -856,10 +859,13 @@ void PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:
case TFileFormatType::FORMAT_CSV_LZOP:
case TFileFormatType::FORMAT_CSV_DEFLATE: {
// file_slots is no use
std::vector<SlotDescriptor*> file_slots;
reader = vectorized::CsvReader::create_unique(profile.get(), params, range, file_slots,
&io_ctx);
reader = vectorized::CsvReader::create_unique(nullptr, profile.get(), nullptr, params,
range, file_slots, &io_ctx);
break;
}
case TFileFormatType::FORMAT_TEXT: {
reader = vectorized::TextReader::create_unique(nullptr, profile.get(), nullptr, params,
range, file_slots, &io_ctx);
break;
}
case TFileFormatType::FORMAT_PARQUET: {
Expand All @@ -871,14 +877,11 @@ void PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
break;
}
case TFileFormatType::FORMAT_JSON: {
std::vector<SlotDescriptor*> file_slots;
reader = vectorized::NewJsonReader::create_unique(profile.get(), params, range,
file_slots, &io_ctx);
break;
}
case TFileFormatType::FORMAT_AVRO: {
// file_slots is no use
std::vector<SlotDescriptor*> file_slots;
reader = vectorized::AvroJNIReader::create_unique(profile.get(), params, range,
file_slots);
st = ((vectorized::AvroJNIReader*)(reader.get()))->init_fetch_table_schema_reader();
Expand Down
64 changes: 27 additions & 37 deletions be/src/util/load_util.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,64 +29,54 @@ void LoadUtil::parse_format(const std::string& format_str, const std::string& co
parse_format("CSV", compress_type_str, format_type, compress_type);
return;
}
*compress_type = TFileCompressType::PLAIN;

// Assign compress_type
if (iequal(compress_type_str, "GZ")) {
*compress_type = TFileCompressType::GZ;
} else if (iequal(compress_type_str, "LZO")) {
*compress_type = TFileCompressType::LZO;
} else if (iequal(compress_type_str, "BZ2")) {
*compress_type = TFileCompressType::BZ2;
} else if (iequal(compress_type_str, "LZ4")) {
*compress_type = TFileCompressType::LZ4FRAME;
} else if (iequal(compress_type_str, "LZ4_BLOCK")) {
*compress_type = TFileCompressType::LZ4BLOCK;
} else if (iequal(compress_type_str, "LZOP")) {
*compress_type = TFileCompressType::LZO;
} else if (iequal(compress_type_str, "SNAPPY_BLOCK")) {
*compress_type = TFileCompressType::SNAPPYBLOCK;
} else if (iequal(compress_type_str, "DEFLATE")) {
*compress_type = TFileCompressType::DEFLATE;
} else {
*compress_type = TFileCompressType::PLAIN;
}

// Assign format_type
*format_type = TFileFormatType::FORMAT_UNKNOWN;
if (iequal(format_str, "CSV")) {
if (compress_type_str.empty()) {
*format_type = TFileFormatType::FORMAT_CSV_PLAIN;
} else if (iequal(compress_type_str, "GZ")) {
*format_type = TFileFormatType::FORMAT_CSV_GZ;
*compress_type = TFileCompressType::GZ;
} else if (iequal(compress_type_str, "LZO")) {
*format_type = TFileFormatType::FORMAT_CSV_LZO;
*compress_type = TFileCompressType::LZO;
} else if (iequal(compress_type_str, "BZ2")) {
*format_type = TFileFormatType::FORMAT_CSV_BZ2;
*compress_type = TFileCompressType::BZ2;
} else if (iequal(compress_type_str, "LZ4")) {
*format_type = TFileFormatType::FORMAT_CSV_LZ4FRAME;
*compress_type = TFileCompressType::LZ4FRAME;
} else if (iequal(compress_type_str, "LZ4_BLOCK")) {
*format_type = TFileFormatType::FORMAT_CSV_LZ4BLOCK;
*compress_type = TFileCompressType::LZ4BLOCK;
} else if (iequal(compress_type_str, "LZOP")) {
*format_type = TFileFormatType::FORMAT_CSV_LZOP;
*compress_type = TFileCompressType::LZO;
} else if (iequal(compress_type_str, "SNAPPY_BLOCK")) {
*format_type = TFileFormatType::FORMAT_CSV_SNAPPYBLOCK;
*compress_type = TFileCompressType::SNAPPYBLOCK;
} else if (iequal(compress_type_str, "DEFLATE")) {
*format_type = TFileFormatType::FORMAT_CSV_DEFLATE;
*compress_type = TFileCompressType::DEFLATE;
}
} else if (iequal(format_str, "HIVE_TEXT")) {
*format_type = TFileFormatType::FORMAT_TEXT;
} else if (iequal(format_str, "JSON")) {
if (compress_type_str.empty()) {
*format_type = TFileFormatType::FORMAT_JSON;
} else if (iequal(compress_type_str, "GZ")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::GZ;
} else if (iequal(compress_type_str, "LZO")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::LZO;
} else if (iequal(compress_type_str, "BZ2")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::BZ2;
} else if (iequal(compress_type_str, "LZ4")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::LZ4FRAME;
} else if (iequal(compress_type_str, "LZ4_BLOCK")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::LZ4BLOCK;
} else if (iequal(compress_type_str, "LZOP")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::LZO;
} else if (iequal(compress_type_str, "SNAPPY_BLOCK")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::SNAPPYBLOCK;
} else if (iequal(compress_type_str, "DEFLATE")) {
*format_type = TFileFormatType::FORMAT_JSON;
*compress_type = TFileCompressType::DEFLATE;
}
*format_type = TFileFormatType::FORMAT_JSON;
} else if (iequal(format_str, "PARQUET")) {
*format_type = TFileFormatType::FORMAT_PARQUET;
} else if (iequal(format_str, "ORC")) {
Expand All @@ -96,7 +86,6 @@ void LoadUtil::parse_format(const std::string& format_str, const std::string& co
} else if (iequal(format_str, "ARROW")) {
*format_type = TFileFormatType::FORMAT_ARROW;
}
return;
}

bool LoadUtil::is_format_support_streaming(TFileFormatType::type format) {
Expand All @@ -110,6 +99,7 @@ bool LoadUtil::is_format_support_streaming(TFileFormatType::type format) {
case TFileFormatType::FORMAT_CSV_LZO:
case TFileFormatType::FORMAT_CSV_LZOP:
case TFileFormatType::FORMAT_JSON:
case TFileFormatType::FORMAT_TEXT:
case TFileFormatType::FORMAT_WAL:
case TFileFormatType::FORMAT_ARROW:
return true;
Expand Down
8 changes: 8 additions & 0 deletions be/src/util/thrift_rpc_helper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,30 +68,38 @@ Status ThriftRpcHelper::rpc(const std::string& ip, const int32_t port,
DBUG_EXECUTE_IF("thriftRpcHelper.rpc.error", { timeout_ms = 30000; });
ClientConnection<T> client(_s_exec_env->get_client_cache<T>(), address, timeout_ms, &status);
if (!status.ok()) {
#ifndef ADDRESS_SANITIZER
LOG(WARNING) << "Connect frontend failed, address=" << address << ", status=" << status;
#endif
return status;
}
try {
try {
callback(client);
} catch (apache::thrift::transport::TTransportException& e) {
std::cerr << "thrift error, reason=" << e.what();
#ifndef ADDRESS_SANITIZER
LOG(WARNING) << "retrying call frontend service after "
<< config::thrift_client_retry_interval_ms << " ms, address=" << address
<< ", reason=" << e.what();
#endif
std::this_thread::sleep_for(
std::chrono::milliseconds(config::thrift_client_retry_interval_ms));
status = client.reopen(timeout_ms);
if (!status.ok()) {
#ifndef ADDRESS_SANITIZER
LOG(WARNING) << "client reopen failed. address=" << address
<< ", status=" << status;
#endif
return status;
}
callback(client);
}
} catch (apache::thrift::TException& e) {
#ifndef ADDRESS_SANITIZER
LOG(WARNING) << "call frontend service failed, address=" << address
<< ", reason=" << e.what();
#endif
std::this_thread::sleep_for(
std::chrono::milliseconds(config::thrift_client_retry_interval_ms * 2));
// just reopen to disable this connection
Expand Down
18 changes: 9 additions & 9 deletions be/src/vec/data_types/serde/data_type_string_serde.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,10 @@ class IColumn;
class Arena;
#include "common/compile_check_begin.h"

inline void escape_string(const char* src, size_t& len, char escape_char) {
inline void escape_string(const char* src, size_t* len, char escape_char) {
const char* start = src;
char* dest_ptr = const_cast<char*>(src);
const char* end = src + len;
const char* end = src + *len;
bool escape_next_char = false;

while (src < end) {
Expand All @@ -61,14 +61,14 @@ inline void escape_string(const char* src, size_t& len, char escape_char) {
}
}

len = dest_ptr - start;
*len = dest_ptr - start;
}

// specially escape quote with double quote
inline void escape_string_for_csv(const char* src, size_t& len, char escape_char, char quote_char) {
inline void escape_string_for_csv(const char* src, size_t* len, char escape_char, char quote_char) {
const char* start = src;
char* dest_ptr = const_cast<char*>(src);
const char* end = src + len;
const char* end = src + *len;
bool escape_next_char = false;

while (src < end) {
Expand All @@ -86,7 +86,7 @@ inline void escape_string_for_csv(const char* src, size_t& len, char escape_char
}
}

len = dest_ptr - start;
*len = dest_ptr - start;
}

template <typename ColumnType>
Expand Down Expand Up @@ -208,7 +208,7 @@ class DataTypeStringSerDeBase : public DataTypeSerDe {
slice.trim_quote();
}
if (options.escape_char != 0) {
escape_string(slice.data, slice.size, options.escape_char);
escape_string(slice.data, &slice.size, options.escape_char);
}
assert_cast<ColumnType&>(column).insert_data(slice.data, slice.size);
return Status::OK();
Expand All @@ -217,7 +217,7 @@ class DataTypeStringSerDeBase : public DataTypeSerDe {
Status deserialize_one_cell_from_csv(IColumn& column, Slice& slice,
const FormatOptions& options) const override {
if (options.escape_char != 0) {
escape_string_for_csv(slice.data, slice.size, options.escape_char, options.quote_char);
escape_string_for_csv(slice.data, &slice.size, options.escape_char, options.quote_char);
}
assert_cast<ColumnType&>(column).insert_data(slice.data, slice.size);
return Status::OK();
Expand All @@ -227,7 +227,7 @@ class DataTypeStringSerDeBase : public DataTypeSerDe {
IColumn& column, Slice& slice, const FormatOptions& options,
int hive_text_complex_type_delimiter_level = 1) const override {
if (options.escape_char != 0) {
escape_string(slice.data, slice.size, options.escape_char);
escape_string(slice.data, &slice.size, options.escape_char);
}
assert_cast<ColumnType&>(column).insert_data(slice.data, slice.size);
return Status::OK();
Expand Down
Loading
Loading