xylaaaaa commented on code in PR #65183: URL: https://github.com/apache/doris/pull/65183#discussion_r3527055433
########## be/src/format_v2/orc/orc_reader.cpp: ########## @@ -0,0 +1,2750 @@ +// 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_reader.h" + +#include <cctz/time_zone.h> +#include <gen_cpp/Types_types.h> + +#include <algorithm> +#include <array> +#include <atomic> +#include <cctype> +#include <charconv> +#include <cmath> +#include <cstdint> +#include <cstring> +#include <list> +#include <map> +#include <memory> +#include <optional> +#include <orc/OrcFile.hh> +#include <orc/Vector.hh> +#include <orc/sargs/Literal.hh> +#include <orc/sargs/SearchArgument.hh> +#include <set> +#include <string> +#include <string_view> +#include <system_error> +#include <type_traits> +#include <utility> +#include <vector> + +#include "common/cast_set.h" +#include "common/consts.h" +#include "common/exception.h" +#include "core/block/block.h" +#include "core/column/column_array.h" +#include "core/column/column_decimal.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_struct.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_date_or_datetime_v2.h" +#include "core/data_type/data_type_date_time.h" +#include "core/data_type/data_type_decimal.h" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_struct.h" +#include "core/data_type_serde/data_type_serde.h" +#include "core/types.h" +#include "core/value/vdatetime_value.h" +#include "exprs/vexpr_context.h" +#include "exprs/vliteral.h" +#include "exprs/vslot_ref.h" +#include "format_v2/column_mapper.h" +#include "format_v2/orc/orc_search_argument.h" +#include "io/fs/file_reader.h" +#include "runtime/exec_env.h" +#include "runtime/runtime_profile.h" +#include "storage/index/zone_map/zone_map_index.h" +#include "storage/utils.h" +#include "util/slice.h" +#include "util/timezone_utils.h" + +namespace doris::format::orc { +namespace { + +constexpr uint64_t DEFAULT_ORC_READ_BATCH_SIZE = 4096; +constexpr uint64_t DEFAULT_ORC_NATURAL_READ_SIZE = 128 * 1024; +constexpr int DECIMAL_PRECISION_FOR_HIVE11 = BeConsts::MAX_DECIMAL128_PRECISION; +constexpr int DECIMAL_SCALE_FOR_HIVE11 = 10; +constexpr const char* ORC_LIST_ELEMENT_NAME = "element"; +constexpr const char* ORC_MAP_ENTRY_NAME = "key_value"; +constexpr const char* ORC_MAP_KEY_NAME = "key"; +constexpr const char* ORC_MAP_VALUE_NAME = "value"; +constexpr const char* ORC_ICEBERG_ID_ATTRIBUTE = "iceberg.id"; + +uint64_t orc_metric_value(const std::atomic<uint64_t>& metric) { + return metric.load(std::memory_order_relaxed); +} + +template <typename Metrics, typename = void> +struct OrcReadRowCountMetric { + static uint64_t value(const Metrics&) { return 0; } +}; + +template <typename Metrics> +struct OrcReadRowCountMetric<Metrics, + std::void_t<decltype(std::declval<const Metrics&>().ReadRowCount)>> { + static uint64_t value(const Metrics& metrics) { return orc_metric_value(metrics.ReadRowCount); } +}; + +uint64_t orc_read_row_count(const ::orc::ReaderMetrics& metrics) { + return OrcReadRowCountMetric<::orc::ReaderMetrics>::value(metrics); +} + +bool is_hour_offset_timezone(std::string_view timezone) { + return timezone.size() == 6 && (timezone[0] == '+' || timezone[0] == '-') && + std::isdigit(static_cast<unsigned char>(timezone[1])) && + std::isdigit(static_cast<unsigned char>(timezone[2])) && timezone[3] == ':' && + timezone[4] == '0' && timezone[5] == '0'; +} + +Status set_orc_reader_timezone(const std::string& timezone, + ::orc::RowReaderOptions* row_reader_options) { + if (timezone == "CST") { + row_reader_options->setTimezoneName("Asia/Shanghai"); + return Status::OK(); + } + + if (!timezone.empty() && (timezone[0] == '+' || timezone[0] == '-')) { + if (!is_hour_offset_timezone(timezone)) { + return Status::NotSupported("ORC reader timezone does not support non-hour offset '{}'", + timezone); + } + + const int hour = (timezone[1] - '0') * 10 + timezone[2] - '0'; + row_reader_options->setTimezoneName( + hour == 0 ? "Etc/GMT" + : fmt::format("Etc/GMT{}{}", timezone[0] == '+' ? '-' : '+', hour)); + return Status::OK(); + } + + row_reader_options->setTimezoneName(timezone.empty() ? "UTC" : timezone); + return Status::OK(); +} + +// Thin adapter from Doris FileReader to ORC's InputStream API. Keep IO policy, +// tracing, and retry behavior in the underlying FileReader. +class DorisOrcInputStream final : public ::orc::InputStream { +public: + DorisOrcInputStream(std::string file_name, io::FileReaderSPtr file_reader, + io::IOContext* io_ctx) + : _file_name(std::move(file_name)), + _file_reader(std::move(file_reader)), + _io_ctx(io_ctx) {} + + uint64_t getLength() const override { return _file_reader->size(); } + + uint64_t getNaturalReadSize() const override { return DEFAULT_ORC_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) { Review Comment: 已加。`DorisOrcInputStream::read()` 每次实际 IO 前检查 `_io_ctx->should_stop` 并中断 ORC 读取；`init()` / `get_block()` 会把这个中断转成 EOF/OK EOF。补了 init 和 get_block 两个 should_stop UT。 -- 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]
