github-actions[bot] commented on code in PR #68853: URL: https://github.com/apache/doris/pull/68853#discussion_r4301813907
########## 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); Review Comment: [P2] Preserve escaped metadata in FILE-containing aggregate states With a partial/global `COLLECT_LIST(ARRAY(f, NULL))`, a valid FILE whose `uri` contains one literal `"` (e.g. `urn:a"b`) is serialized here as escaped JSON. `AggregateFunctionCollectListData::read()` restores the array through `DataTypeArraySerDe`, whose splitter toggles quoting at every `"` without honoring backslash escapes. It consequently gives the FILE decoder one slice containing `{...}, null`; the nullable wrapper replaces that failed parse with one NULL and returns OK. Thus `[f, NULL]` becomes `[NULL]` after state restoration, although local aggregation preserves it. The added nested-state tests use opaque FF/NUL bytes but no escaped quote. Make array state decoding escape-aware or use lossless binary state, and cover an embedded quote in `ARRAY(f, NULL)`. This is separate from the earlier scalar UTF-8 decoder failure: ASCII input silently loses data at the containing-array boundary. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java: ########## @@ -738,6 +739,12 @@ public Expr visitSearchExpression(SearchExpression searchExpression, return searchPredicate; } + @Override + public Expr visitToFile(ToFile toFile, PlanTranslatorContext context) { + context.getConnectContext().getOrRegisterS3Resource(toFile.getResourceName()); Review Comment: [P2] Carry the broker-load job context through TO_FILE resolution A broker load into ordinary scalar columns with `SET(n = ELEMENT_AT(TO_FILE('r', uri), 'size'))` reaches this registration from `NereidsLoadPlanInfoCollector`. On a fresh pending-task worker, `BrokerLoadJob.setComputeGroup()` creates a `ConnectContext` without restoring the job's `userInfo`; this lookup then passes a null identity to `AccessTranslation.subjectOf()` and planning throws even when the submitting user has USAGE. The execution handoff also drops the registered resource map: `LoadLoadingTask` runs on another worker, and its metadata-only legacy `Coordinator` has `context == null` (modern dispatch uses the execution worker's context), so restoring only identity still sends no `s3_resources`. Restore the job identity and retain/re-resolve required resources with the loading task for both dispatchers; cover asynchronous broker loading. This first-load failure is separate from the existing prepared/routine-load credential-rotation threads. -- 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]
