github-actions[bot] commented on code in PR #68853: URL: https://github.com/apache/doris/pull/68853#discussion_r4243342102
########## be/src/exprs/function/function_to_file.cpp: ########## @@ -0,0 +1,203 @@ +// 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 "exprs/function/function_to_file.h" + +#include <mutex> +#include <string_view> + +#include "core/column/column_const.h" +#include "core/column/column_file.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type_file.h" +#include "core/value/file_value.h" +#include "exprs/function/simple_function_factory.h" +#include "io/file_factory.h" +#include "runtime/query_context.h" +#include "runtime/runtime_state.h" +#include "util/s3_uri.h" + +namespace doris { +namespace { + +struct ToFileState { + explicit ToFileState(std::shared_ptr<const TS3Resource> resource_) + : resource(std::move(resource_)) {} + + const std::shared_ptr<const TS3Resource> resource; + std::shared_ptr<io::FileSystem> filesystem; + std::string bucket; + std::mutex mutex; + io::FileStatContext stat_context; +}; + +// Only the path participates in extension inference. Keep the stored/stat URI unchanged. +std::string_view to_file_uri_path(std::string_view uri) { + uri = uri.substr(0, uri.find_first_of("?#")); Review Comment: [P2] Infer the fallback MIME type from the full raw S3 key. `parse(true)` and `stat_impl` now HEAD a key such as `dir/photo?part.jpg` literally, and LIST_FILE infers `image/jpeg` from that full key. This helper truncates at `?`, so TO_FILE reports `application/octet-stream` for the same object when HEAD omits Content-Type (as the Azure HEAD path currently does). The earlier raw-key thread covers object lookup; this is a separate metadata error. Derive the suffix from the parsed key and test LIST_FILE-to-TO_FILE on such a key. ########## be/src/exec/common/data_gen_functions/vlist_file_tvf.cpp: ########## @@ -0,0 +1,142 @@ +// 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 "exec/common/data_gen_functions/vlist_file_tvf.h" + +#include <gen_cpp/PaloInternalService_types.h> +#include <gen_cpp/PlanNodes_types.h> + +#include <algorithm> +#include <array> + +#include "core/data_type/data_type_file.h" +#include "core/field.h" +#include "core/value/file_value.h" +#include "io/file_factory.h" +#include "io/fs/s3_file_system.h" +#include "runtime/runtime_state.h" +#include "util/s3_uri.h" + +namespace doris { +VListFileTVF::VListFileTVF(TupleId tuple_id, const TupleDescriptor* tuple_desc) + : VDataGenFunctionInf(tuple_id, tuple_desc) {} + +VListFileTVF::~VListFileTVF() = default; + +Status VListFileTVF::set_scan_ranges(const std::vector<TScanRangeParams>& scan_ranges) { + DCHECK(scan_ranges.size() == 1); + const auto& params = scan_ranges[0].scan_range.data_gen_scan_range.list_file_params; + const auto key_start = params.uri.find('/', 5); Review Comment: [P2] Parse HTTP(S) listing URIs before splitting the bucket and prefix. With `https://host/bucket/dir`, `find('/', 5)` finds the slash in `://`, so the truncated `https:` parses with an empty bucket and `_fetch_next_page` builds `s3://`, which fails before listing. The shared S3URI parser and TO_FILE support this URI form, and FE forwards it unchanged. Parse the full URI and use its bucket/key, then test HTTP(S) LIST_FILE prefixes. ########## be/src/exprs/function/function_to_file.cpp: ########## @@ -0,0 +1,203 @@ +// 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 "exprs/function/function_to_file.h" + +#include <mutex> +#include <string_view> + +#include "core/column/column_const.h" +#include "core/column/column_file.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type_file.h" +#include "core/value/file_value.h" +#include "exprs/function/simple_function_factory.h" +#include "io/file_factory.h" +#include "runtime/query_context.h" +#include "runtime/runtime_state.h" +#include "util/s3_uri.h" + +namespace doris { +namespace { + +struct ToFileState { + explicit ToFileState(std::shared_ptr<const TS3Resource> resource_) + : resource(std::move(resource_)) {} + + const std::shared_ptr<const TS3Resource> resource; + std::shared_ptr<io::FileSystem> filesystem; + std::string bucket; + std::mutex mutex; + io::FileStatContext stat_context; +}; + +// Only the path participates in extension inference. Keep the stored/stat URI unchanged. +std::string_view to_file_uri_path(std::string_view uri) { + uri = uri.substr(0, uri.find_first_of("?#")); + const auto colon = uri.find(':'); + if (colon == std::string_view::npos) return uri; + auto path = uri.substr(colon + 1); + if (path.starts_with("//")) { + const auto slash = path.find('/', 2); + return slash == std::string_view::npos ? std::string_view() : path.substr(slash); + } + return path; +} + +std::string to_file_content_type(std::string_view uri, const io::FileStat& metadata) { + if (metadata.content_type && is_valid_file_content_type(*metadata.content_type)) { + return *metadata.content_type; + } + const auto path = to_file_uri_path(uri); + const auto filename = path.substr(path.find_last_of('/') + 1); + return std::string(infer_file_content_type_from_name(filename)); +} + +} // namespace + +DataTypePtr FunctionToFile::get_return_type_impl(const ColumnsWithTypeAndName& arguments) const { + const auto& resource = arguments[0]; + const auto& uri = arguments[1]; + if (!is_string_type(remove_nullable(resource.type)->get_primitive_type()) || !resource.column || + !is_column_const(*resource.column) || resource.column->is_null_at(0)) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "TO_FILE resource must be a non-NULL constant string"); + } + if (!is_string_type(remove_nullable(uri.type)->get_primitive_type()) && + !uri.type->is_null_literal()) { + throw Exception(ErrorCode::INVALID_ARGUMENT, "TO_FILE URI must be string-like or NULL"); + } + DataTypePtr result = std::make_shared<DataTypeFile>(); + return uri.type->is_nullable() || uri.type->is_null_literal() ? make_nullable(result) : result; +} + +Status FunctionToFile::create_filesystem(const TS3Resource& resource, const std::string& uri, + std::shared_ptr<io::FileSystem>* filesystem) const { + io::FSPropertiesRef properties(TFileType::FILE_S3); + properties.properties = &resource.properties; + io::FileDescription description; + S3URI s3_uri(uri); + RETURN_IF_ERROR(s3_uri.parse(true)); + description.path = "s3://" + s3_uri.get_bucket(); + *filesystem = DORIS_TRY(FileFactory::create_fs(properties, description)); + return Status::OK(); +} + +Status FunctionToFile::open(FunctionContext* context, FunctionContext::FunctionStateScope scope) { + if (scope == FunctionContext::THREAD_LOCAL || context->get_function_state(scope)) { + return Status::OK(); + } + auto* runtime = context->state(); + if (!runtime || !runtime->get_query_ctx()) { + return Status::InvalidArgument("TO_FILE requires query resources"); + } + const auto& resources = runtime->get_query_ctx()->get_s3_resources(); + if (!resources) { + return Status::InvalidArgument("TO_FILE requires query resources"); + } + const auto* resource_column = context->get_constant_col(0); + if (!resource_column || resource_column->column_ptr->is_null_at(0)) { + return Status::InvalidArgument("TO_FILE resource must be a non-NULL constant string"); + } + const auto resource_name = resource_column->column_ptr->get_data_at(0).to_string(); + const auto resource = resources->find(resource_name); + if (resource == resources->end()) { + return Status::InvalidArgument("TO_FILE resource '{}' is missing from query resources", + resource_name); + } + // Defer client creation until a non-NULL URI needs it. + // Share the immutable query map's ownership without copying this resource per function. + auto state = std::make_shared<ToFileState>( + std::shared_ptr<const TS3Resource>(resources, &resource->second)); + // RuntimeState outlives the fragment contexts, including their clones. + state->stat_context.is_cancelled = [runtime] { return runtime->is_cancelled(); }; + context->set_function_state(scope, std::move(state)); + return Status::OK(); +} + +Status FunctionToFile::close(FunctionContext* context, FunctionContext::FunctionStateScope scope) { + if (scope == FunctionContext::FRAGMENT_LOCAL) context->set_function_state(scope, nullptr); + return Status::OK(); +} + +Status FunctionToFile::execute_impl(FunctionContext* context, Block& block, + const ColumnNumbers& arguments, uint32_t result, + size_t input_rows_count) const { + auto* state = + static_cast<ToFileState*>(context->get_function_state(FunctionContext::FRAGMENT_LOCAL)); + if (!state) return Status::InternalError("TO_FILE has no fragment-local state"); + const auto& source = block.get_by_position(arguments[1]).column; + auto output = ColumnFile::create(); + auto nulls = ColumnUInt8::create(input_rows_count, 0); + for (size_t row = 0; row < input_rows_count; ++row) { + if (context->state()) RETURN_IF_CANCELLED(context->state()); + if (source->is_null_at(row)) { + output->insert_default(); + nulls->get_data()[row] = 1; + continue; + } + const auto uri = source->get_data_at(row).to_string(); + io::FileStat metadata; + std::string bucket; + S3URI s3_uri(uri); + RETURN_IF_ERROR(s3_uri.parse(true)); + bucket = s3_uri.get_bucket(); + std::shared_ptr<io::FileSystem> filesystem; + { + // FunctionContext clones share client initialization and bucket switching. + std::lock_guard lock(state->mutex); + if (context->state()) RETURN_IF_CANCELLED(context->state()); + // S3FileSystem binds its bucket at construction. Keep only the current + // filesystem and switch when a row references a different bucket. + if (!state->filesystem || state->bucket != bucket) { + std::shared_ptr<io::FileSystem> new_filesystem; + RETURN_IF_ERROR(create_filesystem(*state->resource, uri, &new_filesystem)); + DORIS_CHECK(new_filesystem != nullptr); + state->filesystem = std::move(new_filesystem); + state->bucket = std::move(bucket); + if (context->state()) RETURN_IF_CANCELLED(context->state()); + } + filesystem = state->filesystem; + } + // Pin this request's filesystem across concurrent bucket switches. Remote + // metadata calls must not serialize all scanners sharing the fragment state. + const auto stat_status = filesystem->stat(uri, &metadata, &state->stat_context); + if (context->state()) RETURN_IF_CANCELLED(context->state()); + RETURN_IF_ERROR(stat_status); + auto content_type = to_file_content_type(uri, metadata); Review Comment: [P2] Populate Azure HEAD metadata before constructing the FILE value. For an `S3_PROVIDER=AZURE` resource, `AzureObjStorageClient::head_object` reads BlobProperties but returns only BlobSize. This TO_FILE path then guesses `content_type` from the suffix and leaves `checksum` NULL even when the blob has explicit Content-Type and ETag; for example, `payload.bin` stored as `text/csv` becomes `application/octet-stream`. Pass those properties through `ObjStorageHeadResult` and cover an Azure blob whose declared type differs from its name. ########## common/cpp/obj-client/azure_obj_storage_client.cpp: ########## @@ -363,36 +363,40 @@ ObjStorageListPageResult AzureObjStorageClient::list_objects_page( TEST_SYNC_POINT_CALLBACK("AzureObjStorageClient::list_objects", &request); try { - auto response = [&]() { - client_bvar::ScopedLatency scoped_latency(client_bvar::s3_list_latency); - return _client->ListBlobs(request); - }(); - const bool has_more = response.NextPageToken.HasValue(); - auto next_token = has_more ? response.NextPageToken.Value() : std::string {}; - if (has_more && next_token.empty()) { - return { - .resp = {.status = {ObjStorageStatus::INTERNAL_ERROR, - "Azure list response has an empty continuation token"}, - .http_code = 0}, - }; - } - ObjStorageListPageResult page {.resp = ObjStorageResponse::OK(), - .continuation_token = std::move(next_token), - .has_more = has_more}; - page.objects.reserve(response.Blobs.size()); - for (auto&& item : response.Blobs) { - DCHECK(item.Name.starts_with(*request.Prefix)) << item.Name << ' ' << *request.Prefix; - page.objects.emplace_back(ObjectMeta { - .key = std::move(item.Name), - .size = item.BlobSize, - // `Azure::DateTime` adds the offset of `SystemClockEpoch` to the given Unix timestamp, - // so here we need to subtract this offset to obtain the Unix timestamp of the mtime. - // https://github.com/Azure/azure-sdk-for-cpp/blob/azure-core_1.12.0/sdk/core/azure-core/inc/azure/core/datetime.hpp#L129 - .mtime_s = duration_cast<std::chrono::seconds>(item.Details.LastModified - - SystemClockEpoch) - .count()}); + auto make_page = [&](auto response) -> ObjStorageListPageResult { + const bool has_more = response.NextPageToken.HasValue(); + auto next_token = has_more ? response.NextPageToken.Value() : std::string {}; + if (has_more && next_token.empty()) { + return { + .resp = {.status = {ObjStorageStatus::INTERNAL_ERROR, + "Azure list response has an empty continuation token"}, + .http_code = 0}, + }; + } + ObjStorageListPageResult page {.resp = ObjStorageResponse::OK(), + .continuation_token = std::move(next_token), + .has_more = has_more}; + page.objects.reserve(response.Blobs.size()); + for (auto&& item : response.Blobs) { + DCHECK(item.Name.starts_with(*request.Prefix)) + << item.Name << ' ' << *request.Prefix; + page.objects.emplace_back(ObjectMeta { + .key = std::move(item.Name), Review Comment: [P2] Preserve Azure's listed ETag in `ObjectMeta`. This new mapper copies name, size, and modification time, but leaves `etag` empty although BlobItem details contain it. LIST_FILE only fills the FILE checksum from `ObjectMeta.etag`, so every Azure row reports NULL while the S3 path reports `ETAG:<value>`. This is separate from Azure HEAD metadata used by TO_FILE; fixing that path alone leaves listed FILE values incomplete. Copy the listed ETag and test Azure LIST_FILE checksum. -- 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]
