github-actions[bot] commented on code in PR #65183:
URL: https://github.com/apache/doris/pull/65183#discussion_r3533121327
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -234,7 +234,8 @@ Status
FileScannerV2::TEST_rewrite_slot_refs_to_global_index(
bool FileScannerV2::is_supported(const TFileScanRangeParams& params, const
TFileRangeDesc& range) {
const auto format_type = get_range_format_type(params, range);
- if (format_type == TFileFormatType::FORMAT_PARQUET) {
+ if (format_type == TFileFormatType::FORMAT_PARQUET ||
Review Comment:
This makes mixed Hive Parquet/ORC scans eligible for `FileScannerV2`, but
the v2 Hive reader still cannot handle mixed physical formats in one scanner.
The operator selects scanner v2 when `all_scan_ranges_match` returns true, and
that helper only checks each range independently. With this change, a Hive
Parquet range and a Hive ORC range both pass `is_supported`, so the whole split
source can be routed to v2. `FileScannerV2` then creates one `HiveReader` from
the first range and reuses it, while `HiveReader::prepare_split()` rejects any
later split whose `current_split_format` differs from the initialized
`_format`. Please either keep mixed-format Hive sources on the old scanner
until the v2 Hive path can rebuild per-format state, or add scanner selection
logic/tests that require one physical format per v2 Hive scanner.
##########
be/src/format_v2/orc/orc_reader.cpp:
##########
@@ -0,0 +1,2627 @@
+// 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 <limits>
+#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/segment/condition_cache.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_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_orc_stop(const io::IOContext* io_ctx, const std::exception& e) {
+ return io_ctx != nullptr && io_ctx->should_stop &&
std::string_view(e.what()) == "stop";
+}
+
+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) {
+ 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;
+ io::IOContext* _io_ctx = nullptr;
+};
+
+// selected_rows is a source-row remap produced by ORC lazy callback:
+// predicate columns are decoded first, then surviving row ids drive follower
decodes.
+bool is_null_at(const ::orc::ColumnVectorBatch& batch, size_t row) {
+ return batch.hasNulls && !batch.notNull[row];
+}
+
+size_t decode_row_count(size_t rows, const std::vector<size_t>* selected_rows)
{
+ if (selected_rows == nullptr) {
+ return rows;
+ }
+ return selected_rows->size();
+}
+
+size_t source_row_at(size_t row, const std::vector<size_t>* selected_rows) {
+ if (selected_rows == nullptr) {
+ return row;
+ }
+ return (*selected_rows)[row];
+}
+
+DecodedColumnView make_orc_decoded_view(size_t rows, const
std::vector<size_t>* selected_rows,
+ DecodedValueKind value_kind) {
+ DecodedColumnView view;
+ view.value_kind = value_kind;
+ view.row_count = cast_set<int64_t>(decode_row_count(rows, selected_rows));
+ return view;
+}
+
+void fill_orc_decoded_null_map(const ::orc::ColumnVectorBatch& batch, size_t
rows,
+ const std::vector<size_t>* selected_rows,
NullMap* null_map) {
+ DORIS_CHECK(null_map != nullptr);
+ if (!batch.hasNulls) {
+ return;
+ }
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ null_map->resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ (*null_map)[row] = !batch.notNull[source_row_at(row, selected_rows)];
+ }
+}
+
+size_t trim_right_spaces(const char* value, size_t length) {
+ while (length > 0 && value[length - 1] == ' ') {
+ --length;
+ }
+ return length;
+}
+
+Int128 to_int128(::orc::Int128 value) {
+ const auto high_bits =
static_cast<__uint128_t>(static_cast<uint64_t>(value.getHighBits()));
+ const auto low_bits = static_cast<__uint128_t>(value.getLowBits());
+ return static_cast<Int128>((high_bits << 64) | low_bits);
+}
+
+Status read_decoded_values(DataTypePtr data_type, IColumn& column,
DecodedColumnView* view) {
+ DORIS_CHECK(data_type != nullptr);
+ DORIS_CHECK(view != nullptr);
+
RETURN_IF_ERROR(data_type->get_serde()->read_column_from_decoded_values(column,
*view));
+ return Status::OK();
+}
+
+template <typename SourceType>
+void fill_selected_values(const SourceType* source_values, size_t rows,
+ const std::vector<size_t>* selected_rows,
+ std::vector<SourceType>* selected_values) {
+ DORIS_CHECK(source_values != nullptr);
+ DORIS_CHECK(selected_values != nullptr);
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ selected_values->resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ (*selected_values)[row] = source_values[source_row_at(row,
selected_rows)];
+ }
+}
+
+template <typename OrcBatchType, typename SourceType>
+Status decode_fixed_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>* selected_rows,
+ DecodedValueKind value_kind) {
+ const auto* orc_batch = dynamic_cast<const OrcBatchType*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC scalar batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows, value_kind);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ std::vector<SourceType> selected_values;
+ if (selected_rows == nullptr) {
+ view.values = reinterpret_cast<const uint8_t*>(orc_batch->data.data());
+ } else {
+ fill_selected_values(orc_batch->data.data(), rows, selected_rows,
&selected_values);
+ view.values = reinterpret_cast<const uint8_t*>(selected_values.data());
+ }
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_float_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::DoubleVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC float batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::FLOAT);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<float> float_values;
+ float_values.resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ float_values[row] =
static_cast<float>(orc_batch->data[source_row_at(row, selected_rows)]);
+ }
+ view.values = reinterpret_cast<const uint8_t*>(float_values.data());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_boolean_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::LongVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC boolean batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::BOOL);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::unique_ptr<bool[]> bool_values =
std::make_unique<bool[]>(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ bool_values[row] = orc_batch->data[source_row_at(row, selected_rows)]
!= 0;
+ }
+ view.values = reinterpret_cast<const uint8_t*>(bool_values.get());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_string_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::Type& file_type,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::StringVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC string batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::BINARY);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<StringRef> binary_values;
+ binary_values.reserve(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ const auto source_row = source_row_at(row, selected_rows);
+ if (is_null_at(batch, source_row)) {
+ binary_values.emplace_back("", 0);
+ continue;
+ }
+ auto length = static_cast<size_t>(orc_batch->length[source_row]);
+ if (file_type.getKind() == ::orc::TypeKind::CHAR) {
+ length = trim_right_spaces(orc_batch->data[source_row], length);
+ }
+ binary_values.emplace_back(length == 0 ? "" :
orc_batch->data[source_row], length);
+ }
+ view.binary_values = &binary_values;
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_date_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>* selected_rows)
{
+ const auto* orc_batch = dynamic_cast<const
::orc::LongVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC date batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::INT32);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<int32_t> date_values;
+ date_values.resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ date_values[row] =
cast_set<int32_t>(orc_batch->data[source_row_at(row, selected_rows)]);
+ }
+ view.values = reinterpret_cast<const uint8_t*>(date_values.data());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_decimal_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::Type& file_type,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::INT64);
+ view.decimal_precision = file_type.getPrecision() == 0
+ ? DECIMAL_PRECISION_FOR_HIVE11
+ : cast_set<int>(file_type.getPrecision());
+ view.decimal_scale = file_type.getPrecision() == 0 ?
DECIMAL_SCALE_FOR_HIVE11
+ :
cast_set<int>(file_type.getScale());
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ if (const auto* decimal64_batch = dynamic_cast<const
::orc::Decimal64VectorBatch*>(&batch);
+ decimal64_batch != nullptr) {
+ std::vector<int64_t> selected_values;
+ if (selected_rows == nullptr) {
+ view.values = reinterpret_cast<const
uint8_t*>(decimal64_batch->values.data());
+ } else {
+ fill_selected_values(decimal64_batch->values.data(), rows,
selected_rows,
+ &selected_values);
+ view.values = reinterpret_cast<const
uint8_t*>(selected_values.data());
+ }
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column,
&view));
+ return Status::OK();
+ }
+ const auto* decimal128_batch = dynamic_cast<const
::orc::Decimal128VectorBatch*>(&batch);
+ if (decimal128_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC decimal batch type {}",
batch.toString());
+ }
+ std::vector<StringRef> binary_values;
+ std::vector<std::array<uint8_t, sizeof(Int128)>> decimal_values;
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ decimal_values.resize(output_rows);
+ binary_values.reserve(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ auto value = __uint128_t();
+ const auto source_row = source_row_at(row, selected_rows);
+ if (!is_null_at(batch, source_row)) {
+ value =
static_cast<__uint128_t>(to_int128(decimal128_batch->values[source_row]));
+ }
+ for (size_t byte_idx = 0; byte_idx < decimal_values[row].size();
++byte_idx) {
+ const auto shift = (decimal_values[row].size() - byte_idx - 1) * 8;
+ decimal_values[row][byte_idx] = static_cast<uint8_t>(value >>
shift);
+ }
+ binary_values.emplace_back(reinterpret_cast<const
char*>(decimal_values[row].data()),
+ decimal_values[row].size());
+ }
+ view.value_kind = DecodedValueKind::FIXED_BINARY;
+ view.fixed_length = sizeof(Int128);
+ view.binary_values = &binary_values;
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+// ORC nested projection is type-id based. These helpers translate Doris'
+// LocalColumnIndex tree into the ORC type ids expected by includeTypes().
+Status get_projection_child_index(const format::LocalColumnIndex& child,
int32_t child_count,
+ const std::string& column_name, int32_t*
child_idx) {
+ DORIS_CHECK(child_idx != nullptr);
+ *child_idx = child.local_id();
+ if (*child_idx < 0 || *child_idx >= child_count) {
+ return Status::InvalidArgument("Invalid ORC projection child index {}
for column {}",
+ *child_idx, column_name);
+ }
+ return Status::OK();
+}
+
+void collect_type_and_descendant_ids(const ::orc::Type& type,
std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ collect_type_and_descendant_ids(*child_type, type_ids);
+ }
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC MAP projection for column {} contains
no children",
+ projection.local_id());
+ }
+
+ bool selected_key = false;
+ bool selected_value = false;
+ for (const auto& key_value_projection : projection.children) {
+ int32_t key_value_idx = 0;
+ RETURN_IF_ERROR(
+ get_projection_child_index(key_value_projection, 2, "orc_map",
&key_value_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(key_value_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
key_value_projection, type_ids));
+ selected_key = selected_key || key_value_idx == 0;
+ selected_value = selected_value || key_value_idx == 1;
+ }
+ if (!selected_key || !selected_value) {
+ return Status::NotSupported("ORC MAP projection must include both key
and value");
+ }
+ return Status::OK();
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC projection contains no children");
+ }
+ if (type.getKind() == ::orc::TypeKind::MAP) {
+ return collect_projected_map_type_ids(type, projection, type_ids);
+ }
+ if (type.getKind() != ::orc::TypeKind::STRUCT && type.getKind() !=
::orc::TypeKind::LIST) {
+ return Status::InvalidArgument("Cannot project children from
non-complex ORC type {}",
+ static_cast<int>(type.getKind()));
+ }
+
+ const auto child_count = static_cast<int32_t>(type.getSubtypeCount());
+ for (const auto& child_projection : projection.children) {
+ int32_t child_idx = 0;
+ RETURN_IF_ERROR(get_projection_child_index(child_projection,
child_count, "orc_complex",
+ &child_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(child_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
child_projection, type_ids));
+ }
+ return Status::OK();
+}
+
+// For arrays/maps, selected parent rows expand into selected element rows. The
+// returned offsets are compacted so downstream child decoding can append
densely.
+Status append_orc_offsets(ColumnArray::Offsets64& doris_offsets,
+ const ::orc::DataBuffer<int64_t>& orc_offsets,
size_t rows,
+ size_t* element_size, const std::vector<size_t>*
selected_rows = nullptr,
+ std::vector<size_t>* element_selection = nullptr) {
+ if (selected_rows != nullptr) {
+ DORIS_CHECK(element_selection != nullptr);
+ const auto prev_offset = doris_offsets.empty() ? 0 :
doris_offsets.back();
+ ColumnArray::Offset64 current_offset = prev_offset;
+ element_selection->clear();
+ for (size_t row = 0; row < selected_rows->size(); ++row) {
+ const auto source_row = (*selected_rows)[row];
+ DORIS_CHECK(source_row < rows);
+ const auto begin_offset = orc_offsets[source_row];
+ const auto end_offset = orc_offsets[source_row + 1];
+ if (end_offset < begin_offset) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ const auto delta = static_cast<size_t>(end_offset - begin_offset);
+ for (size_t element_idx = 0; element_idx < delta; ++element_idx) {
+ element_selection->push_back(static_cast<size_t>(begin_offset)
+ element_idx);
+ }
+ current_offset += static_cast<ColumnArray::Offset64>(delta);
+ doris_offsets.push_back(current_offset);
+ }
+ *element_size = element_selection->size();
+ return Status::OK();
+ }
+
+ const auto prev_offset = doris_offsets.empty() ? 0 : doris_offsets.back();
+ const auto base_offset = orc_offsets[0];
+ for (size_t idx = 1; idx <= rows; ++idx) {
+ const auto delta = orc_offsets[idx] - base_offset;
+ if (delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ doris_offsets.push_back(prev_offset +
static_cast<ColumnArray::Offset64>(delta));
+ }
+ const auto total_delta = orc_offsets[rows] - base_offset;
+ if (total_delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ *element_size = static_cast<size_t>(total_delta);
+ return Status::OK();
+}
+
+int64_t find_struct_child_index(const ::orc::Type& type, const std::string&
field_name) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ if (type.getFieldName(child_idx) == field_name) {
+ return static_cast<int64_t>(child_idx);
+ }
+ }
+ return -1;
+}
+
+bool is_row_position_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(format::ROW_POSITION_COLUMN_ID);
+}
+
+bool is_global_rowid_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(format::GLOBAL_ROWID_COLUMN_ID);
+}
+
+bool is_virtual_column(format::LocalColumnId file_column_id) {
+ return is_row_position_column(file_column_id) ||
is_global_rowid_column(file_column_id);
+}
+
+format::ColumnDefinition nullable_global_rowid_column_definition() {
+ auto field = format::global_rowid_column_definition();
+ field.type = make_nullable(field.type);
+ return field;
+}
+
+const format::LocalColumnIndex* find_projection(
+ const std::vector<format::LocalColumnIndex>& projections,
+ format::LocalColumnId file_column_id) {
+ const auto it = std::find_if(projections.begin(), projections.end(),
+ [&](const format::LocalColumnIndex&
projection) {
+ return projection.column_id() ==
file_column_id;
+ });
+ return it == projections.end() ? nullptr : &*it;
+}
+
+bool local_column_ids_are_unique(const std::vector<format::LocalColumnIndex>&
projections) {
+ std::set<format::LocalColumnId> column_ids;
+ for (const auto& projection : projections) {
+ if (!column_ids.insert(projection.column_id()).second) {
+ return false;
+ }
+ }
+ return true;
+}
+
+const format::LocalColumnIndex* find_request_projection(const
format::FileScanRequest& request,
+ format::LocalColumnId
file_column_id) {
+ if (const auto* projection = find_projection(request.predicate_columns,
file_column_id);
+ projection != nullptr) {
+ return projection;
+ }
+ return find_projection(request.non_predicate_columns, file_column_id);
+}
+
+bool has_pruned_projection(const format::LocalColumnIndex& projection) {
+ return !projection.project_all_children;
+}
+
+Status collect_lazy_filter_type_ids(const ::orc::Type& root_type,
+ const
std::vector<format::LocalColumnIndex>& projections,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ for (const auto& projection : projections) {
+ const auto file_column_id = projection.column_id();
+ if (is_virtual_column(file_column_id)) {
+ continue;
+ }
+ const auto* type =
root_type.getSubtype(static_cast<uint64_t>(file_column_id.value()));
+ DORIS_CHECK(type != nullptr);
+ if (!has_pruned_projection(projection)) {
+ collect_type_and_descendant_ids(*type, type_ids);
+ continue;
+ }
+ RETURN_IF_ERROR(collect_projected_type_ids(*type, projection,
type_ids));
+ }
+ return Status::OK();
+}
+
+// Stripe pruning maps ORC stripe statistics into Doris ZoneMap semantics.
Missing
+// or unsupported statistics are treated conservatively and never prune.
+bool set_integer_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* integer_statistics =
+ dynamic_cast<const ::orc::IntegerColumnStatistics*>(&statistics);
+ if (integer_statistics == nullptr || !integer_statistics->hasMinimum() ||
+ !integer_statistics->hasMaximum()) {
+ return false;
+ }
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BYTE:
+ zone_map->min_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::SHORT:
+ zone_map->min_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::INT:
+ zone_map->min_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::LONG:
+ zone_map->min_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMaximum());
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool set_boolean_zone_map(const ::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* boolean_statistics =
+ dynamic_cast<const ::orc::BooleanColumnStatistics*>(&statistics);
+ if (boolean_statistics == nullptr || !boolean_statistics->hasCount()) {
+ return false;
+ }
+ const bool has_false = boolean_statistics->getFalseCount() > 0;
+ const bool has_true = boolean_statistics->getTrueCount() > 0;
+ if (!has_false && !has_true) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_false ? 0 : 1));
+ zone_map->max_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_true ? 1 : 0));
+ return true;
+}
+
+bool set_floating_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* double_statistics = dynamic_cast<const
::orc::DoubleColumnStatistics*>(&statistics);
+ if (double_statistics == nullptr || !double_statistics->hasMinimum() ||
+ !double_statistics->hasMaximum()) {
+ return false;
+ }
+ if (type.getKind() == ::orc::TypeKind::FLOAT) {
+ zone_map->min_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMaximum()));
+ return true;
+ }
+ if (type.getKind() == ::orc::TypeKind::DOUBLE) {
+ zone_map->min_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMaximum());
+ return true;
+ }
+ return false;
+}
+
+bool set_string_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* string_statistics = dynamic_cast<const
::orc::StringColumnStatistics*>(&statistics);
+ if (string_statistics == nullptr || !string_statistics->hasMinimum() ||
+ !string_statistics->hasMaximum()) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_STRING>(string_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_STRING>(string_statistics->getMaximum());
+ return true;
+}
+
+bool set_date_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* date_statistics = dynamic_cast<const
::orc::DateColumnStatistics*>(&statistics);
+ if (date_statistics == nullptr || !date_statistics->hasMinimum() ||
+ !date_statistics->hasMaximum()) {
+ return false;
+ }
+ auto& date_dict = date_day_offset_dict::get();
+ zone_map->min_value =
+
Field::create_field<TYPE_DATEV2>(date_dict[date_statistics->getMinimum()]);
+ zone_map->max_value =
+
Field::create_field<TYPE_DATEV2>(date_dict[date_statistics->getMaximum()]);
+ return true;
+}
+
+DateV2Value<DateTimeV2ValueType> datetime_v2_from_orc_millis(int64_t millis,
int32_t nanos_tail) {
+ int64_t seconds = millis / 1000;
+ int64_t millis_remainder = millis % 1000;
+ if (millis_remainder < 0) {
+ --seconds;
+ millis_remainder += 1000;
+ }
+ const auto extra_nanos = std::max<int32_t>(nanos_tail, 0);
+ const auto microseconds = cast_set<uint64_t>(millis_remainder * 1000 +
extra_nanos / 1000);
+ DateV2Value<DateTimeV2ValueType> value;
+ value.from_unixtime(seconds, cctz::utc_time_zone());
+ value.set_microsecond(microseconds);
+ return value;
+}
+
+bool set_timestamp_zone_map(const ::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* timestamp_statistics =
+ dynamic_cast<const ::orc::TimestampColumnStatistics*>(&statistics);
+ if (timestamp_statistics == nullptr || !timestamp_statistics->hasMinimum()
||
+ !timestamp_statistics->hasMaximum()) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_DATETIMEV2>(datetime_v2_from_orc_millis(
+ timestamp_statistics->getMinimum(),
timestamp_statistics->getMinimumNanos()));
+ zone_map->max_value =
Field::create_field<TYPE_DATETIMEV2>(datetime_v2_from_orc_millis(
+ timestamp_statistics->getMaximum(),
timestamp_statistics->getMaximumNanos()));
+ return true;
+}
+
+int32_t decimal_scale_for_orc_type(const ::orc::Type& type) {
+ return type.getPrecision() == 0 ? DECIMAL_SCALE_FOR_HIVE11 :
cast_set<int32_t>(type.getScale());
+}
+
+std::optional<Decimal128V3> decimal_value_at_scale(const ::orc::Decimal&
decimal,
+ int32_t target_scale) {
+ if (decimal.scale == target_scale) {
+ return Decimal128V3(to_int128(decimal.value));
+ }
+ if (decimal.scale < target_scale) {
+ bool overflow = false;
+ const auto scaled = ::orc::scaleUpInt128ByPowerOfTen(
+ decimal.value, target_scale - decimal.scale, overflow);
+ if (overflow) {
+ return std::nullopt;
+ }
+ return Decimal128V3(to_int128(scaled));
+ }
+
+ const auto scale_diff = decimal.scale - target_scale;
+ const auto scaled = ::orc::scaleDownInt128ByPowerOfTen(decimal.value,
scale_diff);
+ bool overflow = false;
+ const auto restored = ::orc::scaleUpInt128ByPowerOfTen(scaled, scale_diff,
overflow);
+ if (overflow || restored != decimal.value) {
+ return std::nullopt;
+ }
+ return Decimal128V3(to_int128(scaled));
+}
+
+bool set_decimal_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* decimal_statistics =
+ dynamic_cast<const ::orc::DecimalColumnStatistics*>(&statistics);
+ if (decimal_statistics == nullptr || !decimal_statistics->hasMinimum() ||
+ !decimal_statistics->hasMaximum()) {
+ return false;
+ }
+ const auto min = decimal_statistics->getMinimum();
+ const auto max = decimal_statistics->getMaximum();
+ const auto expected_scale = decimal_scale_for_orc_type(type);
+ const auto min_value = decimal_value_at_scale(min, expected_scale);
+ const auto max_value = decimal_value_at_scale(max, expected_scale);
+ if (!min_value.has_value() || !max_value.has_value()) {
+ return false;
+ }
+ zone_map->min_value = Field::create_field<TYPE_DECIMAL128I>(*min_value);
+ zone_map->max_value = Field::create_field<TYPE_DECIMAL128I>(*max_value);
+ return true;
+}
+
+bool build_zone_map_from_orc_statistics(const ::orc::Type& type,
+ const ::orc::ColumnStatistics&
statistics,
+ segment_v2::ZoneMap* zone_map) {
+ DORIS_CHECK(zone_map != nullptr);
+ zone_map->has_null = statistics.hasNull();
+ zone_map->has_not_null = statistics.getNumberOfValues() > 0;
+ if (!zone_map->has_not_null) {
+ return true;
+ }
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BOOLEAN:
+ return set_boolean_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::BYTE:
+ case ::orc::TypeKind::SHORT:
+ case ::orc::TypeKind::INT:
+ case ::orc::TypeKind::LONG:
+ return set_integer_zone_map(type, statistics, zone_map);
+ case ::orc::TypeKind::FLOAT:
+ case ::orc::TypeKind::DOUBLE:
+ return set_floating_zone_map(type, statistics, zone_map);
+ case ::orc::TypeKind::STRING:
+ case ::orc::TypeKind::VARCHAR:
+ case ::orc::TypeKind::CHAR:
+ return set_string_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::DATE:
+ return set_date_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::TIMESTAMP:
+ case ::orc::TypeKind::TIMESTAMP_INSTANT:
+ return set_timestamp_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::DECIMAL:
+ return set_decimal_zone_map(type, statistics, zone_map);
+ default:
+ return false;
+ }
+}
+
+Status find_projected_minmax_leaf_in_type(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ const ::orc::Type** leaf_type) {
+ DORIS_CHECK(leaf_type != nullptr);
+ if (projection.project_all_children || projection.children.empty()) {
+ if (type.getSubtypeCount() > 0) {
+ return Status::NotSupported(
+ "ORC aggregate pushdown only supports primitive column
kind {}",
+ static_cast<int>(type.getKind()));
+ }
+ *leaf_type = &type;
+ return Status::OK();
+ }
+ if (projection.children.size() != 1) {
+ return Status::NotSupported(
+ "ORC aggregate pushdown only supports a single nested leaf
under column kind {}",
+ static_cast<int>(type.getKind()));
+ }
+ if (type.getKind() != ::orc::TypeKind::STRUCT) {
+ return Status::NotSupported(
+ "ORC aggregate pushdown only supports struct nested leaf
projection, got kind {}",
+ static_cast<int>(type.getKind()));
+ }
+ const auto& child_projection = projection.children[0];
+ if (child_projection.local_id() < 0 ||
+ child_projection.local_id() >=
static_cast<int32_t>(type.getSubtypeCount())) {
+ return Status::InvalidArgument("Invalid ORC aggregate child local id
{} for kind {}",
+ child_projection.local_id(),
+ static_cast<int>(type.getKind()));
+ }
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(child_projection.local_id()));
+ DORIS_CHECK(child_type != nullptr);
+ return find_projected_minmax_leaf_in_type(*child_type, child_projection,
leaf_type);
+}
+
+Status find_projected_minmax_leaf(const ::orc::Type& root_type,
+ const format::LocalColumnIndex& projection,
+ const ::orc::Type** leaf_type) {
+ DORIS_CHECK(leaf_type != nullptr);
+ if (root_type.getKind() != ::orc::TypeKind::STRUCT) {
+ return Status::NotSupported("ORC aggregate pushdown requires top-level
struct schema");
+ }
+ const auto file_column_id = projection.column_id();
+ if (!file_column_id.is_valid() ||
+ file_column_id.value() >=
static_cast<int32_t>(root_type.getSubtypeCount())) {
+ return Status::InvalidArgument("Invalid ORC aggregate column id {}",
+ file_column_id.value());
+ }
+ const auto* column_type =
root_type.getSubtype(static_cast<uint64_t>(file_column_id.value()));
+ DORIS_CHECK(column_type != nullptr);
+ return find_projected_minmax_leaf_in_type(*column_type, projection,
leaf_type);
+}
+
+} // namespace
+
+class OrcReader::OrcFilterImpl final : public ::orc::ORCFilter {
+public:
+ explicit OrcFilterImpl(OrcReader* reader) : _reader(reader) {}
+
+ void filter(::orc::ColumnVectorBatch& data, uint16_t* sel, uint16_t size,
+ void* arg) const override {
+ THROW_IF_ERROR(_reader->_filter_orc_batch(data, sel, size, arg));
+ }
+
+private:
+ OrcReader* _reader = nullptr;
+};
+
+// Per-open mutable ORC state. close() publishes counters first, then resets
this
+// object so the reader can be opened again without carrying stale scan state.
+struct OrcReaderScanState {
+ struct StripeRange {
+ uint64_t first_stripe = 0;
+ uint64_t last_stripe = 0;
+ uint64_t offset = 0;
+ uint64_t length = 0;
+ };
+
+ std::unique_ptr<::orc::Reader> reader;
+ const ::orc::Type* root_type = nullptr;
+ ::orc::ReaderMetrics reader_metrics;
+ ::orc::RowReaderOptions row_reader_options; // projection + filter + SARG
+ stripe range
+ std::string timezone = TimezoneUtils::default_time_zone;
+ cctz::time_zone timezone_obj;
+ std::unique_ptr<::orc::RowReader> row_reader;
+ const ::orc::Type* selected_type = nullptr;
+ std::unique_ptr<::orc::ColumnVectorBatch> batch;
+
+ std::vector<format::LocalColumnId> read_columns;
+ std::map<format::LocalColumnId, size_t> column_to_selected_batch_index;
+
+ uint64_t current_batch_first_row = 0;
+ uint64_t condition_cache_next_row = 0;
+ std::shared_ptr<ConditionCacheContext> condition_cache_ctx;
+
+ std::vector<size_t> orc_lazy_selected_rows;
+ size_t orc_lazy_input_rows = 0;
+ uint64_t orc_lazy_next_batch_first_row = 0;
+ bool orc_lazy_read_enabled = false;
+ bool orc_lazy_selection_valid = false;
+
+ std::vector<StripeRange> selected_stripe_ranges;
+ size_t current_stripe_range = 0;
+ bool stripe_pruning_applied = false;
+
+ bool row_reader_created = false;
+};
+
+OrcReader::OrcReader(std::shared_ptr<io::FileSystemProperties>&
system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile*
profile,
+ std::optional<format::GlobalRowIdContext>
global_rowid_context)
+ : FileReader(system_properties, file_description, io_ctx, profile),
+ _global_rowid_context(std::move(global_rowid_context)) {}
+
+OrcReader::~OrcReader() = default;
+
+// Expose ORC pruning and lazy-read statistics in RuntimeProfile. These
counters
+// are the quickest way to confirm whether SARG/stripe pruning actually fired.
+void OrcReader::_init_profile() {
+ if (_profile == nullptr) {
+ return;
+ }
+
+ static const char* orc_profile = "OrcReader";
+ ADD_TIMER_WITH_LEVEL(_profile, orc_profile, 1);
+ _orc_profile.reader_call =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "ReaderCall", TUnit::UNIT,
orc_profile, 1);
+ _orc_profile.reader_inclusive_latency_us = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "ReaderInclusiveLatencyUs", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.decompression_call = ADD_CHILD_COUNTER_WITH_LEVEL(_profile,
"DecompressionCall",
+
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.decompression_latency_us = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "DecompressionLatencyUs", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.decoding_call =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "DecodingCall",
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.decoding_latency_us = ADD_CHILD_COUNTER_WITH_LEVEL(_profile,
"DecodingLatencyUs",
+
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.byte_decoding_call =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "ByteDecodingCall",
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.byte_decoding_latency_us = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "ByteDecodingLatencyUs", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.io_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "IOCount", TUnit::UNIT,
orc_profile, 1);
+ _orc_profile.io_blocking_latency_us = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "IOBlockingLatencyUs", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.selected_row_group_count = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "SelectedRowGroupCount", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.evaluated_row_group_count = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "EvaluatedRowGroupCount", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.read_row_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "ReadRowCount",
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.filtered_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(_profile,
"RowGroupsFiltered",
+
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.filtered_row_groups_by_min_max = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsFilteredByMinMax", TUnit::UNIT, orc_profile,
1);
+ _orc_profile.read_row_groups =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "RowGroupsReadNum",
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.filtered_group_rows = ADD_CHILD_COUNTER_WITH_LEVEL(_profile,
"FilteredRowsByGroup",
+
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByLazyRead", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.orc_lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByOrcLazyRead", TUnit::UNIT, orc_profile,
1);
+ _orc_profile.filtered_bytes =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FilteredBytes",
TUnit::BYTES, orc_profile, 1);
+ _orc_profile.open_file_num =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FileNum", TUnit::UNIT,
orc_profile, 1);
+}
+
+void OrcReader::_collect_profile() const {
+ if (_state == nullptr) {
+ return;
+ }
+
+ const auto& reader_metrics = _state->reader_metrics;
+ const uint64_t read_row_count = orc_read_row_count(reader_metrics);
+ if (_profile != nullptr) {
+ COUNTER_UPDATE(_orc_profile.reader_call,
orc_metric_value(reader_metrics.ReaderCall));
+ COUNTER_UPDATE(_orc_profile.reader_inclusive_latency_us,
+
orc_metric_value(reader_metrics.ReaderInclusiveLatencyUs));
+ COUNTER_UPDATE(_orc_profile.decompression_call,
+ orc_metric_value(reader_metrics.DecompressionCall));
+ COUNTER_UPDATE(_orc_profile.decompression_latency_us,
+
orc_metric_value(reader_metrics.DecompressionLatencyUs));
+ COUNTER_UPDATE(_orc_profile.decoding_call,
orc_metric_value(reader_metrics.DecodingCall));
+ COUNTER_UPDATE(_orc_profile.decoding_latency_us,
+ orc_metric_value(reader_metrics.DecodingLatencyUs));
+ COUNTER_UPDATE(_orc_profile.byte_decoding_call,
+ orc_metric_value(reader_metrics.ByteDecodingCall));
+ COUNTER_UPDATE(_orc_profile.byte_decoding_latency_us,
+ orc_metric_value(reader_metrics.ByteDecodingLatencyUs));
+ COUNTER_UPDATE(_orc_profile.io_count,
orc_metric_value(reader_metrics.IOCount));
+ COUNTER_UPDATE(_orc_profile.io_blocking_latency_us,
+ orc_metric_value(reader_metrics.IOBlockingLatencyUs));
+ COUNTER_UPDATE(_orc_profile.selected_row_group_count,
+ orc_metric_value(reader_metrics.SelectedRowGroupCount));
+ COUNTER_UPDATE(_orc_profile.evaluated_row_group_count,
+
orc_metric_value(reader_metrics.EvaluatedRowGroupCount));
+ COUNTER_UPDATE(_orc_profile.read_row_count, read_row_count);
+ COUNTER_UPDATE(_orc_profile.filtered_row_groups,
_reader_statistics.filtered_row_groups);
+ COUNTER_UPDATE(_orc_profile.filtered_row_groups_by_min_max,
+ _reader_statistics.filtered_row_groups_by_min_max);
+ COUNTER_UPDATE(_orc_profile.read_row_groups,
_reader_statistics.read_row_groups);
+ COUNTER_UPDATE(_orc_profile.filtered_group_rows,
_reader_statistics.filtered_group_rows);
+ COUNTER_UPDATE(_orc_profile.lazy_read_filtered_rows,
+ _reader_statistics.lazy_read_filtered_rows);
+ COUNTER_UPDATE(_orc_profile.orc_lazy_read_filtered_rows,
+ _reader_statistics.lazy_read_filtered_rows);
+ COUNTER_UPDATE(_orc_profile.filtered_bytes,
_reader_statistics.filtered_bytes);
+ COUNTER_UPDATE(_orc_profile.open_file_num,
_reader_statistics.open_file_num);
+ }
+ if (_io_ctx != nullptr && _io_ctx->file_reader_stats != nullptr) {
+ _io_ctx->file_reader_stats->read_rows += read_row_count;
+ }
+}
+
+format::ColumnDefinition OrcReader::row_position_column_definition() {
+ return format::row_position_column_definition();
+}
+
+Status OrcReader::init(RuntimeState* state) {
+ RETURN_IF_ERROR(format::FileReader::init(state));
+ _state = std::make_unique<OrcReaderScanState>();
+ TimezoneUtils::find_cctz_time_zone(_state->timezone, _state->timezone_obj);
+ if (state != nullptr) {
+ _state->timezone = state->timezone();
+ _state->timezone_obj = state->timezone_obj();
+ }
+
+ ::orc::ReaderOptions options;
+ options.setMemoryPool(*ExecEnv::GetInstance()->orc_memory_pool());
+ options.setReaderMetrics(&_state->reader_metrics);
+
+ auto input_stream =
std::make_unique<DorisOrcInputStream>(_file_description->path,
+
_tracing_file_reader, _io_ctx.get());
+ try {
+ _state->reader = ::orc::createReader(std::move(input_stream), options);
+ _state->root_type = &_state->reader->getType();
+ } catch (const std::exception& e) {
+ if (is_orc_stop(_io_ctx.get(), e)) {
+ return Status::EndOfFile("stop");
+ }
+ return Status::InternalError("Failed to open ORC file {}: {}",
_file_description->path,
+ e.what());
+ }
+ return Status::OK();
+}
+
+void
OrcReader::set_condition_cache_context(std::shared_ptr<ConditionCacheContext>
ctx) {
+ DORIS_CHECK(_state != nullptr);
+ _state->condition_cache_ctx = std::move(ctx);
+}
+
+int64_t OrcReader::get_total_rows() const {
+ DORIS_CHECK(_state != nullptr);
+ DORIS_CHECK(_state->reader != nullptr);
+ return cast_set<int64_t>(_state->reader->getNumberOfRows());
+}
+
+DataTypePtr OrcReader::_convert_to_doris_type(const ::orc::Type& type) const {
+ DataTypePtr data_type;
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BOOLEAN:
+ data_type = std::make_shared<DataTypeUInt8>();
+ break;
+ case ::orc::TypeKind::BYTE:
+ data_type = std::make_shared<DataTypeInt8>();
+ break;
+ case ::orc::TypeKind::SHORT:
+ data_type = std::make_shared<DataTypeInt16>();
+ break;
+ case ::orc::TypeKind::INT:
+ data_type = std::make_shared<DataTypeInt32>();
+ break;
+ case ::orc::TypeKind::LONG:
+ data_type = std::make_shared<DataTypeInt64>();
+ break;
+ case ::orc::TypeKind::FLOAT:
+ data_type = std::make_shared<DataTypeFloat32>();
+ break;
+ case ::orc::TypeKind::DOUBLE:
+ data_type = std::make_shared<DataTypeFloat64>();
+ break;
+ case ::orc::TypeKind::STRING:
+ case ::orc::TypeKind::BINARY:
+ data_type = std::make_shared<DataTypeString>();
+ break;
+ case ::orc::TypeKind::VARCHAR:
+ data_type =
std::make_shared<DataTypeString>(cast_set<int>(type.getMaximumLength()),
+
PrimitiveType::TYPE_VARCHAR);
+ break;
+ case ::orc::TypeKind::CHAR:
+ data_type =
std::make_shared<DataTypeString>(cast_set<int>(type.getMaximumLength()),
+ PrimitiveType::TYPE_CHAR);
+ break;
+ case ::orc::TypeKind::DATE:
+ data_type = std::make_shared<DataTypeDateV2>();
+ break;
+ case ::orc::TypeKind::TIMESTAMP:
+ case ::orc::TypeKind::TIMESTAMP_INSTANT:
+ data_type = std::make_shared<DataTypeDateTimeV2>(6);
+ break;
+ case ::orc::TypeKind::DECIMAL:
+ data_type = std::make_shared<DataTypeDecimal<TYPE_DECIMAL128I>>(
+ type.getPrecision() == 0 ? DECIMAL_PRECISION_FOR_HIVE11
+ : cast_set<int>(type.getPrecision()),
+ type.getPrecision() == 0 ? DECIMAL_SCALE_FOR_HIVE11
+ : cast_set<int>(type.getScale()));
+ break;
+ case ::orc::TypeKind::LIST:
+ data_type = _convert_list_to_doris_type(type);
+ break;
+ case ::orc::TypeKind::MAP:
+ data_type = _convert_map_to_doris_type(type);
+ break;
+ case ::orc::TypeKind::STRUCT:
+ data_type = _convert_struct_to_doris_type(type);
+ break;
+ default:
+ throw doris::Exception(
+ Status::NotSupported("ORC type {} is not supported by new ORC
reader",
+ static_cast<int>(type.getKind())));
+ }
+ return make_nullable(data_type);
+}
+
+DataTypePtr OrcReader::_convert_list_to_doris_type(const ::orc::Type& type)
const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::LIST);
+ DORIS_CHECK(type.getSubtypeCount() == 1);
+ const auto* element_type = type.getSubtype(0);
+ DORIS_CHECK(element_type != nullptr);
+ return
std::make_shared<DataTypeArray>(_convert_to_doris_type(*element_type));
+}
+
+DataTypePtr OrcReader::_convert_map_to_doris_type(const ::orc::Type& type)
const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ const auto* key_type = type.getSubtype(0);
+ const auto* value_type = type.getSubtype(1);
+ DORIS_CHECK(key_type != nullptr);
+ DORIS_CHECK(value_type != nullptr);
+ return std::make_shared<DataTypeMap>(_convert_to_doris_type(*key_type),
+ _convert_to_doris_type(*value_type));
+}
+
+DataTypePtr OrcReader::_convert_struct_to_doris_type(const ::orc::Type& type)
const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ DataTypes child_types;
+ Strings child_names;
+ child_types.reserve(type.getSubtypeCount());
+ child_names.reserve(type.getSubtypeCount());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ child_types.push_back(_convert_to_doris_type(*child_type));
+ child_names.push_back(type.getFieldName(child_idx));
+ }
+ return std::make_shared<DataTypeStruct>(child_types, child_names);
+}
+
+Status OrcReader::_fill_schema_field(const ::orc::Type& type, int32_t local_id,
+ const std::string& field_name,
+ format::ColumnDefinition* const field)
const {
+ if (field == nullptr) {
+ return Status::InvalidArgument("schema field is null");
+ }
+ field->local_id = local_id;
+ field->name = field_name;
+ field->column_type = format::ColumnType::DATA_COLUMN;
+ if (type.hasAttributeKey(ORC_ICEBERG_ID_ATTRIBUTE)) {
+ const auto iceberg_id =
type.getAttributeValue(ORC_ICEBERG_ID_ATTRIBUTE);
+ int32_t parsed_id = 0;
+ const auto* begin = iceberg_id.data();
+ const auto* end = begin + iceberg_id.size();
+ const auto [ptr, ec] = std::from_chars(begin, end, parsed_id);
+ if (ec != std::errc() || ptr != end) {
+ return Status::InvalidArgument("Invalid ORC Iceberg field id '{}'
for column {}",
+ iceberg_id, field_name);
+ }
+ field->identifier = Field::create_field<TYPE_INT>(parsed_id);
+ }
+ try {
+ field->type = _convert_to_doris_type(type);
+ } catch (const doris::Exception& e) {
+ return e.to_status();
+ }
+ field->children.clear();
+ switch (type.getKind()) {
+ case ::orc::TypeKind::STRUCT:
+ return _fill_struct_schema_children(type, field);
+ case ::orc::TypeKind::LIST:
+ return _fill_list_schema_children(type, field);
+ case ::orc::TypeKind::MAP:
+ return _fill_map_schema_children(type, field);
+ default:
+ break;
+ }
+ return Status::OK();
+}
+
+Status OrcReader::_fill_struct_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const
field) const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ field->children.reserve(type.getSubtypeCount());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ const auto child_name = type.getFieldName(child_idx);
+ format::ColumnDefinition child_field;
+ RETURN_IF_ERROR(_fill_schema_field(*child_type,
static_cast<int32_t>(child_idx), child_name,
+ &child_field));
+ field->children.push_back(std::move(child_field));
+ }
+ return Status::OK();
+}
+
+Status OrcReader::_fill_list_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const
field) const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::LIST);
+ DORIS_CHECK(type.getSubtypeCount() == 1);
+ const auto* element_type = type.getSubtype(0);
+ DORIS_CHECK(element_type != nullptr);
+
+ format::ColumnDefinition element_field;
+ RETURN_IF_ERROR(_fill_schema_field(*element_type, 0,
ORC_LIST_ELEMENT_NAME, &element_field));
+ field->children.push_back(std::move(element_field));
+ return Status::OK();
+}
+
+Status OrcReader::_fill_map_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const
field) const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ const auto* key_type = type.getSubtype(0);
+ const auto* value_type = type.getSubtype(1);
+ DORIS_CHECK(key_type != nullptr);
+ DORIS_CHECK(value_type != nullptr);
+
+ format::ColumnDefinition key_field;
+ RETURN_IF_ERROR(_fill_schema_field(*key_type, 0, ORC_MAP_KEY_NAME,
&key_field));
+ format::ColumnDefinition value_field;
+ RETURN_IF_ERROR(_fill_schema_field(*value_type, 1, ORC_MAP_VALUE_NAME,
&value_field));
+
+ field->children.push_back(std::move(key_field));
+ field->children.push_back(std::move(value_field));
+ return Status::OK();
+}
+
+Status OrcReader::get_schema(std::vector<format::ColumnDefinition>* const
file_schema) const {
+ if (file_schema == nullptr) {
+ return Status::InvalidArgument("file_schema is null");
+ }
+ if (_state == nullptr || _state->root_type == nullptr) {
+ return Status::Uninitialized("OrcReader is not open");
+ }
+ if (_state->root_type->getKind() != ::orc::TypeKind::STRUCT) {
+ return Status::NotSupported("ORC reader only supports top-level struct
schema");
+ }
+ file_schema->clear();
+ const auto extra_columns = _global_rowid_context.has_value() ? 1 : 0;
+ file_schema->reserve(_state->root_type->getSubtypeCount() + extra_columns);
+ for (uint64_t child_idx = 0; child_idx <
_state->root_type->getSubtypeCount(); ++child_idx) {
+ const auto* child_type = _state->root_type->getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ const auto child_name = _state->root_type->getFieldName(child_idx);
+ format::ColumnDefinition field;
+ RETURN_IF_ERROR(_fill_schema_field(*child_type,
static_cast<int32_t>(child_idx), child_name,
+ &field));
+ file_schema->push_back(std::move(field));
+ }
+ if (_global_rowid_context.has_value()) {
+ file_schema->push_back(nullable_global_rowid_column_definition());
+ }
+ return Status::OK();
+}
+
+std::unique_ptr<format::TableColumnMapper> OrcReader::create_column_mapper(
+ format::TableColumnMapperOptions options) const {
+ return std::make_unique<format::TableColumnMapper>(std::move(options));
+}
+
+Status OrcReader::open(std::shared_ptr<format::FileScanRequest> request) {
+ if (_state == nullptr || _state->reader == nullptr || _state->root_type ==
nullptr) {
+ return Status::Uninitialized("OrcReader is not open");
+ }
+ RETURN_IF_ERROR(format::FileReader::open(std::move(request)));
+
+ if (_request->local_positions.empty()) {
+ size_t next_position = 0;
+ for (const auto& projection : _request->predicate_columns) {
+ if (_request->local_positions
+ .emplace(projection.column_id(),
format::LocalIndex(next_position))
+ .second) {
+ ++next_position;
+ }
+ }
+ for (const auto& projection : _request->non_predicate_columns) {
+ if (_request->local_positions
+ .emplace(projection.column_id(),
format::LocalIndex(next_position))
+ .second) {
+ ++next_position;
+ }
+ }
+ }
+
+ _state->read_columns.clear();
+ _state->read_columns.reserve(_request->predicate_columns.size() +
+ _request->non_predicate_columns.size());
+ for (const auto& projection : _request->predicate_columns) {
+ _state->read_columns.push_back(projection.column_id());
+ }
+ for (const auto& projection : _request->non_predicate_columns) {
+ _state->read_columns.push_back(projection.column_id());
+ }
+ DCHECK(local_column_ids_are_unique(_request->predicate_columns));
+ DCHECK(local_column_ids_are_unique(_request->non_predicate_columns));
+ std::sort(_state->read_columns.begin(), _state->read_columns.end());
+ _state->read_columns.erase(
+ std::unique(_state->read_columns.begin(),
_state->read_columns.end()),
+ _state->read_columns.end());
+
+ _state->orc_lazy_read_enabled = _can_apply_orc_lazy_callback();
+
+ RETURN_IF_ERROR(_configure_row_reader_projection());
+ RETURN_IF_ERROR(set_orc_reader_timezone(_state->timezone,
&_state->row_reader_options));
+
_state->row_reader_options.setEnableLazyDecoding(_state->orc_lazy_read_enabled);
+ _state->row_reader_options.setUseTightNumericVector(false);
+
+ RETURN_IF_ERROR(_init_search_argument_from_local_filters());
+
+ RETURN_IF_ERROR(_select_stripe_ranges_by_statistics());
+ if (_state->stripe_pruning_applied &&
_state->selected_stripe_ranges.empty()) {
+ _eof = true;
+ return Status::OK();
+ }
+ _apply_current_stripe_range();
Review Comment:
The ORC v2 reader needs to apply the `TFileRangeDesc` byte range before
creating the row reader. `TableReader::prepare_split()` preserves
`start_offset`/`size` in `FileDescription`, and the existing ORC reader calls
`RowReaderOptions::range(_range_start_offset, _range_size)`, but this new path
only sets a range inside `_apply_current_stripe_range()` when SARG pruning
selected stripes. For ordinary ORC scans with no SARG pruning,
`_state->stripe_pruning_applied` is false, so the row reader is created without
any range and every split can read the whole file. On multi-split ORC scans
that duplicates rows across scanners. Please seed the row-reader range from
`_file_description->range_start_offset/range_size` and then intersect or
replace it with any SARG-selected stripe ranges.
##########
be/src/format_v2/orc/orc_reader.cpp:
##########
@@ -0,0 +1,2627 @@
+// 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 <limits>
+#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/segment/condition_cache.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_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_orc_stop(const io::IOContext* io_ctx, const std::exception& e) {
+ return io_ctx != nullptr && io_ctx->should_stop &&
std::string_view(e.what()) == "stop";
+}
+
+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) {
+ 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;
+ io::IOContext* _io_ctx = nullptr;
+};
+
+// selected_rows is a source-row remap produced by ORC lazy callback:
+// predicate columns are decoded first, then surviving row ids drive follower
decodes.
+bool is_null_at(const ::orc::ColumnVectorBatch& batch, size_t row) {
+ return batch.hasNulls && !batch.notNull[row];
+}
+
+size_t decode_row_count(size_t rows, const std::vector<size_t>* selected_rows)
{
+ if (selected_rows == nullptr) {
+ return rows;
+ }
+ return selected_rows->size();
+}
+
+size_t source_row_at(size_t row, const std::vector<size_t>* selected_rows) {
+ if (selected_rows == nullptr) {
+ return row;
+ }
+ return (*selected_rows)[row];
+}
+
+DecodedColumnView make_orc_decoded_view(size_t rows, const
std::vector<size_t>* selected_rows,
+ DecodedValueKind value_kind) {
+ DecodedColumnView view;
+ view.value_kind = value_kind;
+ view.row_count = cast_set<int64_t>(decode_row_count(rows, selected_rows));
+ return view;
+}
+
+void fill_orc_decoded_null_map(const ::orc::ColumnVectorBatch& batch, size_t
rows,
+ const std::vector<size_t>* selected_rows,
NullMap* null_map) {
+ DORIS_CHECK(null_map != nullptr);
+ if (!batch.hasNulls) {
+ return;
+ }
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ null_map->resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ (*null_map)[row] = !batch.notNull[source_row_at(row, selected_rows)];
+ }
+}
+
+size_t trim_right_spaces(const char* value, size_t length) {
+ while (length > 0 && value[length - 1] == ' ') {
+ --length;
+ }
+ return length;
+}
+
+Int128 to_int128(::orc::Int128 value) {
+ const auto high_bits =
static_cast<__uint128_t>(static_cast<uint64_t>(value.getHighBits()));
+ const auto low_bits = static_cast<__uint128_t>(value.getLowBits());
+ return static_cast<Int128>((high_bits << 64) | low_bits);
+}
+
+Status read_decoded_values(DataTypePtr data_type, IColumn& column,
DecodedColumnView* view) {
+ DORIS_CHECK(data_type != nullptr);
+ DORIS_CHECK(view != nullptr);
+
RETURN_IF_ERROR(data_type->get_serde()->read_column_from_decoded_values(column,
*view));
+ return Status::OK();
+}
+
+template <typename SourceType>
+void fill_selected_values(const SourceType* source_values, size_t rows,
+ const std::vector<size_t>* selected_rows,
+ std::vector<SourceType>* selected_values) {
+ DORIS_CHECK(source_values != nullptr);
+ DORIS_CHECK(selected_values != nullptr);
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ selected_values->resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ (*selected_values)[row] = source_values[source_row_at(row,
selected_rows)];
+ }
+}
+
+template <typename OrcBatchType, typename SourceType>
+Status decode_fixed_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>* selected_rows,
+ DecodedValueKind value_kind) {
+ const auto* orc_batch = dynamic_cast<const OrcBatchType*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC scalar batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows, value_kind);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ std::vector<SourceType> selected_values;
+ if (selected_rows == nullptr) {
+ view.values = reinterpret_cast<const uint8_t*>(orc_batch->data.data());
+ } else {
+ fill_selected_values(orc_batch->data.data(), rows, selected_rows,
&selected_values);
+ view.values = reinterpret_cast<const uint8_t*>(selected_values.data());
+ }
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_float_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::DoubleVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC float batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::FLOAT);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<float> float_values;
+ float_values.resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ float_values[row] =
static_cast<float>(orc_batch->data[source_row_at(row, selected_rows)]);
+ }
+ view.values = reinterpret_cast<const uint8_t*>(float_values.data());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_boolean_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::LongVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC boolean batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::BOOL);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::unique_ptr<bool[]> bool_values =
std::make_unique<bool[]>(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ bool_values[row] = orc_batch->data[source_row_at(row, selected_rows)]
!= 0;
+ }
+ view.values = reinterpret_cast<const uint8_t*>(bool_values.get());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_string_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::Type& file_type,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::StringVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC string batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::BINARY);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<StringRef> binary_values;
+ binary_values.reserve(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ const auto source_row = source_row_at(row, selected_rows);
+ if (is_null_at(batch, source_row)) {
+ binary_values.emplace_back("", 0);
+ continue;
+ }
+ auto length = static_cast<size_t>(orc_batch->length[source_row]);
+ if (file_type.getKind() == ::orc::TypeKind::CHAR) {
+ length = trim_right_spaces(orc_batch->data[source_row], length);
+ }
+ binary_values.emplace_back(length == 0 ? "" :
orc_batch->data[source_row], length);
+ }
+ view.binary_values = &binary_values;
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_date_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>* selected_rows)
{
+ const auto* orc_batch = dynamic_cast<const
::orc::LongVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC date batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::INT32);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<int32_t> date_values;
+ date_values.resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ date_values[row] =
cast_set<int32_t>(orc_batch->data[source_row_at(row, selected_rows)]);
+ }
+ view.values = reinterpret_cast<const uint8_t*>(date_values.data());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_decimal_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::Type& file_type,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::INT64);
+ view.decimal_precision = file_type.getPrecision() == 0
+ ? DECIMAL_PRECISION_FOR_HIVE11
+ : cast_set<int>(file_type.getPrecision());
+ view.decimal_scale = file_type.getPrecision() == 0 ?
DECIMAL_SCALE_FOR_HIVE11
+ :
cast_set<int>(file_type.getScale());
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ if (const auto* decimal64_batch = dynamic_cast<const
::orc::Decimal64VectorBatch*>(&batch);
+ decimal64_batch != nullptr) {
+ std::vector<int64_t> selected_values;
+ if (selected_rows == nullptr) {
+ view.values = reinterpret_cast<const
uint8_t*>(decimal64_batch->values.data());
+ } else {
+ fill_selected_values(decimal64_batch->values.data(), rows,
selected_rows,
+ &selected_values);
+ view.values = reinterpret_cast<const
uint8_t*>(selected_values.data());
+ }
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column,
&view));
+ return Status::OK();
+ }
+ const auto* decimal128_batch = dynamic_cast<const
::orc::Decimal128VectorBatch*>(&batch);
+ if (decimal128_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC decimal batch type {}",
batch.toString());
+ }
+ std::vector<StringRef> binary_values;
+ std::vector<std::array<uint8_t, sizeof(Int128)>> decimal_values;
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ decimal_values.resize(output_rows);
+ binary_values.reserve(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ auto value = __uint128_t();
+ const auto source_row = source_row_at(row, selected_rows);
+ if (!is_null_at(batch, source_row)) {
+ value =
static_cast<__uint128_t>(to_int128(decimal128_batch->values[source_row]));
+ }
+ for (size_t byte_idx = 0; byte_idx < decimal_values[row].size();
++byte_idx) {
+ const auto shift = (decimal_values[row].size() - byte_idx - 1) * 8;
+ decimal_values[row][byte_idx] = static_cast<uint8_t>(value >>
shift);
+ }
+ binary_values.emplace_back(reinterpret_cast<const
char*>(decimal_values[row].data()),
+ decimal_values[row].size());
+ }
+ view.value_kind = DecodedValueKind::FIXED_BINARY;
+ view.fixed_length = sizeof(Int128);
+ view.binary_values = &binary_values;
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+// ORC nested projection is type-id based. These helpers translate Doris'
+// LocalColumnIndex tree into the ORC type ids expected by includeTypes().
+Status get_projection_child_index(const format::LocalColumnIndex& child,
int32_t child_count,
+ const std::string& column_name, int32_t*
child_idx) {
+ DORIS_CHECK(child_idx != nullptr);
+ *child_idx = child.local_id();
+ if (*child_idx < 0 || *child_idx >= child_count) {
+ return Status::InvalidArgument("Invalid ORC projection child index {}
for column {}",
+ *child_idx, column_name);
+ }
+ return Status::OK();
+}
+
+void collect_type_and_descendant_ids(const ::orc::Type& type,
std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ collect_type_and_descendant_ids(*child_type, type_ids);
+ }
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC MAP projection for column {} contains
no children",
+ projection.local_id());
+ }
+
+ bool selected_key = false;
+ bool selected_value = false;
+ for (const auto& key_value_projection : projection.children) {
+ int32_t key_value_idx = 0;
+ RETURN_IF_ERROR(
+ get_projection_child_index(key_value_projection, 2, "orc_map",
&key_value_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(key_value_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
key_value_projection, type_ids));
+ selected_key = selected_key || key_value_idx == 0;
+ selected_value = selected_value || key_value_idx == 1;
+ }
+ if (!selected_key || !selected_value) {
+ return Status::NotSupported("ORC MAP projection must include both key
and value");
+ }
+ return Status::OK();
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC projection contains no children");
+ }
+ if (type.getKind() == ::orc::TypeKind::MAP) {
+ return collect_projected_map_type_ids(type, projection, type_ids);
+ }
+ if (type.getKind() != ::orc::TypeKind::STRUCT && type.getKind() !=
::orc::TypeKind::LIST) {
+ return Status::InvalidArgument("Cannot project children from
non-complex ORC type {}",
+ static_cast<int>(type.getKind()));
+ }
+
+ const auto child_count = static_cast<int32_t>(type.getSubtypeCount());
+ for (const auto& child_projection : projection.children) {
+ int32_t child_idx = 0;
+ RETURN_IF_ERROR(get_projection_child_index(child_projection,
child_count, "orc_complex",
+ &child_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(child_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
child_projection, type_ids));
+ }
+ return Status::OK();
+}
+
+// For arrays/maps, selected parent rows expand into selected element rows. The
+// returned offsets are compacted so downstream child decoding can append
densely.
+Status append_orc_offsets(ColumnArray::Offsets64& doris_offsets,
+ const ::orc::DataBuffer<int64_t>& orc_offsets,
size_t rows,
+ size_t* element_size, const std::vector<size_t>*
selected_rows = nullptr,
+ std::vector<size_t>* element_selection = nullptr) {
+ if (selected_rows != nullptr) {
+ DORIS_CHECK(element_selection != nullptr);
+ const auto prev_offset = doris_offsets.empty() ? 0 :
doris_offsets.back();
+ ColumnArray::Offset64 current_offset = prev_offset;
+ element_selection->clear();
+ for (size_t row = 0; row < selected_rows->size(); ++row) {
+ const auto source_row = (*selected_rows)[row];
+ DORIS_CHECK(source_row < rows);
+ const auto begin_offset = orc_offsets[source_row];
+ const auto end_offset = orc_offsets[source_row + 1];
+ if (end_offset < begin_offset) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ const auto delta = static_cast<size_t>(end_offset - begin_offset);
+ for (size_t element_idx = 0; element_idx < delta; ++element_idx) {
+ element_selection->push_back(static_cast<size_t>(begin_offset)
+ element_idx);
+ }
+ current_offset += static_cast<ColumnArray::Offset64>(delta);
+ doris_offsets.push_back(current_offset);
+ }
+ *element_size = element_selection->size();
+ return Status::OK();
+ }
+
+ const auto prev_offset = doris_offsets.empty() ? 0 : doris_offsets.back();
+ const auto base_offset = orc_offsets[0];
+ for (size_t idx = 1; idx <= rows; ++idx) {
+ const auto delta = orc_offsets[idx] - base_offset;
+ if (delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ doris_offsets.push_back(prev_offset +
static_cast<ColumnArray::Offset64>(delta));
+ }
+ const auto total_delta = orc_offsets[rows] - base_offset;
+ if (total_delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ *element_size = static_cast<size_t>(total_delta);
+ return Status::OK();
+}
+
+int64_t find_struct_child_index(const ::orc::Type& type, const std::string&
field_name) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ if (type.getFieldName(child_idx) == field_name) {
+ return static_cast<int64_t>(child_idx);
+ }
+ }
+ return -1;
+}
+
+bool is_row_position_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(format::ROW_POSITION_COLUMN_ID);
+}
+
+bool is_global_rowid_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(format::GLOBAL_ROWID_COLUMN_ID);
+}
+
+bool is_virtual_column(format::LocalColumnId file_column_id) {
+ return is_row_position_column(file_column_id) ||
is_global_rowid_column(file_column_id);
+}
+
+format::ColumnDefinition nullable_global_rowid_column_definition() {
+ auto field = format::global_rowid_column_definition();
+ field.type = make_nullable(field.type);
+ return field;
+}
+
+const format::LocalColumnIndex* find_projection(
+ const std::vector<format::LocalColumnIndex>& projections,
+ format::LocalColumnId file_column_id) {
+ const auto it = std::find_if(projections.begin(), projections.end(),
+ [&](const format::LocalColumnIndex&
projection) {
+ return projection.column_id() ==
file_column_id;
+ });
+ return it == projections.end() ? nullptr : &*it;
+}
+
+bool local_column_ids_are_unique(const std::vector<format::LocalColumnIndex>&
projections) {
+ std::set<format::LocalColumnId> column_ids;
+ for (const auto& projection : projections) {
+ if (!column_ids.insert(projection.column_id()).second) {
+ return false;
+ }
+ }
+ return true;
+}
+
+const format::LocalColumnIndex* find_request_projection(const
format::FileScanRequest& request,
+ format::LocalColumnId
file_column_id) {
+ if (const auto* projection = find_projection(request.predicate_columns,
file_column_id);
+ projection != nullptr) {
+ return projection;
+ }
+ return find_projection(request.non_predicate_columns, file_column_id);
+}
+
+bool has_pruned_projection(const format::LocalColumnIndex& projection) {
+ return !projection.project_all_children;
+}
+
+Status collect_lazy_filter_type_ids(const ::orc::Type& root_type,
+ const
std::vector<format::LocalColumnIndex>& projections,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ for (const auto& projection : projections) {
+ const auto file_column_id = projection.column_id();
+ if (is_virtual_column(file_column_id)) {
+ continue;
+ }
+ const auto* type =
root_type.getSubtype(static_cast<uint64_t>(file_column_id.value()));
+ DORIS_CHECK(type != nullptr);
+ if (!has_pruned_projection(projection)) {
+ collect_type_and_descendant_ids(*type, type_ids);
+ continue;
+ }
+ RETURN_IF_ERROR(collect_projected_type_ids(*type, projection,
type_ids));
+ }
+ return Status::OK();
+}
+
+// Stripe pruning maps ORC stripe statistics into Doris ZoneMap semantics.
Missing
+// or unsupported statistics are treated conservatively and never prune.
+bool set_integer_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* integer_statistics =
+ dynamic_cast<const ::orc::IntegerColumnStatistics*>(&statistics);
+ if (integer_statistics == nullptr || !integer_statistics->hasMinimum() ||
+ !integer_statistics->hasMaximum()) {
+ return false;
+ }
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BYTE:
+ zone_map->min_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::SHORT:
+ zone_map->min_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::INT:
+ zone_map->min_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::LONG:
+ zone_map->min_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMaximum());
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool set_boolean_zone_map(const ::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* boolean_statistics =
+ dynamic_cast<const ::orc::BooleanColumnStatistics*>(&statistics);
+ if (boolean_statistics == nullptr || !boolean_statistics->hasCount()) {
+ return false;
+ }
+ const bool has_false = boolean_statistics->getFalseCount() > 0;
+ const bool has_true = boolean_statistics->getTrueCount() > 0;
+ if (!has_false && !has_true) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_false ? 0 : 1));
+ zone_map->max_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_true ? 1 : 0));
+ return true;
+}
+
+bool set_floating_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* double_statistics = dynamic_cast<const
::orc::DoubleColumnStatistics*>(&statistics);
+ if (double_statistics == nullptr || !double_statistics->hasMinimum() ||
+ !double_statistics->hasMaximum()) {
+ return false;
+ }
+ if (type.getKind() == ::orc::TypeKind::FLOAT) {
+ zone_map->min_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMaximum()));
+ return true;
+ }
+ if (type.getKind() == ::orc::TypeKind::DOUBLE) {
+ zone_map->min_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMaximum());
+ return true;
+ }
+ return false;
+}
+
+bool set_string_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* string_statistics = dynamic_cast<const
::orc::StringColumnStatistics*>(&statistics);
+ if (string_statistics == nullptr || !string_statistics->hasMinimum() ||
+ !string_statistics->hasMaximum()) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_STRING>(string_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_STRING>(string_statistics->getMaximum());
+ return true;
+}
+
+bool set_date_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* date_statistics = dynamic_cast<const
::orc::DateColumnStatistics*>(&statistics);
+ if (date_statistics == nullptr || !date_statistics->hasMinimum() ||
+ !date_statistics->hasMaximum()) {
+ return false;
+ }
+ auto& date_dict = date_day_offset_dict::get();
+ zone_map->min_value =
+
Field::create_field<TYPE_DATEV2>(date_dict[date_statistics->getMinimum()]);
+ zone_map->max_value =
+
Field::create_field<TYPE_DATEV2>(date_dict[date_statistics->getMaximum()]);
+ return true;
+}
+
+DateV2Value<DateTimeV2ValueType> datetime_v2_from_orc_millis(int64_t millis,
int32_t nanos_tail) {
+ int64_t seconds = millis / 1000;
+ int64_t millis_remainder = millis % 1000;
+ if (millis_remainder < 0) {
+ --seconds;
+ millis_remainder += 1000;
+ }
+ const auto extra_nanos = std::max<int32_t>(nanos_tail, 0);
+ const auto microseconds = cast_set<uint64_t>(millis_remainder * 1000 +
extra_nanos / 1000);
+ DateV2Value<DateTimeV2ValueType> value;
+ value.from_unixtime(seconds, cctz::utc_time_zone());
Review Comment:
This makes TIMESTAMP MIN/MAX pushdown disagree with the normal scan path.
Normal `TypeKind::TIMESTAMP` decoding calls `_decode_timestamp_column(...,
_state->timezone_obj, ...)`, and that path materializes values with the session
timezone. The metadata aggregate path reaches this helper instead, where
`from_unixtime` is hardcoded to UTC. `TableReader` can enable MIN/MAX pushdown
for a direct timestamp column when there are no filters/deletes, so a non-UTC
session can get UTC min/max rows while the non-pushed scan would aggregate
timezone-adjusted values. Please either convert timestamp statistics with the
same timezone semantics as decoding, or return `NotSupported` for ORC
TIMESTAMP/TIMESTAMP_INSTANT MIN/MAX pushdown.
##########
be/src/format_v2/orc/orc_reader.cpp:
##########
@@ -0,0 +1,2627 @@
+// 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 <limits>
+#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/segment/condition_cache.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_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_orc_stop(const io::IOContext* io_ctx, const std::exception& e) {
+ return io_ctx != nullptr && io_ctx->should_stop &&
std::string_view(e.what()) == "stop";
+}
+
+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) {
+ 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;
+ io::IOContext* _io_ctx = nullptr;
+};
+
+// selected_rows is a source-row remap produced by ORC lazy callback:
+// predicate columns are decoded first, then surviving row ids drive follower
decodes.
+bool is_null_at(const ::orc::ColumnVectorBatch& batch, size_t row) {
+ return batch.hasNulls && !batch.notNull[row];
+}
+
+size_t decode_row_count(size_t rows, const std::vector<size_t>* selected_rows)
{
+ if (selected_rows == nullptr) {
+ return rows;
+ }
+ return selected_rows->size();
+}
+
+size_t source_row_at(size_t row, const std::vector<size_t>* selected_rows) {
+ if (selected_rows == nullptr) {
+ return row;
+ }
+ return (*selected_rows)[row];
+}
+
+DecodedColumnView make_orc_decoded_view(size_t rows, const
std::vector<size_t>* selected_rows,
+ DecodedValueKind value_kind) {
+ DecodedColumnView view;
+ view.value_kind = value_kind;
+ view.row_count = cast_set<int64_t>(decode_row_count(rows, selected_rows));
+ return view;
+}
+
+void fill_orc_decoded_null_map(const ::orc::ColumnVectorBatch& batch, size_t
rows,
+ const std::vector<size_t>* selected_rows,
NullMap* null_map) {
+ DORIS_CHECK(null_map != nullptr);
+ if (!batch.hasNulls) {
+ return;
+ }
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ null_map->resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ (*null_map)[row] = !batch.notNull[source_row_at(row, selected_rows)];
+ }
+}
+
+size_t trim_right_spaces(const char* value, size_t length) {
+ while (length > 0 && value[length - 1] == ' ') {
+ --length;
+ }
+ return length;
+}
+
+Int128 to_int128(::orc::Int128 value) {
+ const auto high_bits =
static_cast<__uint128_t>(static_cast<uint64_t>(value.getHighBits()));
+ const auto low_bits = static_cast<__uint128_t>(value.getLowBits());
+ return static_cast<Int128>((high_bits << 64) | low_bits);
+}
+
+Status read_decoded_values(DataTypePtr data_type, IColumn& column,
DecodedColumnView* view) {
+ DORIS_CHECK(data_type != nullptr);
+ DORIS_CHECK(view != nullptr);
+
RETURN_IF_ERROR(data_type->get_serde()->read_column_from_decoded_values(column,
*view));
+ return Status::OK();
+}
+
+template <typename SourceType>
+void fill_selected_values(const SourceType* source_values, size_t rows,
+ const std::vector<size_t>* selected_rows,
+ std::vector<SourceType>* selected_values) {
+ DORIS_CHECK(source_values != nullptr);
+ DORIS_CHECK(selected_values != nullptr);
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ selected_values->resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ (*selected_values)[row] = source_values[source_row_at(row,
selected_rows)];
+ }
+}
+
+template <typename OrcBatchType, typename SourceType>
+Status decode_fixed_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>* selected_rows,
+ DecodedValueKind value_kind) {
+ const auto* orc_batch = dynamic_cast<const OrcBatchType*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC scalar batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows, value_kind);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ std::vector<SourceType> selected_values;
+ if (selected_rows == nullptr) {
+ view.values = reinterpret_cast<const uint8_t*>(orc_batch->data.data());
+ } else {
+ fill_selected_values(orc_batch->data.data(), rows, selected_rows,
&selected_values);
+ view.values = reinterpret_cast<const uint8_t*>(selected_values.data());
+ }
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_float_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::DoubleVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC float batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::FLOAT);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<float> float_values;
+ float_values.resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ float_values[row] =
static_cast<float>(orc_batch->data[source_row_at(row, selected_rows)]);
+ }
+ view.values = reinterpret_cast<const uint8_t*>(float_values.data());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_boolean_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::LongVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC boolean batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::BOOL);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::unique_ptr<bool[]> bool_values =
std::make_unique<bool[]>(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ bool_values[row] = orc_batch->data[source_row_at(row, selected_rows)]
!= 0;
+ }
+ view.values = reinterpret_cast<const uint8_t*>(bool_values.get());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_string_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::Type& file_type,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::StringVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC string batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::BINARY);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<StringRef> binary_values;
+ binary_values.reserve(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ const auto source_row = source_row_at(row, selected_rows);
+ if (is_null_at(batch, source_row)) {
+ binary_values.emplace_back("", 0);
+ continue;
+ }
+ auto length = static_cast<size_t>(orc_batch->length[source_row]);
+ if (file_type.getKind() == ::orc::TypeKind::CHAR) {
+ length = trim_right_spaces(orc_batch->data[source_row], length);
+ }
+ binary_values.emplace_back(length == 0 ? "" :
orc_batch->data[source_row], length);
+ }
+ view.binary_values = &binary_values;
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_date_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>* selected_rows)
{
+ const auto* orc_batch = dynamic_cast<const
::orc::LongVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC date batch type {}",
batch.toString());
+ }
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::INT32);
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ std::vector<int32_t> date_values;
+ date_values.resize(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ date_values[row] =
cast_set<int32_t>(orc_batch->data[source_row_at(row, selected_rows)]);
+ }
+ view.values = reinterpret_cast<const uint8_t*>(date_values.data());
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+Status decode_decimal_values_with_serde(DataTypePtr data_type, IColumn& column,
+ const ::orc::Type& file_type,
+ const ::orc::ColumnVectorBatch& batch,
size_t rows,
+ const std::vector<size_t>*
selected_rows) {
+ auto view = make_orc_decoded_view(rows, selected_rows,
DecodedValueKind::INT64);
+ view.decimal_precision = file_type.getPrecision() == 0
+ ? DECIMAL_PRECISION_FOR_HIVE11
+ : cast_set<int>(file_type.getPrecision());
+ view.decimal_scale = file_type.getPrecision() == 0 ?
DECIMAL_SCALE_FOR_HIVE11
+ :
cast_set<int>(file_type.getScale());
+ NullMap null_map;
+ fill_orc_decoded_null_map(batch, rows, selected_rows, &null_map);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ if (const auto* decimal64_batch = dynamic_cast<const
::orc::Decimal64VectorBatch*>(&batch);
+ decimal64_batch != nullptr) {
+ std::vector<int64_t> selected_values;
+ if (selected_rows == nullptr) {
+ view.values = reinterpret_cast<const
uint8_t*>(decimal64_batch->values.data());
+ } else {
+ fill_selected_values(decimal64_batch->values.data(), rows,
selected_rows,
+ &selected_values);
+ view.values = reinterpret_cast<const
uint8_t*>(selected_values.data());
+ }
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column,
&view));
+ return Status::OK();
+ }
+ const auto* decimal128_batch = dynamic_cast<const
::orc::Decimal128VectorBatch*>(&batch);
+ if (decimal128_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC decimal batch type {}",
batch.toString());
+ }
+ std::vector<StringRef> binary_values;
+ std::vector<std::array<uint8_t, sizeof(Int128)>> decimal_values;
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ decimal_values.resize(output_rows);
+ binary_values.reserve(output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ auto value = __uint128_t();
+ const auto source_row = source_row_at(row, selected_rows);
+ if (!is_null_at(batch, source_row)) {
+ value =
static_cast<__uint128_t>(to_int128(decimal128_batch->values[source_row]));
+ }
+ for (size_t byte_idx = 0; byte_idx < decimal_values[row].size();
++byte_idx) {
+ const auto shift = (decimal_values[row].size() - byte_idx - 1) * 8;
+ decimal_values[row][byte_idx] = static_cast<uint8_t>(value >>
shift);
+ }
+ binary_values.emplace_back(reinterpret_cast<const
char*>(decimal_values[row].data()),
+ decimal_values[row].size());
+ }
+ view.value_kind = DecodedValueKind::FIXED_BINARY;
+ view.fixed_length = sizeof(Int128);
+ view.binary_values = &binary_values;
+ RETURN_IF_ERROR(read_decoded_values(std::move(data_type), column, &view));
+ return Status::OK();
+}
+
+// ORC nested projection is type-id based. These helpers translate Doris'
+// LocalColumnIndex tree into the ORC type ids expected by includeTypes().
+Status get_projection_child_index(const format::LocalColumnIndex& child,
int32_t child_count,
+ const std::string& column_name, int32_t*
child_idx) {
+ DORIS_CHECK(child_idx != nullptr);
+ *child_idx = child.local_id();
+ if (*child_idx < 0 || *child_idx >= child_count) {
+ return Status::InvalidArgument("Invalid ORC projection child index {}
for column {}",
+ *child_idx, column_name);
+ }
+ return Status::OK();
+}
+
+void collect_type_and_descendant_ids(const ::orc::Type& type,
std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ collect_type_and_descendant_ids(*child_type, type_ids);
+ }
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC MAP projection for column {} contains
no children",
+ projection.local_id());
+ }
+
+ bool selected_key = false;
+ bool selected_value = false;
+ for (const auto& key_value_projection : projection.children) {
+ int32_t key_value_idx = 0;
+ RETURN_IF_ERROR(
+ get_projection_child_index(key_value_projection, 2, "orc_map",
&key_value_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(key_value_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
key_value_projection, type_ids));
+ selected_key = selected_key || key_value_idx == 0;
+ selected_value = selected_value || key_value_idx == 1;
+ }
+ if (!selected_key || !selected_value) {
+ return Status::NotSupported("ORC MAP projection must include both key
and value");
+ }
+ return Status::OK();
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC projection contains no children");
+ }
+ if (type.getKind() == ::orc::TypeKind::MAP) {
+ return collect_projected_map_type_ids(type, projection, type_ids);
+ }
+ if (type.getKind() != ::orc::TypeKind::STRUCT && type.getKind() !=
::orc::TypeKind::LIST) {
+ return Status::InvalidArgument("Cannot project children from
non-complex ORC type {}",
+ static_cast<int>(type.getKind()));
+ }
+
+ const auto child_count = static_cast<int32_t>(type.getSubtypeCount());
+ for (const auto& child_projection : projection.children) {
+ int32_t child_idx = 0;
+ RETURN_IF_ERROR(get_projection_child_index(child_projection,
child_count, "orc_complex",
+ &child_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(child_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
child_projection, type_ids));
+ }
+ return Status::OK();
+}
+
+// For arrays/maps, selected parent rows expand into selected element rows. The
+// returned offsets are compacted so downstream child decoding can append
densely.
+Status append_orc_offsets(ColumnArray::Offsets64& doris_offsets,
+ const ::orc::DataBuffer<int64_t>& orc_offsets,
size_t rows,
+ size_t* element_size, const std::vector<size_t>*
selected_rows = nullptr,
+ std::vector<size_t>* element_selection = nullptr) {
+ if (selected_rows != nullptr) {
+ DORIS_CHECK(element_selection != nullptr);
+ const auto prev_offset = doris_offsets.empty() ? 0 :
doris_offsets.back();
+ ColumnArray::Offset64 current_offset = prev_offset;
+ element_selection->clear();
+ for (size_t row = 0; row < selected_rows->size(); ++row) {
+ const auto source_row = (*selected_rows)[row];
+ DORIS_CHECK(source_row < rows);
+ const auto begin_offset = orc_offsets[source_row];
+ const auto end_offset = orc_offsets[source_row + 1];
+ if (end_offset < begin_offset) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ const auto delta = static_cast<size_t>(end_offset - begin_offset);
+ for (size_t element_idx = 0; element_idx < delta; ++element_idx) {
+ element_selection->push_back(static_cast<size_t>(begin_offset)
+ element_idx);
+ }
+ current_offset += static_cast<ColumnArray::Offset64>(delta);
+ doris_offsets.push_back(current_offset);
+ }
+ *element_size = element_selection->size();
+ return Status::OK();
+ }
+
+ const auto prev_offset = doris_offsets.empty() ? 0 : doris_offsets.back();
+ const auto base_offset = orc_offsets[0];
+ for (size_t idx = 1; idx <= rows; ++idx) {
+ const auto delta = orc_offsets[idx] - base_offset;
+ if (delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ doris_offsets.push_back(prev_offset +
static_cast<ColumnArray::Offset64>(delta));
+ }
+ const auto total_delta = orc_offsets[rows] - base_offset;
+ if (total_delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ *element_size = static_cast<size_t>(total_delta);
+ return Status::OK();
+}
+
+int64_t find_struct_child_index(const ::orc::Type& type, const std::string&
field_name) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ if (type.getFieldName(child_idx) == field_name) {
+ return static_cast<int64_t>(child_idx);
+ }
+ }
+ return -1;
+}
+
+bool is_row_position_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(format::ROW_POSITION_COLUMN_ID);
+}
+
+bool is_global_rowid_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(format::GLOBAL_ROWID_COLUMN_ID);
+}
+
+bool is_virtual_column(format::LocalColumnId file_column_id) {
+ return is_row_position_column(file_column_id) ||
is_global_rowid_column(file_column_id);
+}
+
+format::ColumnDefinition nullable_global_rowid_column_definition() {
+ auto field = format::global_rowid_column_definition();
+ field.type = make_nullable(field.type);
+ return field;
+}
+
+const format::LocalColumnIndex* find_projection(
+ const std::vector<format::LocalColumnIndex>& projections,
+ format::LocalColumnId file_column_id) {
+ const auto it = std::find_if(projections.begin(), projections.end(),
+ [&](const format::LocalColumnIndex&
projection) {
+ return projection.column_id() ==
file_column_id;
+ });
+ return it == projections.end() ? nullptr : &*it;
+}
+
+bool local_column_ids_are_unique(const std::vector<format::LocalColumnIndex>&
projections) {
+ std::set<format::LocalColumnId> column_ids;
+ for (const auto& projection : projections) {
+ if (!column_ids.insert(projection.column_id()).second) {
+ return false;
+ }
+ }
+ return true;
+}
+
+const format::LocalColumnIndex* find_request_projection(const
format::FileScanRequest& request,
+ format::LocalColumnId
file_column_id) {
+ if (const auto* projection = find_projection(request.predicate_columns,
file_column_id);
+ projection != nullptr) {
+ return projection;
+ }
+ return find_projection(request.non_predicate_columns, file_column_id);
+}
+
+bool has_pruned_projection(const format::LocalColumnIndex& projection) {
+ return !projection.project_all_children;
+}
+
+Status collect_lazy_filter_type_ids(const ::orc::Type& root_type,
+ const
std::vector<format::LocalColumnIndex>& projections,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ for (const auto& projection : projections) {
+ const auto file_column_id = projection.column_id();
+ if (is_virtual_column(file_column_id)) {
+ continue;
+ }
+ const auto* type =
root_type.getSubtype(static_cast<uint64_t>(file_column_id.value()));
+ DORIS_CHECK(type != nullptr);
+ if (!has_pruned_projection(projection)) {
+ collect_type_and_descendant_ids(*type, type_ids);
+ continue;
+ }
+ RETURN_IF_ERROR(collect_projected_type_ids(*type, projection,
type_ids));
+ }
+ return Status::OK();
+}
+
+// Stripe pruning maps ORC stripe statistics into Doris ZoneMap semantics.
Missing
+// or unsupported statistics are treated conservatively and never prune.
+bool set_integer_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* integer_statistics =
+ dynamic_cast<const ::orc::IntegerColumnStatistics*>(&statistics);
+ if (integer_statistics == nullptr || !integer_statistics->hasMinimum() ||
+ !integer_statistics->hasMaximum()) {
+ return false;
+ }
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BYTE:
+ zone_map->min_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::SHORT:
+ zone_map->min_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::INT:
+ zone_map->min_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::LONG:
+ zone_map->min_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMaximum());
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool set_boolean_zone_map(const ::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* boolean_statistics =
+ dynamic_cast<const ::orc::BooleanColumnStatistics*>(&statistics);
+ if (boolean_statistics == nullptr || !boolean_statistics->hasCount()) {
+ return false;
+ }
+ const bool has_false = boolean_statistics->getFalseCount() > 0;
+ const bool has_true = boolean_statistics->getTrueCount() > 0;
+ if (!has_false && !has_true) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_false ? 0 : 1));
+ zone_map->max_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_true ? 1 : 0));
+ return true;
+}
+
+bool set_floating_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* double_statistics = dynamic_cast<const
::orc::DoubleColumnStatistics*>(&statistics);
+ if (double_statistics == nullptr || !double_statistics->hasMinimum() ||
+ !double_statistics->hasMaximum()) {
+ return false;
+ }
+ if (type.getKind() == ::orc::TypeKind::FLOAT) {
+ zone_map->min_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMaximum()));
+ return true;
+ }
+ if (type.getKind() == ::orc::TypeKind::DOUBLE) {
+ zone_map->min_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMaximum());
+ return true;
+ }
+ return false;
+}
+
+bool set_string_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* string_statistics = dynamic_cast<const
::orc::StringColumnStatistics*>(&statistics);
+ if (string_statistics == nullptr || !string_statistics->hasMinimum() ||
+ !string_statistics->hasMaximum()) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_STRING>(string_statistics->getMinimum());
Review Comment:
CHAR statistics need the same normalization as normal reads before they are
used for MIN/MAX pushdown. The decode path trims trailing spaces for
`TypeKind::CHAR`, but this metadata path stores
`StringColumnStatistics::getMinimum()/getMaximum()` directly. When
`TableReader` pushes down `MIN(char_col)` or `MAX(char_col)`, the upper
aggregate sees these raw statistic bytes instead of the trimmed Doris values a
regular scan would produce. Please trim CHAR statistic min/max with the same
`trim_right_spaces()` rule, or disable ORC CHAR MIN/MAX pushdown.
--
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]
