diff options
Diffstat (limited to '')
| -rw-r--r-- | src/rgw/rgw_s3select.cc | 1001 |
1 files changed, 1001 insertions, 0 deletions
diff --git a/src/rgw/rgw_s3select.cc b/src/rgw/rgw_s3select.cc new file mode 100644 index 000000000..c7eaa6984 --- /dev/null +++ b/src/rgw/rgw_s3select.cc @@ -0,0 +1,1001 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab ft=cpp + +#include "rgw_s3select_private.h" + +#define dout_subsys ceph_subsys_rgw + +namespace rgw::s3select { +RGWOp* create_s3select_op() +{ + return new RGWSelectObj_ObjStore_S3(); +} +}; + +using namespace s3selectEngine; + +std::string& aws_response_handler::get_sql_result() +{ + return sql_result; +} + +uint64_t aws_response_handler::get_processed_size() +{ + return processed_size; +} + +void aws_response_handler::update_processed_size(uint64_t value) +{ + processed_size += value; +} + +uint64_t aws_response_handler::get_total_bytes_returned() +{ + return total_bytes_returned; +} + +void aws_response_handler::update_total_bytes_returned(uint64_t value) +{ + total_bytes_returned = value; +} + +void aws_response_handler::push_header(const char* header_name, const char* header_value) +{ + char x; + short s; + x = char(strlen(header_name)); + m_buff_header.append(&x, sizeof(x)); + m_buff_header.append(header_name); + x = char(7); + m_buff_header.append(&x, sizeof(x)); + s = htons(uint16_t(strlen(header_value))); + m_buff_header.append(reinterpret_cast<char*>(&s), sizeof(s)); + m_buff_header.append(header_value); +} + +#define IDX( x ) static_cast<int>( x ) + +int aws_response_handler::create_header_records() +{ + //headers description(AWS) + //[header-name-byte-length:1][header-name:variable-length][header-value-type:1][header-value:variable-length] + //1 + push_header(header_name_str[IDX(header_name_En::EVENT_TYPE)], header_value_str[IDX(header_value_En::RECORDS)]); + //2 + push_header(header_name_str[IDX(header_name_En::CONTENT_TYPE)], header_value_str[IDX(header_value_En::OCTET_STREAM)]); + //3 + push_header(header_name_str[IDX(header_name_En::MESSAGE_TYPE)], header_value_str[IDX(header_value_En::EVENT)]); + return m_buff_header.size(); +} + +int aws_response_handler::create_header_continuation() +{ + //headers description(AWS) + //1 + push_header(header_name_str[IDX(header_name_En::EVENT_TYPE)], header_value_str[IDX(header_value_En::CONT)]); + //2 + push_header(header_name_str[IDX(header_name_En::MESSAGE_TYPE)], header_value_str[IDX(header_value_En::EVENT)]); + return m_buff_header.size(); +} + +int aws_response_handler::create_header_progress() +{ + //headers description(AWS) + //1 + push_header(header_name_str[IDX(header_name_En::EVENT_TYPE)], header_value_str[IDX(header_value_En::PROGRESS)]); + //2 + push_header(header_name_str[IDX(header_name_En::CONTENT_TYPE)], header_value_str[IDX(header_value_En::XML)]); + //3 + push_header(header_name_str[IDX(header_name_En::MESSAGE_TYPE)], header_value_str[IDX(header_value_En::EVENT)]); + return m_buff_header.size(); +} + +int aws_response_handler::create_header_stats() +{ + //headers description(AWS) + //1 + push_header(header_name_str[IDX(header_name_En::EVENT_TYPE)], header_value_str[IDX(header_value_En::STATS)]); + //2 + push_header(header_name_str[IDX(header_name_En::CONTENT_TYPE)], header_value_str[IDX(header_value_En::XML)]); + //3 + push_header(header_name_str[IDX(header_name_En::MESSAGE_TYPE)], header_value_str[IDX(header_value_En::EVENT)]); + return m_buff_header.size(); +} + +int aws_response_handler::create_header_end() +{ + //headers description(AWS) + //1 + push_header(header_name_str[IDX(header_name_En::EVENT_TYPE)], header_value_str[IDX(header_value_En::END)]); + //2 + push_header(header_name_str[IDX(header_name_En::MESSAGE_TYPE)], header_value_str[IDX(header_value_En::EVENT)]); + return m_buff_header.size(); +} + +int aws_response_handler::create_error_header_records(const char* error_message) +{ + //headers description(AWS) + //[header-name-byte-length:1][header-name:variable-length][header-value-type:1][header-value:variable-length] + //1 + push_header(header_name_str[IDX(header_name_En::ERROR_CODE)], header_value_str[IDX(header_value_En::ENGINE_ERROR)]); + //2 + push_header(header_name_str[IDX(header_name_En::ERROR_MESSAGE)], error_message); + //3 + push_header(header_name_str[IDX(header_name_En::MESSAGE_TYPE)], header_value_str[IDX(header_value_En::ERROR_TYPE)]); + return m_buff_header.size(); +} + +int aws_response_handler::create_message(u_int32_t header_len) +{ + //message description(AWS): + //[total-byte-length:4][header-byte-length:4][crc:4][headers:variable-length][payload:variable-length][crc:4] + //s3select result is produced into sql_result, the sql_result is also the response-message, thus the attach headers and CRC + //are created later to the produced SQL result, and actually wrapping the payload. + auto push_encode_int = [&](u_int32_t s, int pos) { + u_int32_t x = htonl(s); + sql_result.replace(pos, sizeof(x), reinterpret_cast<char*>(&x), sizeof(x)); + }; + u_int32_t total_byte_len = 0; + u_int32_t preload_crc = 0; + u_int32_t message_crc = 0; + total_byte_len = sql_result.size() + 4; //the total is greater in 4 bytes than current size + push_encode_int(total_byte_len, 0); + push_encode_int(header_len, 4); + crc32.reset(); + crc32 = std::for_each(sql_result.data(), sql_result.data() + 8, crc32); //crc for starting 8 bytes + preload_crc = crc32(); + push_encode_int(preload_crc, 8); + crc32.reset(); + crc32 = std::for_each(sql_result.begin(), sql_result.end(), crc32); //crc for payload + checksum + message_crc = crc32(); + u_int32_t x = htonl(message_crc); + sql_result.append(reinterpret_cast<char*>(&x), sizeof(x)); + return sql_result.size(); +} + +void aws_response_handler::init_response() +{ + //12 positions for header-crc + sql_result.resize(header_crc_size, '\0'); +} + +void aws_response_handler::init_success_response() +{ + m_buff_header.clear(); + header_size = create_header_records(); + sql_result.append(m_buff_header.c_str(), header_size); +#ifdef PAYLOAD_TAG + sql_result.append(PAYLOAD_LINE); +#endif +} + +void aws_response_handler::send_continuation_response() +{ + sql_result.resize(header_crc_size, '\0'); + m_buff_header.clear(); + header_size = create_header_continuation(); + sql_result.append(m_buff_header.c_str(), header_size); + int buff_len = create_message(header_size); + s->formatter->write_bin_data(sql_result.data(), buff_len); + rgw_flush_formatter_and_reset(s, s->formatter); +} + +void aws_response_handler::init_progress_response() +{ + sql_result.resize(header_crc_size, '\0'); + m_buff_header.clear(); + header_size = create_header_progress(); + sql_result.append(m_buff_header.c_str(), header_size); +} + +void aws_response_handler::init_stats_response() +{ + sql_result.resize(header_crc_size, '\0'); + m_buff_header.clear(); + header_size = create_header_stats(); + sql_result.append(m_buff_header.c_str(), header_size); +} + +void aws_response_handler::init_end_response() +{ + sql_result.resize(header_crc_size, '\0'); + m_buff_header.clear(); + header_size = create_header_end(); + sql_result.append(m_buff_header.c_str(), header_size); + int buff_len = create_message(header_size); + s->formatter->write_bin_data(sql_result.data(), buff_len); + rgw_flush_formatter_and_reset(s, s->formatter); +} + +void aws_response_handler::init_error_response(const char* error_message) +{ + //currently not in use. the headers in the case of error, are not extracted by AWS-cli. + m_buff_header.clear(); + header_size = create_error_header_records(error_message); + sql_result.append(m_buff_header.c_str(), header_size); +} + +void aws_response_handler::send_success_response() +{ +#ifdef PAYLOAD_TAG + sql_result.append(END_PAYLOAD_LINE); +#endif + int buff_len = create_message(header_size); + s->formatter->write_bin_data(sql_result.data(), buff_len); + rgw_flush_formatter_and_reset(s, s->formatter); +} + +void aws_response_handler::send_error_response(const char* error_code, + const char* error_message, + const char* resource_id) +{ + set_req_state_err(s, 0); + dump_errno(s, 400); + end_header(s, m_rgwop, "application/xml", CHUNKED_TRANSFER_ENCODING); + dump_start(s); + s->formatter->open_object_section("Error"); + s->formatter->dump_string("Code", error_code); + s->formatter->dump_string("Message", error_message); + s->formatter->dump_string("Resource", "#Resource#"); + s->formatter->dump_string("RequestId", resource_id); + s->formatter->close_section(); + rgw_flush_formatter_and_reset(s, s->formatter); +} + +void aws_response_handler::send_progress_response() +{ + std::string progress_payload = fmt::format("<?xml version=\"1.0\" encoding=\"UTF-8\"?><Progress><BytesScanned>{}</BytesScanned><BytesProcessed>{}</BytesProcessed><BytesReturned>{}</BytesReturned></Progress>" + , get_processed_size(), get_processed_size(), get_total_bytes_returned()); + sql_result.append(progress_payload); + int buff_len = create_message(header_size); + s->formatter->write_bin_data(sql_result.data(), buff_len); + rgw_flush_formatter_and_reset(s, s->formatter); +} + +void aws_response_handler::send_stats_response() +{ + std::string stats_payload = fmt::format("<?xml version=\"1.0\" encoding=\"UTF-8\"?><Stats><BytesScanned>{}</BytesScanned><BytesProcessed>{}</BytesProcessed><BytesReturned>{}</BytesReturned></Stats>" + , get_processed_size(), get_processed_size(), get_total_bytes_returned()); + sql_result.append(stats_payload); + int buff_len = create_message(header_size); + s->formatter->write_bin_data(sql_result.data(), buff_len); + rgw_flush_formatter_and_reset(s, s->formatter); +} + +RGWSelectObj_ObjStore_S3::RGWSelectObj_ObjStore_S3(): + m_buff_header(std::make_unique<char[]>(1000)), + m_scan_range_ind(false), + m_start_scan_sz(0), + m_end_scan_sz(0), + m_object_size_for_processing(0), + m_parquet_type(false), + m_json_type(false), + chunk_number(0), + m_requested_range(0), + m_scan_offset(1024), + m_skip_next_chunk(false), + m_is_trino_request(false) +{ + set_get_data(true); + fp_get_obj_size = [&]() { + return get_obj_size(); + }; + fp_range_req = [&](int64_t start, int64_t len, void* buff, optional_yield* y) { + ldout(s->cct, 10) << "S3select: range-request start: " << start << " length: " << len << dendl; + auto status = range_request(start, len, buff, *y); + return status; + }; +#ifdef _ARROW_EXIST + m_rgw_api.set_get_size_api(fp_get_obj_size); + m_rgw_api.set_range_req_api(fp_range_req); +#endif + fp_result_header_format = [this](std::string& result) { + m_aws_response_handler.init_response(); + m_aws_response_handler.init_success_response(); + return 0; + }; + fp_s3select_result_format = [this](std::string& result) { + fp_chunked_transfer_encoding(); + m_aws_response_handler.send_success_response(); + return 0; + }; + + fp_debug_mesg = [&](const char* mesg){ + ldpp_dout(this, 10) << mesg << dendl; + }; + + fp_chunked_transfer_encoding = [&](void){ + if (chunk_number == 0) { + if (op_ret < 0) { + set_req_state_err(s, op_ret); + } + dump_errno(s); + // Explicitly use chunked transfer encoding so that we can stream the result + // to the user without having to wait for the full length of it. + end_header(s, this, "application/xml", CHUNKED_TRANSFER_ENCODING); + } + chunk_number++; + }; +} + +RGWSelectObj_ObjStore_S3::~RGWSelectObj_ObjStore_S3() +{} + +int RGWSelectObj_ObjStore_S3::get_params(optional_yield y) +{ + if(m_s3select_query.empty() == false) { + return 0; + } +#ifndef _ARROW_EXIST + m_parquet_type = false; + ldpp_dout(this, 10) << "arrow library is not installed" << dendl; +#endif + + //retrieve s3-select query from payload + bufferlist data; + int ret; + int max_size = 4096; + std::tie(ret, data) = read_all_input(s, max_size, false); + if (ret != 0) { + ldpp_dout(this, 10) << "s3-select query: failed to retrieve query; ret = " << ret << dendl; + return ret; + } + m_s3select_query = data.to_str(); + if (m_s3select_query.length() > 0) { + ldpp_dout(this, 10) << "s3-select query: " << m_s3select_query << dendl; + } else { + ldpp_dout(this, 10) << "s3-select query: failed to retrieve query;" << dendl; + return -1; + } + const auto& m = s->info.env->get_map(); + auto user_agent = m.find("HTTP_USER_AGENT"); { + if (user_agent != m.end()){ + if (user_agent->second.find("Trino") != std::string::npos){ + m_is_trino_request = true; + ldpp_dout(this, 10) << "s3-select query: request sent by Trino." << dendl; + } + } + } + + int status = handle_aws_cli_parameters(m_sql_query); + if (status<0) { + return status; + } + return RGWGetObj_ObjStore_S3::get_params(y); +} + +int RGWSelectObj_ObjStore_S3::run_s3select_on_csv(const char* query, const char* input, size_t input_length) +{ + int status = 0; + uint32_t length_before_processing, length_post_processing; + csv_object::csv_defintions csv; + const char* s3select_syntax_error = "s3select-Syntax-Error"; + const char* s3select_resource_id = "resourcse-id"; + const char* s3select_processTime_error = "s3select-ProcessingTime-Error"; + + s3select_syntax.parse_query(query); + if (m_row_delimiter.size()) { + csv.row_delimiter = *m_row_delimiter.c_str(); + } + if (m_column_delimiter.size()) { + csv.column_delimiter = *m_column_delimiter.c_str(); + } + if (m_quot.size()) { + csv.quot_char = *m_quot.c_str(); + } + if (m_escape_char.size()) { + csv.escape_char = *m_escape_char.c_str(); + } + if (output_row_delimiter.size()) { + csv.output_row_delimiter = *output_row_delimiter.c_str(); + } + if (output_column_delimiter.size()) { + csv.output_column_delimiter = *output_column_delimiter.c_str(); + } + if (output_quot.size()) { + csv.output_quot_char = *output_quot.c_str(); + } + if (output_escape_char.size()) { + csv.output_escape_char = *output_escape_char.c_str(); + } + if(output_quote_fields.compare("ALWAYS") == 0) { + csv.quote_fields_always = true; + } else if(output_quote_fields.compare("ASNEEDED") == 0) { + csv.quote_fields_asneeded = true; + } + if(m_header_info.compare("IGNORE")==0) { + csv.ignore_header_info=true; + } else if(m_header_info.compare("USE")==0) { + csv.use_header_info=true; + } + //m_s3_csv_object.set_external_debug_system(fp_debug_mesg); + m_s3_csv_object.set_result_formatters(fp_s3select_result_format,fp_result_header_format); + m_s3_csv_object.set_csv_query(&s3select_syntax, csv); + if (s3select_syntax.get_error_description().empty() == false) { + //error-flow (syntax-error) + m_aws_response_handler.send_error_response(s3select_syntax_error, + s3select_syntax.get_error_description().c_str(), + s3select_resource_id); + ldpp_dout(this, 10) << "s3-select query: failed to prase the following query {" << query << "}" << dendl; + ldpp_dout(this, 10) << "s3-select query: syntax-error {" << s3select_syntax.get_error_description() << "}" << dendl; + return -1; + } else { + if (input == nullptr) { + input = ""; + } + fp_result_header_format(m_aws_response_handler.get_sql_result()); + length_before_processing = m_s3_csv_object.get_return_result_size(); + //query is correct(syntax), processing is starting. + status = m_s3_csv_object.run_s3select_on_stream(m_aws_response_handler.get_sql_result(), input, input_length, m_object_size_for_processing); + length_post_processing = m_s3_csv_object.get_return_result_size(); + m_aws_response_handler.update_total_bytes_returned( m_s3_csv_object.get_return_result_size() ); + + if (status < 0) { + //error flow(processing-time) + m_aws_response_handler.send_error_response(s3select_processTime_error, + m_s3_csv_object.get_error_description().c_str(), + s3select_resource_id); + ldpp_dout(this, 10) << "s3-select query: failed to process query; {" << m_s3_csv_object.get_error_description() << "}" << dendl; + return -1; + } + + } + if ((length_post_processing-length_before_processing) != 0) { + ldpp_dout(this, 10) << "s3-select: sql-result-size = " << m_aws_response_handler.get_sql_result().size() << dendl; + } else { + m_aws_response_handler.send_continuation_response(); + } + ldpp_dout(this, 10) << "s3-select: complete chunk processing : chunk length = " << input_length << dendl; + if (enable_progress == true) { + fp_chunked_transfer_encoding(); + m_aws_response_handler.init_progress_response(); + m_aws_response_handler.send_progress_response(); + } + return status; +} + +int RGWSelectObj_ObjStore_S3::run_s3select_on_parquet(const char* query) +{ + int status = 0; +#ifdef _ARROW_EXIST + if (!m_s3_parquet_object.is_set()) { + //parsing the SQL statement. + s3select_syntax.parse_query(m_sql_query.c_str()); + //m_s3_parquet_object.set_external_debug_system(fp_debug_mesg); + try { + //at this stage the Parquet-processing requires for the meta-data that reside on Parquet object + m_s3_parquet_object.set_parquet_object(std::string("s3object"), &s3select_syntax, &m_rgw_api); + } catch(base_s3select_exception& e) { + ldpp_dout(this, 10) << "S3select: failed upon parquet-reader construction: " << e.what() << dendl; + fp_result_header_format(m_aws_response_handler.get_sql_result()); + m_aws_response_handler.get_sql_result().append(e.what()); + fp_s3select_result_format(m_aws_response_handler.get_sql_result()); + return -1; + } + } + if (s3select_syntax.get_error_description().empty() == false) { + //the SQL statement failed the syntax parser + fp_result_header_format(m_aws_response_handler.get_sql_result()); + m_aws_response_handler.get_sql_result().append(s3select_syntax.get_error_description().data()); + fp_s3select_result_format(m_aws_response_handler.get_sql_result()); + ldpp_dout(this, 10) << "s3-select query: failed to prase query; {" << s3select_syntax.get_error_description() << "}" << dendl; + status = -1; + } else { + fp_result_header_format(m_aws_response_handler.get_sql_result()); + //at this stage the Parquet-processing "takes control", it keep calling to s3-range-request according to the SQL statement. + status = m_s3_parquet_object.run_s3select_on_object(m_aws_response_handler.get_sql_result(), fp_s3select_result_format, fp_result_header_format); + if (status < 0) { + m_aws_response_handler.get_sql_result().append(m_s3_parquet_object.get_error_description()); + fp_s3select_result_format(m_aws_response_handler.get_sql_result()); + ldout(s->cct, 10) << "S3select: failure while execution" << m_s3_parquet_object.get_error_description() << dendl; + } + } +#endif + return status; +} + +int RGWSelectObj_ObjStore_S3::run_s3select_on_json(const char* query, const char* input, size_t input_length) +{ + int status = 0; + + const char* s3select_processTime_error = "s3select-ProcessingTime-Error"; + const char* s3select_syntax_error = "s3select-Syntax-Error"; + const char* s3select_resource_id = "resourcse-id"; + const char* s3select_json_error = "json-Format-Error"; + + m_aws_response_handler.init_response(); + + //the JSON data-type should be(currently) only DOCUMENT + if (m_json_datatype.compare("DOCUMENT") != 0) { + const char* s3select_json_error_msg = "s3-select query: wrong json dataType should use DOCUMENT; "; + m_aws_response_handler.send_error_response(s3select_json_error, + s3select_json_error_msg, + s3select_resource_id); + ldpp_dout(this, 10) << s3select_json_error_msg << dendl; + return -EINVAL; + } + + //parsing the SQL statement + s3select_syntax.parse_query(m_sql_query.c_str()); + if (s3select_syntax.get_error_description().empty() == false) { + //SQL statement is wrong(syntax). + m_aws_response_handler.send_error_response(s3select_syntax_error, + s3select_syntax.get_error_description().c_str(), + s3select_resource_id); + ldpp_dout(this, 10) << "s3-select query: failed to prase query; {" << s3select_syntax.get_error_description() << "}" << dendl; + return -EINVAL; + } + + //initializing json processor + m_s3_json_object.set_json_query(&s3select_syntax); + + if (input == nullptr) { + input = ""; + } + m_aws_response_handler.init_success_response(); + uint32_t length_before_processing = m_aws_response_handler.get_sql_result().size(); + //query is correct(syntax), processing is starting. + try { + status = m_s3_json_object.run_s3select_on_stream(m_aws_response_handler.get_sql_result(), input, input_length, m_object_size_for_processing); + } catch(base_s3select_exception& e) { + ldpp_dout(this, 10) << "S3select: failed to process JSON object: " << e.what() << dendl; + m_aws_response_handler.get_sql_result().append(e.what()); + m_aws_response_handler.send_error_response(s3select_processTime_error, + e.what(), + s3select_resource_id); + return -EINVAL; + } + uint32_t length_post_processing = m_aws_response_handler.get_sql_result().size(); + m_aws_response_handler.update_total_bytes_returned(length_post_processing - length_before_processing); + if (status < 0) { + //error flow(processing-time) + m_aws_response_handler.send_error_response(s3select_processTime_error, + m_s3_json_object.get_error_description().c_str(), + s3select_resource_id); + ldpp_dout(this, 10) << "s3-select query: failed to process query; {" << m_s3_json_object.get_error_description() << "}" << dendl; + return -EINVAL; + } + fp_chunked_transfer_encoding(); + + if (length_post_processing-length_before_processing != 0) { + m_aws_response_handler.send_success_response(); + } else { + m_aws_response_handler.send_continuation_response(); + } + if (enable_progress == true) { + m_aws_response_handler.init_progress_response(); + m_aws_response_handler.send_progress_response(); + } + + return status; +} + +int RGWSelectObj_ObjStore_S3::handle_aws_cli_parameters(std::string& sql_query) +{ + std::string input_tag{"InputSerialization"}; + std::string output_tag{"OutputSerialization"}; + if (chunk_number !=0) { + return 0; + } +#define GT ">" +#define LT "<" +#define APOS "'" + + if (m_s3select_query.find(GT) != std::string::npos) { + boost::replace_all(m_s3select_query, GT, ">"); + } + if (m_s3select_query.find(LT) != std::string::npos) { + boost::replace_all(m_s3select_query, LT, "<"); + } + if (m_s3select_query.find(APOS) != std::string::npos) { + boost::replace_all(m_s3select_query, APOS, "'"); + } + //AWS cli s3select parameters + if (m_s3select_query.find(input_tag+"><CSV") != std::string::npos) { + ldpp_dout(this, 10) << "s3select: engine is set to process CSV objects" << dendl; + } + else if (m_s3select_query.find(input_tag+"><JSON") != std::string::npos) { + m_json_type=true; + ldpp_dout(this, 10) << "s3select: engine is set to process JSON objects" << dendl; + } else if (m_s3select_query.find(input_tag+"><Parquet") != std::string::npos) { + m_parquet_type=true; + ldpp_dout(this, 10) << "s3select: engine is set to process Parquet objects" << dendl; + } + + extract_by_tag(m_s3select_query, "Expression", sql_query); + extract_by_tag(m_s3select_query, "Enabled", m_enable_progress); + size_t _qi = m_s3select_query.find("<" + input_tag + ">", 0); + size_t _qe = m_s3select_query.find("</" + input_tag + ">", _qi); + m_s3select_input = m_s3select_query.substr(_qi + input_tag.size() + 2, _qe - (_qi + input_tag.size() + 2)); + extract_by_tag(m_s3select_input, "FieldDelimiter", m_column_delimiter); + extract_by_tag(m_s3select_input, "QuoteCharacter", m_quot); + extract_by_tag(m_s3select_input, "RecordDelimiter", m_row_delimiter); + extract_by_tag(m_s3select_input, "FileHeaderInfo", m_header_info); + extract_by_tag(m_s3select_input, "Type", m_json_datatype); + if (m_row_delimiter.size()==0) { + m_row_delimiter='\n'; + } else if (m_row_delimiter.compare(" ") == 0) { + //presto change + m_row_delimiter='\n'; + } + extract_by_tag(m_s3select_input, "QuoteEscapeCharacter", m_escape_char); + extract_by_tag(m_s3select_input, "CompressionType", m_compression_type); + size_t _qo = m_s3select_query.find("<" + output_tag + ">", 0); + size_t _qs = m_s3select_query.find("</" + output_tag + ">", _qi); + m_s3select_output = m_s3select_query.substr(_qo + output_tag.size() + 2, _qs - (_qo + output_tag.size() + 2)); + extract_by_tag(m_s3select_output, "FieldDelimiter", output_column_delimiter); + extract_by_tag(m_s3select_output, "QuoteCharacter", output_quot); + extract_by_tag(m_s3select_output, "QuoteEscapeCharacter", output_escape_char); + extract_by_tag(m_s3select_output, "QuoteFields", output_quote_fields); + extract_by_tag(m_s3select_output, "RecordDelimiter", output_row_delimiter); + if (output_row_delimiter.size()==0) { + output_row_delimiter='\n'; + } else if (output_row_delimiter.compare(" ") == 0) { + //presto change + output_row_delimiter='\n'; + } + if (m_compression_type.length()>0 && m_compression_type.compare("NONE") != 0) { + ldpp_dout(this, 10) << "RGW supports currently only NONE option for compression type" << dendl; + return -1; + } + extract_by_tag(m_s3select_query, "Start", m_start_scan); + extract_by_tag(m_s3select_query, "End", m_end_scan); + if (m_start_scan.size() || m_end_scan.size()) { + m_scan_range_ind = true; + if (m_start_scan.size()) { + m_start_scan_sz = std::stol(m_start_scan); + } + if (m_end_scan.size()) { + m_end_scan_sz = std::stol(m_end_scan); + } else { + m_end_scan_sz = std::numeric_limits<std::int64_t>::max(); + } + } + if (m_enable_progress.compare("true")==0) { + enable_progress = true; + } else { + enable_progress = false; + } + return 0; +} + +int RGWSelectObj_ObjStore_S3::extract_by_tag(std::string input, std::string tag_name, std::string& result) +{ + result = ""; + size_t _qs = input.find("<" + tag_name + ">", 0); + size_t qs_input = _qs + tag_name.size() + 2; + if (_qs == std::string::npos) { + return -1; + } + size_t _qe = input.find("</" + tag_name + ">", qs_input); + if (_qe == std::string::npos) { + return -1; + } + result = input.substr(qs_input, _qe - qs_input); + return 0; +} + +size_t RGWSelectObj_ObjStore_S3::get_obj_size() +{ + return s->obj_size; +} + +int RGWSelectObj_ObjStore_S3::range_request(int64_t ofs, int64_t len, void* buff, optional_yield y) +{ + //purpose: implementation for arrow::ReadAt, this may take several async calls. + //send_response_date(call_back) accumulate buffer, upon completion control is back to ReadAt. + range_req_str = "bytes=" + std::to_string(ofs) + "-" + std::to_string(ofs+len-1); + range_str = range_req_str.c_str(); + range_parsed = false; + RGWGetObj::parse_range(); + requested_buffer.clear(); + m_request_range = len; + ldout(s->cct, 10) << "S3select: calling execute(async):" << " request-offset :" << ofs << " request-length :" << len << " buffer size : " << requested_buffer.size() << dendl; + RGWGetObj::execute(y); + if (buff) { + memcpy(buff, requested_buffer.data(), len); + } + ldout(s->cct, 10) << "S3select: done waiting, buffer is complete buffer-size:" << requested_buffer.size() << dendl; + return len; +} + +void RGWSelectObj_ObjStore_S3::execute(optional_yield y) +{ + int status = 0; + char parquet_magic[4]; + static constexpr uint8_t parquet_magic1[4] = {'P', 'A', 'R', '1'}; + static constexpr uint8_t parquet_magicE[4] = {'P', 'A', 'R', 'E'}; + get_params(y); +#ifdef _ARROW_EXIST + m_rgw_api.m_y = &y; +#endif + if (m_parquet_type) { + //parquet processing + range_request(0, 4, parquet_magic, y); + if (memcmp(parquet_magic, parquet_magic1, 4) && memcmp(parquet_magic, parquet_magicE, 4)) { + ldout(s->cct, 10) << s->object->get_name() << " does not contain parquet magic" << dendl; + op_ret = -ERR_INVALID_REQUEST; + return; + } + s3select_syntax.parse_query(m_sql_query.c_str()); + status = run_s3select_on_parquet(m_sql_query.c_str()); + if (status) { + ldout(s->cct, 10) << "S3select: failed to process query <" << m_sql_query << "> on object " << s->object->get_name() << dendl; + op_ret = -ERR_INVALID_REQUEST; + } else { + ldout(s->cct, 10) << "S3select: complete query with success " << dendl; + } + } else { + //CSV or JSON processing + if (m_scan_range_ind) { + + m_requested_range = (m_end_scan_sz - m_start_scan_sz); + + if(m_is_trino_request){ + // fetch more than requested(m_scan_offset), that additional bytes are scanned for end of row, + // thus the additional length will be processed, and no broken row for Trino. + // assumption: row is smaller than m_scan_offset. (a different approach is to request for additional range) + range_request(m_start_scan_sz, m_requested_range + m_scan_offset, nullptr, y); + } else { + range_request(m_start_scan_sz, m_requested_range, nullptr, y); + } + + } else { + RGWGetObj::execute(y); + } + }//if (m_parquet_type) +} + +int RGWSelectObj_ObjStore_S3::parquet_processing(bufferlist& bl, off_t ofs, off_t len) +{ + fp_chunked_transfer_encoding(); + size_t append_in_callback = 0; + int part_no = 1; + //concat the requested buffer + for (auto& it : bl.buffers()) { + if (it.length() == 0) { + ldout(s->cct, 10) << "S3select: get zero-buffer while appending request-buffer " << dendl; + } + append_in_callback += it.length(); + ldout(s->cct, 10) << "S3select: part " << part_no++ << " it.length() = " << it.length() << dendl; + requested_buffer.append(&(it)[0]+ofs, len); + } + ldout(s->cct, 10) << "S3select:append_in_callback = " << append_in_callback << dendl; + if (requested_buffer.size() < m_request_range) { + ldout(s->cct, 10) << "S3select: need another round buffe-size: " << requested_buffer.size() << " request range length:" << m_request_range << dendl; + return 0; + } else {//buffer is complete + ldout(s->cct, 10) << "S3select: buffer is complete " << requested_buffer.size() << " request range length:" << m_request_range << dendl; + m_request_range = 0; + } + return 0; +} + +void RGWSelectObj_ObjStore_S3::shape_chunk_per_trino_requests(const char* it_cp, off_t& ofs, off_t& len) +{ +//in case it is a scan range request and sent by Trino client. +//this routine chops the start/end of chunks. +//the purpose is to return "perfect" results, with no broken or missing lines. + + off_t new_offset = 0; + if(m_scan_range_ind){//only upon range-scan + int64_t sc=0; + int64_t start =0; + const char* row_delimiter = m_row_delimiter.c_str(); + + ldpp_dout(this, 10) << "s3select query: per Trino request the first and last chunk should modified." << dendl; + + //chop the head of the first chunk and only upon the slice does not include the head of the object. + if(m_start_scan_sz && (m_aws_response_handler.get_processed_size()==0)){ + char* p = const_cast<char*>(it_cp+ofs); + while(strncmp(row_delimiter,p,1) && (p - (it_cp+ofs)) < len)p++; + if(!strncmp(row_delimiter,p,1)){ + new_offset += (p - (it_cp+ofs))+1; + } + } + + //RR : end of the range-request. the original request sent by Trino client + //RD : row-delimiter + //[ ... ] : chunk boundaries + + //chop the end of the last chunk for this request + //if it's the last chunk, search for first row-delimiter for the following different use-cases + if((m_aws_response_handler.get_processed_size()+len) >= m_requested_range){ + //had pass the requested range, start to search for first delimiter + if(m_aws_response_handler.get_processed_size()>m_requested_range){ + //the previous chunk contain the complete request(all data) and an extra bytes. + //thus, search for the first row-delimiter + //[:previous (RR) ... ][:current (RD) ] + start = 0; + } else if(m_aws_response_handler.get_processed_size()){ + //the *current* chunk contain the complete request in the middle of the chunk. + //thus, search for the first row-delimiter after the complete request position + //[:current (RR) .... (RD) ] + start = m_requested_range - m_aws_response_handler.get_processed_size(); + } else { + //the current chunk is the first chunk and it contains complete request + //[:current:first-chunk (RR) .... (RD) ] + start = m_requested_range; + } + + for(sc=start;sc<len;sc++)//assumption : row-delimiter must exist or its end ebject + { + char* p = const_cast<char*>(it_cp) + ofs + sc; + if(!strncmp(row_delimiter,p,1)){ + ldout(s->cct, 10) << "S3select: found row-delimiter on " << sc << " get_processed_size = " << m_aws_response_handler.get_processed_size() << dendl; + len = sc + 1;//+1 is for delimiter. TODO what about m_object_size_for_processing (to update according to len) + //the end of row exist in current chunk. + //thus, the next chunk should be skipped + m_skip_next_chunk = true; + break; + } + } + } + ofs += new_offset; + } + + ldout(s->cct, 10) << "S3select: shape_chunk_per_trino_requests:update progress len = " << len << dendl; + len -= new_offset; +} + +int RGWSelectObj_ObjStore_S3::csv_processing(bufferlist& bl, off_t ofs, off_t len) +{ + int status = 0; + if(m_skip_next_chunk == true){ + return status; + } + + if (s->obj_size == 0 || m_object_size_for_processing == 0) { + status = run_s3select_on_csv(m_sql_query.c_str(), nullptr, 0); + if (status<0){ + return -EINVAL; + } + } else { + auto bl_len = bl.get_num_buffers(); + int buff_no=0; + for(auto& it : bl.buffers()) { + ldpp_dout(this, 10) << "s3select :processing segment " << buff_no << " out of " << bl_len << " off " << ofs + << " len " << len << " obj-size " << m_object_size_for_processing << dendl; + if (it.length() == 0 || len == 0) { + ldpp_dout(this, 10) << "s3select :it->_len is zero. segment " << buff_no << " out of " << bl_len + << " obj-size " << m_object_size_for_processing << dendl; + continue; + } + + if((ofs + len) > it.length()){ + ldpp_dout(this, 10) << "offset and length may cause invalid read: ofs = " << ofs << " len = " << len << " it.length() = " << it.length() << dendl; + ofs = 0; + len = it.length(); + } + + if(m_is_trino_request){ + shape_chunk_per_trino_requests(&(it)[0], ofs, len); + } + + ldpp_dout(this, 10) << "s3select: chunk: ofs = " << ofs << " len = " << len << " it.length() = " << it.length() << " m_object_size_for_processing = " << m_object_size_for_processing << dendl; + + m_aws_response_handler.update_processed_size(it.length());//NOTE : to run analysis to validate len is aligned with m_processed_bytes + status = run_s3select_on_csv(m_sql_query.c_str(), &(it)[0] + ofs, len); + if (status<0) { + return -EINVAL; + } + if (m_s3_csv_object.is_sql_limit_reached()) { + break; + } + buff_no++; + }//for + }//else + + ldpp_dout(this, 10) << "s3select : m_aws_response_handler.get_processed_size() " << m_aws_response_handler.get_processed_size() + << " m_object_size_for_processing " << uint64_t(m_object_size_for_processing) << dendl; + + if (m_aws_response_handler.get_processed_size() >= uint64_t(m_object_size_for_processing) || m_s3_csv_object.is_sql_limit_reached()) { + if (status >=0) { + m_aws_response_handler.init_stats_response(); + m_aws_response_handler.send_stats_response(); + m_aws_response_handler.init_end_response(); + ldpp_dout(this, 10) << "s3select : reached the end of query request : aws_response_handler.get_processed_size() " << m_aws_response_handler.get_processed_size() + << "m_object_size_for_processing : " << m_object_size_for_processing << dendl; + } + if (m_s3_csv_object.is_sql_limit_reached()) { + //stop fetching chunks + ldpp_dout(this, 10) << "s3select : reached the limit :" << m_aws_response_handler.get_processed_size() << dendl; + status = -ENOENT; + } + } + + return status; +} + +int RGWSelectObj_ObjStore_S3::json_processing(bufferlist& bl, off_t ofs, off_t len) +{ + int status = 0; + + if (s->obj_size == 0 || m_object_size_for_processing == 0) { + //in case of empty object the s3select-function returns a correct "empty" result(for aggregation and non-aggregation queries). + status = run_s3select_on_json(m_sql_query.c_str(), nullptr, 0); + if (status<0) + return -EINVAL; + } else { + //loop on buffer-list(chunks) + auto bl_len = bl.get_num_buffers(); + int i=0; + for(auto& it : bl.buffers()) { + ldpp_dout(this, 10) << "processing segment " << i << " out of " << bl_len << " off " << ofs + << " len " << len << " obj-size " << m_object_size_for_processing << dendl; + //skipping the empty chunks + if (len == 0) { + ldpp_dout(this, 10) << "s3select:it->_len is zero. segment " << i << " out of " << bl_len + << " obj-size " << m_object_size_for_processing << dendl; + continue; + } + + if((ofs + len) > it.length()){ + ldpp_dout(this, 10) << "s3select: offset and length may cause invalid read: ofs = " << ofs << " len = " << len << " it.length() = " << it.length() << dendl; + ofs = 0; + len = it.length(); + } + + m_aws_response_handler.update_processed_size(len); + status = run_s3select_on_json(m_sql_query.c_str(), &(it)[0] + ofs, len); + if (status<0) { + status = -EINVAL; + break; + } + if (m_s3_json_object.is_sql_limit_reached()) { + break; + } + i++; + }//for + }//else + + if (status>=0 && (m_aws_response_handler.get_processed_size() == uint64_t(m_object_size_for_processing) || m_s3_json_object.is_sql_limit_reached())) { + //flush the internal JSON buffer(upon last chunk) + status = run_s3select_on_json(m_sql_query.c_str(), nullptr, 0); + if (status<0) { + return -EINVAL; + } + if (status >=0) { + m_aws_response_handler.init_stats_response(); + m_aws_response_handler.send_stats_response(); + m_aws_response_handler.init_end_response(); + } + if (m_s3_json_object.is_sql_limit_reached()){ + //stop fetching chunks + status = -ENOENT; + ldpp_dout(this, 10) << "s3select : reached the limit :" << m_aws_response_handler.get_processed_size() << dendl; + } + } + return status; +} + +int RGWSelectObj_ObjStore_S3::send_response_data(bufferlist& bl, off_t ofs, off_t len) +{ + if (m_scan_range_ind == false){ + m_object_size_for_processing = s->obj_size; + } + if (m_scan_range_ind == true){ + if (m_end_scan_sz == -1){ + m_end_scan_sz = s->obj_size; + } + if (static_cast<uint64_t>((m_end_scan_sz - m_start_scan_sz))>s->obj_size){ //in the case user provides range bigger than object-size + m_object_size_for_processing = s->obj_size; + } else { + m_object_size_for_processing = m_end_scan_sz - m_start_scan_sz; + } + } + if (!m_aws_response_handler.is_set()) { + m_aws_response_handler.set(s, this); + } + if (len == 0 && s->obj_size != 0) { + return 0; + } + if (m_parquet_type) { + return parquet_processing(bl,ofs,len); + } + if (m_json_type) { + return json_processing(bl,ofs,len); + } + return csv_processing(bl,ofs,len); +} + |