github-actions[bot] commented on code in PR #68853: URL: https://github.com/apache/doris/pull/68853#discussion_r4280492165
########## be/src/core/data_type_serde/data_type_file_serde.cpp: ########## @@ -0,0 +1,457 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "core/data_type_serde/data_type_file_serde.h" + +#include <gen_cpp/types.pb.h> +#include <rapidjson/document.h> +#include <rapidjson/memorystream.h> + +#include <algorithm> +#include <array> +#include <limits> +#include <string_view> +#include <utility> + +#include "core/column/column_const.h" +#include "core/column/column_file.h" +#include "core/column/column_nullable.h" +#include "core/custom_allocator.h" +#include "core/data_type/data_type_file.h" +#include "core/string_buffer.hpp" +#include "core/value/file_value.h" +#include "util/jsonb_writer.h" +#include "util/mysql_row_buffer.h" +#include "util/url_coding.h" + +namespace doris { +namespace { + +constexpr size_t FILE_FIELDS = DataTypeFile::FIELD_COUNT; +constexpr size_t INLINE_FIELD_INDEX = DataTypeFile::INLINE_FIELD_INDEX; + +int base64_digit(char ch) { + if (ch >= 'A' && ch <= 'Z') { + return ch - 'A'; + } + if (ch >= 'a' && ch <= 'z') { + return ch - 'a' + 26; + } + if (ch >= '0' && ch <= '9') { + return ch - '0' + 52; + } + if (ch == '+') { + return 62; + } + if (ch == '/') { + return 63; + } + return -1; +} + +Status check_inline_base64(StringRef encoded, size_t padding) { + const auto digits = encoded.size - padding; + for (size_t i = 0; i < digits; ++i) { + if (base64_digit(encoded.data[i]) < 0) { + return Status::InvalidArgument("FILE inline requires standard padded Base64"); + } + } + // RFC 4648's unused low bits must be zero, so each byte sequence has one encoding. + if (padding != 0 && (base64_digit(encoded.data[digits - 1]) & ((1 << (2 * padding)) - 1))) { + return Status::InvalidArgument("FILE inline Base64 has nonzero pad bits"); + } + return Status::OK(); +} + +// Multiples of three keep padding confined to the final chunk. Scratch space is bounded +// independently of the inline value's size. +void write_base64(StringRef bytes, BufferWritable& bw) { + constexpr size_t INPUT_CHUNK = 3 * 1024; + std::array<unsigned char, INPUT_CHUNK / 3 * 4> output; + ColumnString::check_chars_length((bytes.size / 3 + (bytes.size % 3 != 0)) * 4 + 2, 1); + bw.write('"'); + for (size_t offset = 0; offset < bytes.size; offset += INPUT_CHUNK) { + const auto length = std::min(INPUT_CHUNK, bytes.size - offset); + const auto written = base64_encode( + reinterpret_cast<const unsigned char*>(bytes.data + offset), length, output.data()); + bw.write(reinterpret_cast<const char*>(output.data()), written); + } + bw.write('"'); +} + +Status append_file(IColumn& column, File value) { + RETURN_IF_ERROR(validate_file(value)); + column.insert(Field::create_field<TYPE_FILE>(std::move(value))); + return Status::OK(); +} + +Status check_pb_null_map(const PValues& value, int rows) { + if ((value.has_null() && value.null_map_size() != rows) || + (!value.has_null() && value.null_map_size() != 0)) { + return Status::InvalidArgument("FILE protobuf null map does not match row count {}", rows); + } + return Status::OK(); +} + +} // namespace + +DataTypeFileSerDe::DataTypeFileSerDe(int nesting_level) : DataTypeSerDe(nesting_level) { + const DataTypeFile type; + _names = type.get_element_names(); + for (const auto& element : type.get_elements()) { + _element_serdes.push_back(element->get_serde(nesting_level + 1)); + } +} + +void DataTypeFileSerDe::set_return_object_as_string(bool value) { + DataTypeSerDe::set_return_object_as_string(value); + for (auto& serde : _element_serdes) { + serde->set_return_object_as_string(value); + } +} + +void DataTypeFileSerDe::_write_json(const IColumn& column, int64_t row, BufferWritable& bw) const { + const auto [data, index] = check_column_const_set_readability(column, row); + const auto& file = assert_cast<const ColumnFile&>(*data); + bw.write('{'); + for (size_t i = 0; i < FILE_FIELDS; ++i) { + if (i != 0) { + bw.write(','); + } + bw.write_json_string(_names[i].data(), _names[i].size()); + bw.write(':'); + const auto& child = assert_cast<const ColumnNullable&>(file.get_column(i)); + if (child.is_null_at(index)) { + bw.write("null", 4); + } else if (i == 1 || i == 2) { + bw.write_number(child.get_nested_column().get_int(index)); + } else { + const auto bytes = child.get_nested_column().get_data_at(index); + if (i == INLINE_FIELD_INDEX) { + write_base64(bytes, bw); + } else { + bw.write_json_string(bytes.data, bytes.size); + } + } + } + bw.write('}'); +} + +void DataTypeFileSerDe::to_string(const IColumn& column, size_t row_num, BufferWritable& bw, + const FormatOptions& options) const { + _write_json(column, row_num, bw); +} + +bool DataTypeFileSerDe::write_column_to_mysql_text(const IColumn& column, BufferWritable& bw, + int64_t row_idx, + const FormatOptions& options) const { + to_string(column, row_idx, bw, options); + return true; +} + +Status DataTypeFileSerDe::write_column_to_mysql_binary(const IColumn& column, + MysqlRowBinaryBuffer& row_buffer, + int64_t row_idx, bool col_const, + const FormatOptions& options) const { + auto output = ColumnString::create(); + BufferWritable bw(*output); + to_string(column, index_check_const(row_idx, col_const), bw, options); + bw.commit(); + const auto json = output->get_data_at(0); + if (row_buffer.push_string(json.data, json.size) != 0) { + return Status::InternalError("Packing FILE into MySQL binary buffer failed"); + } + return Status::OK(); +} + +Status DataTypeFileSerDe::serialize_one_cell_to_json(const IColumn& column, int64_t row_num, + BufferWritable& bw, + FormatOptions& options) const { + _write_json(column, row_num, bw); + return Status::OK(); +} + +Status DataTypeFileSerDe::serialize_column_to_json(const IColumn& column, int64_t start_idx, + int64_t end_idx, BufferWritable& bw, + FormatOptions& options) const { + SERIALIZE_COLUMN_TO_JSON(); +} + +Status DataTypeFileSerDe::from_string(StringRef& str, IColumn& column, + const FormatOptions& options) const { + return _read_json(str, column); +} + +Status DataTypeFileSerDe::_checked_inline_size(size_t encoded_size, size_t padding, + uint32_t* decoded_size) { + if (encoded_size % 4 != 0 || padding > 2 || (encoded_size == 0 && padding != 0)) { + return Status::InvalidArgument("FILE inline requires standard padded Base64"); + } + constexpr uint64_t MAX_INLINE_SIZE = std::numeric_limits<uint32_t>::max(); + const auto groups = encoded_size / 4; + // Check before multiplication/allocation; the limit is VARBINARY's uint32_t length. + if (groups > (MAX_INLINE_SIZE + padding) / 3) { + return Status::InvalidArgument("FILE inline exceeds VARBINARY value length"); + } + *decoded_size = static_cast<uint32_t>(groups * 3 - padding); + return Status::OK(); +} + +Status DataTypeFileSerDe::_decode_inline(StringRef encoded, DorisVector<char>& scratch, + Field& value) { + size_t padding = 0; + if (encoded.size != 0 && encoded.data[encoded.size - 1] == '=') { + padding = encoded.size > 1 && encoded.data[encoded.size - 2] == '=' ? 2 : 1; + } + uint32_t decoded_size = 0; + RETURN_IF_ERROR(_checked_inline_size(encoded.size, padding, &decoded_size)); + RETURN_IF_ERROR(check_inline_base64(encoded, padding)); + // libbase64 needs room for 3/4 of the encoded input, including padding. + // The checked group count bounds this scratch allocation before it is made. + scratch.resize(static_cast<size_t>(decoded_size) + padding); + if (decoded_size != 0 && + base64_decode(encoded.data, encoded.size, scratch.data()) != decoded_size) { + return Status::InvalidArgument("Invalid FILE inline Base64"); + } + value = Field::create_field<TYPE_VARBINARY>(StringView(scratch.data(), decoded_size)); + return Status::OK(); +} + +Status DataTypeFileSerDe::_read_json(StringRef str, IColumn& column) const { + if (str.size == 0) { + return Status::InvalidArgument("FILE requires a JSON object"); + } + rapidjson::Document document; + rapidjson::MemoryStream stream(str.data, str.size); + document.ParseStream<rapidjson::kParseValidateEncodingFlag>(stream); Review Comment: [P2] Preserve opaque FILE bytes in aggregate-state serialization A valid six-field FILE can use `uri='urn:x'` and `checksum=CONCAT('ETAG:', UNHEX('FF'))`: STRUCT-to-FILE retains these bytes and `validate_file` explicitly permits opaque ETag digests. In a partial/global `COLLECT_LIST(f)` plan, `AggregateFunctionCollectListData` serializes FILE rows through `serialize_one_cell_to_json()` and later calls this decoder. `_write_json()` copies the FF byte into JSON unchanged, so `kParseValidateEncodingFlag` rejects the aggregate's own state and the merge fails with `Failed to deserialize data for File`. The same value can be collected locally, making success depend on exchange/spill execution. Use a lossless internal representation for FILE-containing aggregate state and cover opaque non-UTF-8 checksum/URI bytes across partial-state serialization. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
