github-actions[bot] commented on code in PR #65478: URL: https://github.com/apache/doris/pull/65478#discussion_r3563265563
########## be/src/format_v2/orc/orc_file_input_stream.cpp: ########## @@ -0,0 +1,355 @@ +// 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 "format_v2/orc/orc_file_input_stream.h" + +#include <algorithm> +#include <cstring> +#include <utility> + +#include "common/status.h" +#include "io/fs/buffered_reader.h" +#include "io/fs/tracing_file_reader.h" +#include "io/io_common.h" +#include "runtime/runtime_profile.h" +#include "util/slice.h" + +namespace doris::format::orc { +namespace { + +struct OrcMergedRangeStatistics { + int64_t copy_time = 0; + int64_t read_time = 0; + int64_t request_io = 0; + int64_t merged_io = 0; + int64_t request_bytes = 0; + int64_t merged_bytes = 0; +}; + +class OrcMergedRangeFileReader final : public io::FileReader { +public: + OrcMergedRangeFileReader(RuntimeProfile* profile, io::FileReaderSPtr file_reader, + io::PrefetchRange range) + : _profile(profile), + _file_reader(std::move(file_reader)), + _range(range), + _size(_file_reader->size()) { + if (_profile != nullptr) { + const char* profile_name = "MergedSmallIO"; + ADD_TIMER_WITH_LEVEL(_profile, profile_name, 1); + _copy_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "CopyTime", profile_name, 1); + _read_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ReadTime", profile_name, 1); + _request_io = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "RequestIO", TUnit::UNIT, + profile_name, 1); + _merged_io = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "MergedIO", TUnit::UNIT, + profile_name, 1); + _request_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "RequestBytes", TUnit::BYTES, + profile_name, 1); + _merged_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "MergedBytes", TUnit::BYTES, + profile_name, 1); + } + } + + Status close() override { + _closed = true; + return Status::OK(); + } + + const io::Path& path() const override { return _file_reader->path(); } + size_t size() const override { return _size; } + bool closed() const override { return _closed; } + int64_t mtime() const override { return _file_reader->mtime(); } + +protected: + Status read_at_impl(size_t offset, Slice result, size_t* bytes_read, + const io::IOContext* io_ctx) override { + ++_statistics.request_io; + _statistics.request_bytes += static_cast<int64_t>(result.size); + *bytes_read = 0; + if (result.size == 0) { + return Status::OK(); + } + if (offset < _range.start_offset || offset + result.size > _range.end_offset) { + return Status::IOError("ORC stripe read [{}, {}) is outside merged range [{}, {})", + offset, offset + result.size, _range.start_offset, + _range.end_offset); + } + + RETURN_IF_ERROR(_load(io_ctx)); + { + SCOPED_RAW_TIMER(&_statistics.copy_time); + std::memcpy(result.data, _cache.data() + offset - _range.start_offset, result.size); + } + *bytes_read = result.size; + return Status::OK(); + } + + void _collect_profile_before_close() override { + if (_profile == nullptr) { + return; + } + COUNTER_UPDATE(_copy_time, _statistics.copy_time); + COUNTER_UPDATE(_read_time, _statistics.read_time); + COUNTER_UPDATE(_request_io, _statistics.request_io); + COUNTER_UPDATE(_merged_io, _statistics.merged_io); + COUNTER_UPDATE(_request_bytes, _statistics.request_bytes); + COUNTER_UPDATE(_merged_bytes, _statistics.merged_bytes); + } + +private: + Status _load(const io::IOContext* io_ctx) { + if (_loaded) { + return Status::OK(); + } + + const size_t range_size = _range.end_offset - _range.start_offset; + _cache.resize(range_size); + size_t total_read = 0; + { + SCOPED_RAW_TIMER(&_statistics.read_time); + while (total_read < range_size) { + size_t loop_read = 0; + RETURN_IF_ERROR(_file_reader->read_at( + _range.start_offset + total_read, + Slice(_cache.data() + total_read, range_size - total_read), &loop_read, + io_ctx)); + ++_statistics.merged_io; + _statistics.merged_bytes += static_cast<int64_t>(loop_read); + if (loop_read == 0) { + return Status::IOError("Short read for ORC merged range [{}, {})", + _range.start_offset, _range.end_offset); + } + total_read += loop_read; + } + } + _loaded = true; + return Status::OK(); + } + + RuntimeProfile* _profile = nullptr; + io::FileReaderSPtr _file_reader; + io::PrefetchRange _range; + size_t _size = 0; + bool _closed = false; + bool _loaded = false; + std::vector<char> _cache; + OrcMergedRangeStatistics _statistics; + + RuntimeProfile::Counter* _copy_time = nullptr; + RuntimeProfile::Counter* _read_time = nullptr; + RuntimeProfile::Counter* _request_io = nullptr; + RuntimeProfile::Counter* _merged_io = nullptr; + RuntimeProfile::Counter* _request_bytes = nullptr; + RuntimeProfile::Counter* _merged_bytes = nullptr; +}; + +class OrcStripeInputStream final : public ::orc::InputStream { +public: + OrcStripeInputStream(std::string file_name, io::FileReaderSPtr file_reader, + const io::IOContext* io_ctx, uint64_t natural_read_size) + : _file_name(std::move(file_name)), + _file_reader(std::move(file_reader)), + _io_ctx(io_ctx), + _natural_read_size(natural_read_size) {} + + uint64_t getLength() const override { return _file_reader->size(); } + uint64_t getNaturalReadSize() const override { return _natural_read_size; } + + void read(void* buf, uint64_t length, uint64_t offset) override { + uint64_t bytes_read = 0; + auto* out = static_cast<uint8_t*>(buf); + while (bytes_read < length) { + if (_io_ctx != nullptr && _io_ctx->should_stop) { + throw ::orc::ParseError("stop"); + } + size_t loop_read = 0; + Status st = _file_reader->read_at( + static_cast<size_t>(offset + bytes_read), + Slice(out + bytes_read, static_cast<size_t>(length - bytes_read)), &loop_read, + _io_ctx); + if (!st.ok()) { + throw ::orc::ParseError("Failed to read " + _file_name + ": " + + st.to_string_no_stack()); + } + if (loop_read == 0) { + break; + } + bytes_read += loop_read; + } + if (bytes_read != length) { + throw ::orc::ParseError("Short read from " + _file_name); + } + } + + const std::string& getName() const override { return _file_name; } + +private: + std::string _file_name; + io::FileReaderSPtr _file_reader; + const io::IOContext* _io_ctx = nullptr; + uint64_t _natural_read_size = 0; +}; + +struct StripeStreamRange { + ::orc::StreamId stream_id; + io::PrefetchRange range; +}; + +} // namespace + +OrcFileInputStream::OrcFileInputStream(std::string file_name, io::FileReaderSPtr file_reader, + const io::IOContext* io_ctx, RuntimeProfile* profile, + OrcFileInputStreamOptions options) + : _file_name(std::move(file_name)), + _file_reader(std::move(file_reader)), + _default_reader(io_ctx != nullptr && io_ctx->file_reader_stats != nullptr + ? std::make_shared<io::TracingFileReader>( + _file_reader, io_ctx->file_reader_stats) + : _file_reader), + _io_ctx(io_ctx), + _profile(profile), + _options(options) { + DORIS_CHECK_GT(_options.natural_read_size, 0); + DORIS_CHECK_GT(_options.once_max_read_bytes, 0); Review Comment: This assertion makes the new v2 ORC path incompatible with an already-accepted session value. `SET orc_once_max_read_bytes = 0` is allowed by `SessionVariable.setOrcOnceMaxReadBytes()` because it validates with a minimum of 0, and `OrcReader::init()` forwards that value directly into these options. In the existing v1 ORC path, zero just keeps positive-length streams out of the small-range merge set, so the query can continue with direct reads; with this constructor it becomes a BE crash during reader initialization. Please preserve that zero-value behavior here, for example by treating `once_max_read_bytes <= 0` as no merging/direct streams, or reject zero at the session-variable boundary consistently before it reaches BE. -- 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]
