Gabriel39 commented on code in PR #69723:
URL: https://github.com/apache/doris/pull/69723#discussion_r4420464173
##########
be/src/exec/operator/file_scan_operator.cpp:
##########
@@ -247,41 +140,12 @@ Status
FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
scan_params = _split_source->get_params();
}
RETURN_IF_ERROR(validate_file_scan_support(*scan_params));
- const bool is_load =
-
state()->desc_tbl().get_tuple_descriptor(scan_params->src_tuple_id) != nullptr;
- // TODO: Use scanner v2 for all queries.
- const bool use_file_scanner_v2 =
- should_use_file_scanner_v2(state()->query_options(), is_load,
*scan_params);
- _operator_profile->add_info_string("UseScannerV2", use_file_scanner_v2 ?
"true" : "false");
- const auto* output_tuple_desc =
state()->desc_tbl().get_tuple_descriptor(_output_tuple_id);
- DORIS_CHECK(output_tuple_desc != nullptr);
- const bool metadata_only_count =
- is_count_star_pushdown() &&
_split_source->all_ranges_have_table_level_row_count();
- if (!is_load && !use_file_scanner_v2 && !metadata_only_count &&
- std::ranges::any_of(output_tuple_desc->slots(), [](const
SlotDescriptor* slot) {
- return contains_variant_type(slot->type());
- })) {
- // A syntactic COUNT(*) alone is insufficient: every assigned range
must prove that the
- // legacy scanner will emit metadata counts without decoding a Variant
carrier.
- return Status::NotSupported(
- "External VARIANT columns require FileScannerV2; the legacy
file scanner does "
- "not support VARIANT");
- }
- if (use_file_scanner_v2) {
- _parquet_profile_cache =
-
std::make_unique<format::parquet::ParquetProfileCache>(_scanner_profile.get());
- }
+ _parquet_profile_cache =
+
std::make_unique<format::parquet::ParquetProfileCache>(_scanner_profile.get());
for (int i = 0; i < _max_scanners; ++i) {
- ScannerSPtr scanner;
- if (use_file_scanner_v2) {
- scanner = FileScannerV2::create_shared(state(), this, p._limit,
_split_source,
- _scanner_profile.get(),
p._kv_cache.get(),
- &p._colname_to_slot_id);
- } else {
- scanner = FileScanner::create_shared(state(), this, p._limit,
_split_source,
- _scanner_profile.get(),
p._kv_cache.get(),
- &p._colname_to_slot_id);
- }
+ ScannerSPtr scanner = FileScannerV2::create_shared(
Review Comment:
Addressed in 843a4c6a372. V2 ORC now applies the scanner-provided batch cap
before physical decoding and preserves the RowReader position when that cap
changes. SessionBatchSizeLimitsPhysicalReads and
BatchSizeChangesPreserveReadPosition cover both cases. All 187 ORC tests passed
again in the latest ASAN run.
##########
be/src/format/orc/vorc_reader.cpp:
##########
@@ -1,3476 +0,0 @@
-// 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/orc/vorc_reader.h"
-
-#include <cctz/civil_time_detail.h>
-#include <gen_cpp/Metrics_types.h>
-#include <gen_cpp/Opcodes_types.h>
-#include <gen_cpp/PlanNodes_types.h>
-#include <gen_cpp/Types_types.h>
-#include <glog/logging.h>
-
-#include <algorithm>
-#include <cctype>
-#include <limits>
-#include <list>
-
-#include "exprs/runtime_filter_expr.h"
-#include "exprs/vdirect_in_predicate.h"
-#include "exprs/vexpr.h"
-#include "exprs/vslot_ref.h"
-#include "exprs/vtopn_pred.h"
-
-// IWYU pragma: no_include <bits/chrono.h>
-#include <chrono> // IWYU pragma: keep
-#include <exception>
-#include <iterator>
-#include <map>
-#include <memory>
-#include <ostream>
-#include <tuple>
-#include <utility>
-
-#include "absl/strings/substitute.h"
-#include "cctz/civil_time.h"
-#include "cctz/time_zone.h"
-#include "common/config.h"
-#include "common/consts.h"
-#include "common/exception.h"
-#include "core/block/block.h"
-#include "core/block/column_with_type_and_name.h"
-#include "core/column/column.h"
-#include "core/column/column_array.h"
-#include "core/column/column_const.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/data_type/data_type_array.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/define_primitive_type.h"
-#include "core/data_type/primitive_type.h"
-#include "core/string_ref.h"
-#include "core/types.h"
-#include "core/value/decimalv2_value.h"
-#include "core/value/vdatetime_value.h"
-#include "exec/scan/file_scanner.h"
-#include "exprs/create_predicate_function.h"
-#include "exprs/hybrid_set.h"
-#include "exprs/runtime_filter_expr.h"
-#include "exprs/vbloom_predicate.h"
-#include "exprs/vdirect_in_predicate.h"
-#include "exprs/vectorized_fn_call.h"
-#include "exprs/vexpr_context.h"
-#include "exprs/vexpr_fwd.h"
-#include "exprs/vin_predicate.h"
-#include "format/orc/orc_file_reader.h"
-#include "format/table/iceberg_default_value.h"
-#include "format/table/iceberg_reader.h"
-#include "io/fs/buffered_reader.h"
-#include "io/fs/file_reader.h"
-#include "orc/Exceptions.hh"
-#include "orc/Int128.hh"
-#include "orc/MemoryPool.hh"
-#include "orc/OrcFile.hh"
-#include "orc/sargs/Literal.hh"
-#include "orc/sargs/SearchArgument.hh"
-#include "runtime/descriptors.h"
-#include "runtime/runtime_profile.h"
-#include "runtime/runtime_state.h"
-#include "runtime/thread_context.h"
-#include "storage/id_manager.h"
-#include "storage/segment/column_reader.h"
-#include "storage/utils.h"
-#include "util/debug_points.h"
-#include "util/slice.h"
-#include "util/timezone_utils.h"
-#include "util/unaligned.h"
-
-namespace doris {
-class RuntimeState;
-namespace io {
-struct IOContext;
-enum class FileCachePolicy : uint8_t;
-} // namespace io
-} // namespace doris
-
-namespace doris {
-
-// TODO: we need to determine it by test.
-static constexpr uint32_t MAX_DICT_CODE_PREDICATE_TO_REWRITE =
std::numeric_limits<uint32_t>::max();
-static constexpr char
EMPTY_STRING_FOR_OVERFLOW[ColumnString::MAX_STRINGS_OVERFLOW_SIZE] = "";
-
-static std::string normalized_orc_timezone_name(const std::string& ctz) {
- return ctz.empty() ? "UTC" : (ctz == "CST" ? "Asia/Shanghai" : ctz);
-}
-
-static void fill_orc_null_map(ColumnNullable* nullable_column, const
orc::ColumnVectorBatch* cvb,
- size_t num_values) {
- NullMap& map_data_column = nullable_column->get_null_map_data();
- const auto origin_size = map_data_column.size();
- map_data_column.resize(origin_size + num_values);
- if (cvb->hasNulls) {
- const auto* cvb_nulls = cvb->notNull.data();
- for (int i = 0; i < num_values; ++i) {
- map_data_column[origin_size + i] = !cvb_nulls[i];
- }
- } else {
- memset(map_data_column.data() + origin_size, 0, num_values);
- }
-}
-
-static void align_orc_null_map(const ColumnPtr& src_column, ColumnNullable*
dst_nullable_column,
- size_t src_null_map_start, size_t new_rows) {
- auto& dst_null_map = dst_nullable_column->get_null_map_column();
- const size_t old_rows = dst_nullable_column->get_nested_column().size();
- const size_t expected_rows = old_rows + new_rows;
- if (dst_null_map.size() == expected_rows) {
- return;
- }
- DCHECK_EQ(dst_null_map.size(), old_rows);
- if (is_column_nullable(*src_column)) {
- const auto* src_nullable = assert_cast<const
ColumnNullable*>(src_column.get());
- DCHECK_GE(src_nullable->get_null_map_column().size(),
src_null_map_start + new_rows);
- dst_null_map.insert_range_from(src_nullable->get_null_map_column(),
src_null_map_start,
- new_rows);
- } else {
- dst_null_map.insert_many_vals(0, new_rows);
- }
-}
-// Because HIVE 0.11 & 0.12 does not support precision and scale for decimal
-// The decimal type of orc file produced by HIVE 0.11 & 0.12 are DECIMAL(0,0)
-// We should set a default precision and scale for these orc files.
-static constexpr int decimal_precision_for_hive11 =
BeConsts::MAX_DECIMAL128_PRECISION;
-static constexpr int decimal_scale_for_hive11 = 10;
-
-#define FOR_FLAT_ORC_COLUMNS(M) \
- M(PrimitiveType::TYPE_TINYINT, Int8, orc::LongVectorBatch) \
- M(PrimitiveType::TYPE_BOOLEAN, UInt8, orc::LongVectorBatch) \
- M(PrimitiveType::TYPE_SMALLINT, Int16, orc::LongVectorBatch) \
- M(PrimitiveType::TYPE_BIGINT, Int64, orc::LongVectorBatch) \
- M(PrimitiveType::TYPE_FLOAT, Float32, orc::DoubleVectorBatch) \
- M(PrimitiveType::TYPE_DOUBLE, Float64, orc::DoubleVectorBatch)
-
-void ORCFileInputStream::read(void* buf, uint64_t length, uint64_t offset) {
- uint64_t has_read = 0;
- char* out = reinterpret_cast<char*>(buf);
- while (has_read < length) {
- if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {
- throw orc::ParseError("stop");
- }
- size_t loop_read;
- Slice result(out + has_read, length - has_read);
- Status st = _tracing_file_reader->read_at(offset + has_read, result,
&loop_read, _io_ctx);
- if (!st.ok()) {
- throw orc::ParseError(
- absl::Substitute("Failed to read $0: $1", _file_name,
st.to_string()));
- }
- if (loop_read == 0) {
- break;
- }
- has_read += loop_read;
- }
- if (has_read != length) {
- throw orc::ParseError(absl::Substitute("Try to read $0 bytes from $1,
actually read $2",
- length, has_read, _file_name));
- }
-}
-
-void StripeStreamInputStream::read(void* buf, uint64_t length, uint64_t
offset) {
- uint64_t has_read = 0;
- char* out = reinterpret_cast<char*>(buf);
- while (has_read < length) {
- if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {
- throw orc::ParseError("stop");
- }
- size_t loop_read;
- Slice result(out + has_read, length - has_read);
- Status st = _inner_reader->read_at(offset + has_read, result,
&loop_read, _io_ctx);
- if (!st.ok()) {
- throw orc::ParseError(
- absl::Substitute("Failed to read $0: $1", _file_name,
st.to_string()));
- }
- if (loop_read == 0) {
- break;
- }
- has_read += loop_read;
- }
- if (has_read != length) {
- throw orc::ParseError(absl::Substitute("Try to read $0 bytes from $1,
actually read $2",
- length, has_read, _file_name));
- }
-}
-
-OrcReader::OrcReader(RuntimeProfile* profile, RuntimeState* state,
- const TFileScanRangeParams& params, const TFileRangeDesc&
range,
- size_t batch_size, const std::string& ctz, io::IOContext*
io_ctx,
- FileMetaCache* meta_cache, bool enable_lazy_mat)
- : _profile(profile),
- _state(state),
- _scan_params(params),
- _scan_range(range),
- _batch_size(std::max(batch_size, 1UL)),
- _range_start_offset(range.start_offset),
- _range_size(range.size),
- _ctz(ctz),
- _io_ctx(io_ctx),
- _enable_lazy_mat(enable_lazy_mat),
- _enable_filter_by_min_max(
- state == nullptr ? true :
state->query_options().enable_orc_filter_by_min_max),
- _dict_cols_has_converted(false) {
- TimezoneUtils::find_cctz_time_zone(normalized_orc_timezone_name(ctz),
_time_zone);
- _meta_cache = meta_cache;
- _init_profile();
- _init_system_properties();
- _init_file_description();
-}
-
-OrcReader::OrcReader(RuntimeProfile* profile, RuntimeState* state,
- const TFileScanRangeParams& params, const TFileRangeDesc&
range,
- size_t batch_size, const std::string& ctz,
- std::shared_ptr<io::IOContext> io_ctx_holder,
FileMetaCache* meta_cache,
- bool enable_lazy_mat)
- : _profile(profile),
- _state(state),
- _scan_params(params),
- _scan_range(range),
- _batch_size(std::max(batch_size, 1UL)),
- _range_start_offset(range.start_offset),
- _range_size(range.size),
- _ctz(ctz),
- _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),
- _io_ctx_holder(std::move(io_ctx_holder)),
- _enable_lazy_mat(enable_lazy_mat),
- _enable_filter_by_min_max(
- state == nullptr ? true :
state->query_options().enable_orc_filter_by_min_max),
- _dict_cols_has_converted(false) {
- TimezoneUtils::find_cctz_time_zone(normalized_orc_timezone_name(ctz),
_time_zone);
- _meta_cache = meta_cache;
- _init_profile();
- _init_system_properties();
- _init_file_description();
-}
-
-void OrcReader::set_batch_size(size_t batch_size) {
- DCHECK_GT(batch_size, 0);
- if (_batch_size == batch_size) {
- return;
- }
-
- _batch_size = batch_size;
- if (_row_reader != nullptr) {
- // ORC stores the batch capacity inside the row batch object returned
by createRowBatch().
- // Rebuild it when the requested batch size changes so the next call
uses the new limit.
- _batch = _row_reader->createRowBatch(_batch_size);
- }
-}
-
-OrcReader::OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc&
range,
- size_t batch_size, const std::string& ctz, io::IOContext*
io_ctx,
- FileMetaCache* meta_cache, bool enable_lazy_mat)
- : _profile(nullptr),
- _scan_params(params),
- _scan_range(range),
- _batch_size(std::max(batch_size, 1UL)),
- _ctz(ctz),
- _file_system(nullptr),
- _io_ctx(io_ctx),
- _enable_lazy_mat(enable_lazy_mat),
- _enable_filter_by_min_max(true),
- _dict_cols_has_converted(false) {
- TimezoneUtils::find_cctz_time_zone(normalized_orc_timezone_name(ctz),
_time_zone);
- _meta_cache = meta_cache;
- _init_system_properties();
- _init_file_description();
-}
-
-OrcReader::OrcReader(const TFileScanRangeParams& params, const TFileRangeDesc&
range,
- size_t batch_size, const std::string& ctz,
- std::shared_ptr<io::IOContext> io_ctx_holder,
FileMetaCache* meta_cache,
- bool enable_lazy_mat)
- : _profile(nullptr),
- _scan_params(params),
- _scan_range(range),
- _batch_size(std::max(batch_size, 1UL)),
- _ctz(ctz),
- _file_system(nullptr),
- _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),
- _io_ctx_holder(std::move(io_ctx_holder)),
- _enable_lazy_mat(enable_lazy_mat),
- _enable_filter_by_min_max(true),
- _dict_cols_has_converted(false) {
- TimezoneUtils::find_cctz_time_zone(normalized_orc_timezone_name(ctz),
_time_zone);
- _meta_cache = meta_cache;
- _init_system_properties();
- _init_file_description();
-}
-
-void OrcReader::_collect_profile_before_close() {
- if (_profile != nullptr) {
- COUNTER_UPDATE(_orc_profile.column_read_time,
_statistics.column_read_time);
- COUNTER_UPDATE(_orc_profile.get_batch_time,
_statistics.get_batch_time);
- COUNTER_UPDATE(_orc_profile.create_reader_time,
_statistics.create_reader_time);
- COUNTER_UPDATE(_orc_profile.init_column_time,
_statistics.init_column_time);
- COUNTER_UPDATE(_orc_profile.decode_value_time,
_statistics.decode_value_time);
- COUNTER_UPDATE(_orc_profile.decode_null_map_time,
_statistics.decode_null_map_time);
- COUNTER_UPDATE(_orc_profile.predicate_filter_time,
_statistics.predicate_filter_time);
- COUNTER_UPDATE(_orc_profile.dict_filter_rewrite_time,
_statistics.dict_filter_rewrite_time);
- COUNTER_UPDATE(_orc_profile.lazy_read_filtered_rows,
_statistics.lazy_read_filtered_rows);
- COUNTER_UPDATE(_orc_profile.file_footer_read_calls,
_statistics.file_footer_read_calls);
- COUNTER_UPDATE(_orc_profile.file_footer_hit_cache,
_statistics.file_footer_hit_cache);
- if (_file_input_stream != nullptr) {
- _file_input_stream->collect_profile_before_close();
- }
- }
-}
-
-int64_t OrcReader::size() const {
- return _file_input_stream->getLength();
-}
-
-void OrcReader::_init_profile() {
- if (_profile != nullptr) {
- static const char* orc_profile = "OrcReader";
- ADD_TIMER_WITH_LEVEL(_profile, orc_profile, 1);
- _orc_profile.column_read_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ColumnReadTime",
orc_profile, 1);
- _orc_profile.get_batch_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "GetBatchTime",
orc_profile, 1);
- _orc_profile.create_reader_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "CreateReaderTime",
orc_profile, 1);
- _orc_profile.init_column_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "InitColumnTime",
orc_profile, 1);
- _orc_profile.decode_value_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeValueTime",
orc_profile, 1);
- _orc_profile.decode_null_map_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeNullMapTime",
orc_profile, 1);
- _orc_profile.predicate_filter_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PredicateFilterTime",
orc_profile, 1);
- _orc_profile.dict_filter_rewrite_time =
- ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DictFilterRewriteTime",
orc_profile, 1);
- _orc_profile.lazy_read_filtered_rows =
- ADD_COUNTER_WITH_LEVEL(_profile, "FilteredRowsByLazyRead",
TUnit::UNIT, 1);
- _orc_profile.selected_row_group_count =
- ADD_COUNTER_WITH_LEVEL(_profile, "SelectedRowGroupCount",
TUnit::UNIT, 1);
- _orc_profile.evaluated_row_group_count =
- ADD_COUNTER_WITH_LEVEL(_profile, "EvaluatedRowGroupCount",
TUnit::UNIT, 1);
- _orc_profile.file_footer_read_calls =
- ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterReadCalls",
TUnit::UNIT, 1);
- _orc_profile.file_footer_hit_cache =
- ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterHitCache",
TUnit::UNIT, 1);
- }
-}
-
-Status OrcReader::_create_file_reader() {
- SCOPED_RAW_TIMER(&_statistics.create_reader_time);
- if (_reader != nullptr) {
- return Status::OK();
- }
-
- if (_file_input_stream == nullptr) {
- _file_description.mtime =
- _scan_range.__isset.modification_time ?
_scan_range.modification_time : 0;
- io::FileReaderOptions reader_options = FileFactory::get_reader_options(
- _state ? _state->query_options() : _default_query_options,
_file_description);
- io::FileReaderSPtr inner_reader;
- if (_io_ctx_holder != nullptr) {
- inner_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
- _profile, _system_properties, _file_description,
reader_options,
- io::DelegateReader::AccessMode::RANDOM, _io_ctx_holder));
- } else {
- inner_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
- _profile, _system_properties, _file_description,
reader_options,
- io::DelegateReader::AccessMode::RANDOM, _io_ctx));
- }
- _file_input_stream = std::make_unique<ORCFileInputStream>(
- _scan_range.path, std::move(inner_reader), _io_ctx, _profile,
- _orc_once_max_read_bytes, _orc_max_merge_distance_bytes);
- }
- if (_file_input_stream->getLength() == 0) {
- return Status::EndOfFile("empty orc file: " + _scan_range.path);
- }
-
- // create orc reader
- orc::ReaderOptions options;
- options.setMemoryPool(*ExecEnv::GetInstance()->orc_memory_pool());
- options.setReaderMetrics(&_reader_metrics);
-
- auto create_orc_reader = [&]() {
- try {
- _reader = orc::createReader(
-
std::unique_ptr<ORCFileInputStream>(_file_input_stream.release()), options);
- } catch (std::exception& e) {
- // invoker maybe just skip Status.NotFound and continue
- // so we need distinguish between it and other kinds of errors
- std::string _err_msg = e.what();
- if (_io_ctx && _io_ctx->should_stop && _err_msg == "stop") {
- return Status::EndOfFile("stop");
- }
- // one for fs, the other is for oss.
- if (_err_msg.find("No such file or directory") !=
std::string::npos ||
- _err_msg.find("NoSuchKey") != std::string::npos) {
- return Status::NotFound(_err_msg);
- }
- return Status::InternalError("Init OrcReader failed. reason = {}",
_err_msg);
- }
- return Status::OK();
- };
-
- if (_meta_cache == nullptr) {
- _statistics.file_footer_read_calls++;
- RETURN_IF_ERROR(create_orc_reader());
- } else {
- auto inner_file_reader = _file_input_stream->get_inner_reader();
- const auto& file_meta_cache_key =
- FileMetaCache::get_key(inner_file_reader, _file_description);
-
- // Local variables can be required because setSerializedFileTail is an
assignment operation, not a reference.
- ObjLRUCache::CacheHandle _meta_cache_handle;
- if (_meta_cache->lookup(file_meta_cache_key, &_meta_cache_handle)) {
- const std::string* footer_ptr = _meta_cache_handle.data<String>();
- options.setSerializedFileTail(*footer_ptr);
- RETURN_IF_ERROR(create_orc_reader());
- _statistics.file_footer_hit_cache++;
- } else {
- _statistics.file_footer_read_calls++;
- RETURN_IF_ERROR(create_orc_reader());
- std::string* footer_ptr = new std::string
{_reader->getSerializedFileTail()};
- _meta_cache->insert(file_meta_cache_key, footer_ptr,
&_meta_cache_handle);
- }
- }
-
- return Status::OK();
-}
-
-// ---- Unified init_reader(ReaderInitContext*) overrides ----
-
-Status OrcReader::_open_file_reader(ReaderInitContext* /*ctx*/) {
- if (_state != nullptr) {
- _orc_tiny_stripe_threshold_bytes =
_state->query_options().orc_tiny_stripe_threshold_bytes;
- _orc_once_max_read_bytes =
_state->query_options().orc_once_max_read_bytes;
- _orc_max_merge_distance_bytes =
_state->query_options().orc_max_merge_distance_bytes;
- }
- return _create_file_reader();
-}
-
-Status OrcReader::_do_init_reader(ReaderInitContext* base_ctx) {
- auto* ctx = checked_context_cast<OrcInitContext>(base_ctx);
- _table_column_names = base_ctx->column_names;
- _col_name_to_block_idx = base_ctx->col_name_to_block_idx;
- if (ctx->conjuncts != nullptr) {
- _lazy_read_ctx.conjuncts = *ctx->conjuncts;
- }
- _tuple_descriptor = ctx->tuple_descriptor;
- _row_descriptor = ctx->row_descriptor;
- _table_info_node_ptr = base_ctx->table_info_node;
- _column_ids = base_ctx->column_ids;
- _filter_column_ids = base_ctx->filter_column_ids;
-
- if (ctx->not_single_slot_filter_conjuncts != nullptr &&
- !ctx->not_single_slot_filter_conjuncts->empty()) {
-
_not_single_slot_filter_conjuncts.insert(_not_single_slot_filter_conjuncts.end(),
-
ctx->not_single_slot_filter_conjuncts->begin(),
-
ctx->not_single_slot_filter_conjuncts->end());
- for (const auto& conjunct : _not_single_slot_filter_conjuncts) {
- _block_dict_filter_for_slots(conjunct->root());
- }
- }
- _slot_id_to_filter_conjuncts = ctx->slot_id_to_filter_conjuncts;
- _obj_pool = std::make_unique<ObjectPool>();
-
- // _open_file_reader (called by init_reader NVI before hooks) must have
opened the file.
- DCHECK(_reader != nullptr) << "OrcReader::_do_init_reader called without
_open_file_reader";
- RETURN_IF_ERROR(_init_read_columns());
-
- // Compute missing columns and file↔table column mapping.
- // This runs in _do_init_reader (not on_before_init_reader) because
table-format readers
- // (Iceberg, Paimon, Hive, Hudi) override on_before_init_reader completely.
- _nested_initial_default_values.clear();
- if (has_column_descs()) {
- _fill_missing_cols.clear();
- _fill_missing_defaults.clear();
- for (const auto& col_name : _table_column_names) {
- if (!_table_info_node_ptr->children_column_exists(col_name)) {
- _fill_missing_cols.insert(col_name);
- }
- }
- if (_column_descs && !_fill_missing_cols.empty()) {
- for (const auto& desc : *_column_descs) {
- if (_fill_missing_cols.contains(desc.name) &&
- !_fill_partition_values.contains(desc.name)) {
- _fill_missing_defaults[desc.name] = desc.default_expr;
- }
- }
- }
- }
- // Resolve file-column ↔ table-column mapping.
- // _init_file_column_mapping handles both normal path (missing cols
populated above)
- // and standalone path (_fill_missing_cols empty, _table_info_node_ptr may
be null).
- _init_file_column_mapping();
-
- // ---- Inlined set_fill_columns logic (partition/missing/synthesized
classification) ----
-
- // 1. Collect predicate columns from conjuncts for lazy materialization
- std::unordered_map<std::string, std::pair<uint32_t, int>>
predicate_table_columns;
- _collect_predicate_columns_from_conjuncts(predicate_table_columns);
-
- // 2. Classify read/partition/missing/synthesized columns into lazy vs
predicate groups
- _classify_columns_for_lazy_read(predicate_table_columns,
_fill_partition_values,
- _fill_missing_defaults);
-
- // 3. Init search argument for min-max filtering
- if (_lazy_read_ctx.conjuncts.empty()) {
- _lazy_read_ctx.can_lazy_read = false;
- } else if (_enable_filter_by_min_max) {
- auto res = _init_search_argument(_push_down_exprs);
- if (_state->query_options().check_orc_init_sargs_success && !res) {
- std::stringstream ss;
- for (const auto& conjunct : _lazy_read_ctx.conjuncts) {
- ss << conjunct->root()->debug_string() << "\n";
- }
- return Status::InternalError(
- "Session variable check_orc_init_sargs_success is set, but
"
- "_init_search_argument returns false because all exprs can
not be pushed "
- "down:\n " +
- ss.str());
- }
- }
-
- // 4. Create ORC row reader (includes tiny stripe optimization and type
map)
- RETURN_IF_ERROR(_init_orc_row_reader());
-
- // 5. Build filter conjuncts from not_single_slot and
predicate_partition_columns
- if (!_not_single_slot_filter_conjuncts.empty()) {
- _filter_conjuncts.insert(_filter_conjuncts.end(),
_not_single_slot_filter_conjuncts.begin(),
- _not_single_slot_filter_conjuncts.end());
- }
- if (_slot_id_to_filter_conjuncts &&
!_slot_id_to_filter_conjuncts->empty()) {
- for (auto& kv : _lazy_read_ctx.predicate_partition_columns) {
- auto& [value, slot_desc] = kv.second;
- auto iter = _slot_id_to_filter_conjuncts->find(slot_desc->id());
- if (iter != _slot_id_to_filter_conjuncts->end()) {
- for (const auto& conjunct_ctx : iter->second) {
- _filter_conjuncts.push_back(conjunct_ctx);
- }
- }
- }
- }
-
- return Status::OK();
-}
-
-Status OrcReader::on_before_init_reader(ReaderInitContext* ctx) {
- _column_descs = ctx->column_descs;
- _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
- RETURN_IF_ERROR(_extract_partition_values(*ctx->range,
ctx->tuple_descriptor,
- _fill_partition_values,
- &_fill_partition_value_is_null));
- for (const auto& desc : *ctx->column_descs) {
- if (desc.category == ColumnCategory::REGULAR ||
- desc.category == ColumnCategory::GENERATED) {
- ctx->column_names.push_back(desc.name);
- } else if (desc.category == ColumnCategory::SYNTHESIZED &&
- desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
- auto topn_row_id_column_iter =
_create_topn_row_id_column_iterator();
- this->register_synthesized_column_handler(
- desc.name,
- [iter = std::move(topn_row_id_column_iter), this, &desc](
- Block* block, size_t rows) -> Status {
- return fill_topn_row_id(iter, desc.name, block, rows);
- });
- continue;
- }
- }
-
- // Build table_info_node from ORC file type with case-insensitive
recursive matching.
- // _reader is available here because init_reader calls
_create_file_reader() before this hook.
- // tuple_descriptor may be null in unit tests that only set column_descs.
- if (ctx->tuple_descriptor != nullptr) {
-
RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_orc_name(
- ctx->tuple_descriptor, &_reader->getType(),
ctx->table_info_node));
- }
-
- return Status::OK();
-}
-
-// init file reader for parsing schema
-Status OrcReader::init_schema_reader() {
- return _create_file_reader();
-}
-
-Status OrcReader::get_parsed_schema(std::vector<std::string>* col_names,
- std::vector<DataTypePtr>* col_types) {
- const auto& root_type = _reader->getType();
- for (int i = 0; i < root_type.getSubtypeCount(); ++i) {
- col_names->emplace_back(root_type.getFieldName(i));
-
col_types->emplace_back(convert_to_doris_type(root_type.getSubtype(i)));
- }
- return Status::OK();
-}
-
-Status OrcReader::_init_read_columns() {
- SCOPED_RAW_TIMER(&_statistics.init_column_time);
- const auto& root_type = _reader->getType();
-
- // Build column ID to file type mapping for all columns in file
- _column_id_to_file_type.clear();
- std::function<void(const orc::Type*, int)> build_column_id_map = [&](const
orc::Type* type,
- int
depth) {
- if (type == nullptr) return;
- uint64_t col_id = type->getColumnId();
- _column_id_to_file_type[col_id] = type;
-
- std::string indent(depth * 2, ' ');
- VLOG_DEBUG << indent << "[OrcReader] Mapping column_id=" << col_id
- << ", kind=" << static_cast<int>(type->getKind())
- << ", subtype_count=" << type->getSubtypeCount();
-
- for (uint64_t i = 0; i < type->getSubtypeCount(); ++i) {
- build_column_id_map(type->getSubtype(i), depth + 1);
- }
- };
-
- VLOG_DEBUG << "[OrcReader] Building column ID to file type mapping...";
- build_column_id_map(&root_type, 0);
- VLOG_DEBUG << "[OrcReader] Total mapped columns: " <<
_column_id_to_file_type.size();
-
- for (int i = 0; i < root_type.getSubtypeCount(); ++i) {
- _type_map.emplace(root_type.getFieldName(i), root_type.getSubtype(i));
- }
-
- return Status::OK();
-}
-
-void OrcReader::_init_file_column_mapping() {
- for (const auto& col_name : _table_column_names) {
- if (_fill_missing_cols.contains(col_name)) {
- continue;
- }
- std::string file_col = col_name;
- if (_table_info_node_ptr &&
_table_info_node_ptr->children_column_exists(col_name)) {
- file_col =
_table_info_node_ptr->children_file_column_name(col_name);
- }
- _read_file_cols.emplace_back(file_col);
- _read_table_cols.emplace_back(col_name);
- }
-}
-
-// orc only support LONG, FLOAT, STRING, DATE, DECIMAL, TIMESTAMP, BOOLEAN to
push down predicates
-static std::unordered_map<orc::TypeKind, orc::PredicateDataType>
TYPEKIND_TO_PREDICATE_TYPE = {
- {orc::TypeKind::BYTE, orc::PredicateDataType::LONG},
- {orc::TypeKind::SHORT, orc::PredicateDataType::LONG},
- {orc::TypeKind::INT, orc::PredicateDataType::LONG},
- {orc::TypeKind::LONG, orc::PredicateDataType::LONG},
- {orc::TypeKind::FLOAT, orc::PredicateDataType::FLOAT},
- {orc::TypeKind::DOUBLE, orc::PredicateDataType::FLOAT},
- {orc::TypeKind::STRING, orc::PredicateDataType::STRING},
- {orc::TypeKind::BINARY, orc::PredicateDataType::STRING},
- // should not pust down CHAR type, because CHAR type is fixed length
and will be padded
- // {orc::TypeKind::CHAR, orc::PredicateDataType::STRING},
- {orc::TypeKind::VARCHAR, orc::PredicateDataType::STRING},
- {orc::TypeKind::DATE, orc::PredicateDataType::DATE},
- {orc::TypeKind::DECIMAL, orc::PredicateDataType::DECIMAL},
- {orc::TypeKind::TIMESTAMP, orc::PredicateDataType::TIMESTAMP},
- {orc::TypeKind::BOOLEAN, orc::PredicateDataType::BOOLEAN}};
-
-template <PrimitiveType primitive_type>
-std::tuple<bool, orc::Literal> convert_to_orc_literal(const orc::Type* type,
- StringRef& literal_data,
int precision,
- int scale) {
- const auto* value = literal_data.data;
- try {
- switch (type->getKind()) {
- case orc::TypeKind::BOOLEAN: {
- if (primitive_type != TYPE_BOOLEAN) {
- return std::make_tuple(false, orc::Literal(false));
- }
- return std::make_tuple(true,
orc::Literal(bool(*((uint8_t*)value))));
- }
- case orc::TypeKind::BYTE:
- case orc::TypeKind::SHORT:
- case orc::TypeKind::INT:
- case orc::TypeKind::LONG: {
- if constexpr (primitive_type == TYPE_TINYINT) {
- return std::make_tuple(true,
orc::Literal(int64_t(*((int8_t*)value))));
- } else if constexpr (primitive_type == TYPE_SMALLINT) {
- return std::make_tuple(true,
orc::Literal(int64_t(*((int16_t*)value))));
- } else if constexpr (primitive_type == TYPE_INT) {
- return std::make_tuple(true,
orc::Literal(int64_t(*((int32_t*)value))));
- } else if constexpr (primitive_type == TYPE_BIGINT) {
- return std::make_tuple(true, orc::Literal(*((int64_t*)value)));
- }
- return std::make_tuple(false, orc::Literal(false));
- }
- case orc::TypeKind::FLOAT: {
- if constexpr (primitive_type == TYPE_FLOAT) {
- return std::make_tuple(true,
orc::Literal(double(*((float*)value))));
- } else if constexpr (primitive_type == TYPE_DOUBLE) {
- return std::make_tuple(true, orc::Literal(*((double*)value)));
- }
- return std::make_tuple(false, orc::Literal(false));
- }
- case orc::TypeKind::DOUBLE: {
- if (primitive_type == TYPE_DOUBLE) {
- return std::make_tuple(true, orc::Literal(*((double*)value)));
- }
- return std::make_tuple(false, orc::Literal(false));
- }
- case orc::TypeKind::STRING:
- [[fallthrough]];
- case orc::TypeKind::BINARY:
- [[fallthrough]];
- // should not pust down CHAR type, because CHAR type is fixed length
and will be padded
- // case orc::TypeKind::CHAR:
- // [[fallthrough]];
- case orc::TypeKind::VARCHAR: {
- if (primitive_type == TYPE_STRING || primitive_type == TYPE_CHAR ||
- primitive_type == TYPE_VARCHAR) {
- return std::make_tuple(true, orc::Literal(literal_data.data,
literal_data.size));
- }
- return std::make_tuple(false, orc::Literal(false));
- }
- case orc::TypeKind::DECIMAL: {
- int128_t decimal_value;
- if constexpr (primitive_type == TYPE_DECIMALV2) {
- // value may not be 16-byte aligned; use unaligned_load to
avoid UB.
- decimal_value = unaligned_load<int128_t>(value);
- precision = DecimalV2Value::PRECISION;
- scale = DecimalV2Value::SCALE;
- } else if constexpr (primitive_type == TYPE_DECIMAL32) {
- decimal_value = *((int32_t*)value);
- } else if constexpr (primitive_type == TYPE_DECIMAL64) {
- decimal_value = *((int64_t*)value);
- } else if constexpr (primitive_type == TYPE_DECIMAL128I) {
- decimal_value = unaligned_load<int128_t>(value);
- } else {
- return std::make_tuple(false, orc::Literal(false));
- }
- return std::make_tuple(true,
orc::Literal(orc::Int128(uint64_t(decimal_value >> 64),
-
uint64_t(decimal_value)),
- precision, scale));
- }
- case orc::TypeKind::DATE: {
- int64_t day_offset;
- static const cctz::time_zone utc0 = cctz::utc_time_zone();
- if constexpr (primitive_type == TYPE_DATE) {
- const VecDateTimeValue date_v1 = *reinterpret_cast<const
VecDateTimeValue*>(value);
- cctz::civil_day civil_date(date_v1.year(), date_v1.month(),
date_v1.day());
- day_offset =
- cctz::convert(civil_date,
utc0).time_since_epoch().count() / (24 * 60 * 60);
- } else if (primitive_type == TYPE_DATEV2) {
- const DateV2Value<DateV2ValueType> date_v2 =
- *reinterpret_cast<const
DateV2Value<DateV2ValueType>*>(value);
- cctz::civil_day civil_date(date_v2.year(), date_v2.month(),
date_v2.day());
- day_offset =
- cctz::convert(civil_date,
utc0).time_since_epoch().count() / (24 * 60 * 60);
- } else {
- return std::make_tuple(false, orc::Literal(false));
- }
- return std::make_tuple(true,
orc::Literal(orc::PredicateDataType::DATE, day_offset));
- }
- case orc::TypeKind::TIMESTAMP: {
- int64_t seconds;
- int32_t nanos;
- static const cctz::time_zone utc0 = cctz::utc_time_zone();
- // TODO: ColumnValueRange has lost the precision of microsecond
- if constexpr (primitive_type == TYPE_DATETIME) {
- const VecDateTimeValue datetime_v1 =
- *reinterpret_cast<const VecDateTimeValue*>(value);
- cctz::civil_second civil_seconds(datetime_v1.year(),
datetime_v1.month(),
- datetime_v1.day(),
datetime_v1.hour(),
- datetime_v1.minute(),
datetime_v1.second());
- seconds = cctz::convert(civil_seconds,
utc0).time_since_epoch().count();
- nanos = 0;
- } else if (primitive_type == TYPE_DATETIMEV2) {
- const DateV2Value<DateTimeV2ValueType> datetime_v2 =
- *reinterpret_cast<const
DateV2Value<DateTimeV2ValueType>*>(value);
- cctz::civil_second civil_seconds(datetime_v2.year(),
datetime_v2.month(),
- datetime_v2.day(),
datetime_v2.hour(),
- datetime_v2.minute(),
datetime_v2.second());
- seconds = cctz::convert(civil_seconds,
utc0).time_since_epoch().count();
- nanos = datetime_v2.microsecond() * 1000;
- } else {
- return std::make_tuple(false, orc::Literal(false));
- }
- return std::make_tuple(true, orc::Literal(seconds, nanos));
- }
- default:
- return std::make_tuple(false, orc::Literal(false));
- }
- } catch (Exception& e) {
- // When table schema changed, and using new schema to read old data.
- LOG(WARNING) << "Failed to convert doris value to orc predicate
literal, error = "
- << e.what();
- return std::make_tuple(false, orc::Literal(false));
- }
-}
-
-std::pair<bool, orc::PredicateDataType> OrcReader::_get_orc_predicate_type(
- const VSlotRef* slot_ref) {
-
DCHECK(_table_info_node_ptr->children_column_exists(slot_ref->expr_name()));
- auto file_col_name =
_table_info_node_ptr->children_file_column_name(slot_ref->expr_name());
- if (!_type_map.contains(file_col_name)) {
- LOG(WARNING) << "Column " << slot_ref->expr_name() << "in file name"
<< file_col_name
- << " not found in _type_map";
- return {false, orc::PredicateDataType::LONG};
- }
- DCHECK(_type_map.contains(file_col_name));
- const auto* orc_type = _type_map[file_col_name];
- if (!TYPEKIND_TO_PREDICATE_TYPE.contains(orc_type->getKind())) {
- LOG(WARNING) << "Unsupported Push Down Orc Type [TypeKind=" <<
orc_type->getKind() << "]";
- return {false, orc::PredicateDataType::LONG};
- }
- const auto predicate_type =
TYPEKIND_TO_PREDICATE_TYPE[orc_type->getKind()];
- return {true, predicate_type};
-}
-
-std::pair<bool, orc::Literal> OrcReader::_make_orc_literal(const VSlotRef*
slot_ref,
- const VLiteral*
literal) {
- // Get predicate type using new function
- auto [valid_pred_type, predicate_type] = _get_orc_predicate_type(slot_ref);
- if (!valid_pred_type) {
- return {false, orc::Literal(false)};
- }
-
- // Get the orc_type again here as it's needed for convert_to_orc_literal
- auto file_col_name =
_table_info_node_ptr->children_file_column_name(slot_ref->expr_name());
- const auto* orc_type = _type_map[file_col_name];
-
- DCHECK(literal != nullptr);
- // this only happens when the literals of in_predicate contains null
value, like in (1, null)
- if (literal->get_column_ptr()->is_null_at(0)) {
- return {false, orc::Literal(false)};
- }
- auto literal_data = literal->get_column_ptr()->get_data_at(0);
- auto* slot = _tuple_descriptor->slots()[slot_ref->column_id()];
- auto slot_type = slot->type();
- auto primitive_type = slot_type->get_primitive_type();
- auto src_type = convert_to_doris_type(orc_type)->get_primitive_type();
- // should not down predicate for string type change from other type
- if (src_type != primitive_type && !is_string_type(src_type) &&
is_string_type(primitive_type)) {
- LOG(WARNING) << "Unsupported Push Down Schema Changed Column " <<
primitive_type << " to "
- << src_type;
- return {false, orc::Literal(false)};
- }
- switch (primitive_type) {
-#define M(NAME)
\
- case TYPE_##NAME: {
\
- auto [valid, orc_literal] = convert_to_orc_literal<TYPE_##NAME>(
\
- orc_type, literal_data, slot_type->get_precision(),
slot_type->get_scale()); \
- return {valid, orc_literal};
\
- }
-#define APPLY_FOR_PRIMITIVE_TYPE(M) \
- M(TINYINT) \
- M(SMALLINT) \
- M(INT) \
- M(BIGINT) \
- M(LARGEINT) \
- M(DATE) \
- M(DATETIME) \
- M(DATEV2) \
- M(DATETIMEV2) \
- M(VARCHAR) \
- M(STRING) \
- M(HLL) \
- M(DECIMAL32) \
- M(DECIMAL64) \
- M(DECIMAL128I) \
- M(DECIMAL256) \
- M(DECIMALV2) \
- M(BOOLEAN) \
- M(IPV4) \
- M(IPV6) \
- M(UUID)
- APPLY_FOR_PRIMITIVE_TYPE(M)
-#undef M
- default: {
- VLOG_CRITICAL << "Unsupported Convert Orc Literal [ColName=" <<
slot->col_name() << "]";
- return {false, orc::Literal(false)};
- }
- }
-}
-
-// check if the slot of expr can be pushed down to orc reader and make orc
predicate type
-bool OrcReader::_check_slot_can_push_down(const VExprSPtr& expr) {
- if (!expr->children()[0]->is_slot_ref()) {
- return false;
- }
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- // check if the slot exists in orc file and not partition column
- if
(_lazy_read_ctx.predicate_partition_columns.contains(slot_ref->expr_name()) ||
-
(!_table_info_node_ptr->children_column_exists(slot_ref->expr_name()))) {
- return false;
- }
-
- if (!has_column_optimization(slot_ref->expr_name(),
ColumnOptimizationTypes::MIN_MAX)) {
- return false;
- }
-
- // Directly use _get_orc_predicate_type since we only need the type
- auto [valid, predicate_type] = _get_orc_predicate_type(slot_ref);
- if (valid) {
- _vslot_ref_to_orc_predicate_data_type[slot_ref] = predicate_type;
- }
- return valid;
-}
-
-// check if the literal of expr can be pushed down to orc reader and make orc
literal
-bool OrcReader::_check_literal_can_push_down(const VExprSPtr& expr, size_t
child_id) {
- if (!expr->children()[child_id]->is_literal()) {
- return false;
- }
- // the slot has been checked in _check_slot_can_push_down before calling
this function
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- const auto* literal = static_cast<const
VLiteral*>(expr->children()[child_id].get());
- auto [valid, orc_literal] = _make_orc_literal(slot_ref, literal);
- if (valid) {
- _vliteral_to_orc_literal.insert(std::make_pair(literal, orc_literal));
- }
- return valid;
-}
-
-// check if there are rest children of expr can be pushed down to orc reader
-bool OrcReader::_check_rest_children_can_push_down(const VExprSPtr& expr) {
- if (expr->children().size() < 2) {
- return false;
- }
-
- bool at_least_one_child_can_push_down = false;
- for (size_t i = 1; i < expr->children().size(); ++i) {
- if (_check_literal_can_push_down(expr, i)) {
- at_least_one_child_can_push_down = true;
- }
- }
- return at_least_one_child_can_push_down;
-}
-
-// check if the expr can be pushed down to orc reader
-bool OrcReader::_check_expr_can_push_down(const VExprSPtr& expr) {
- if (expr == nullptr) {
- return false;
- }
-
- switch (expr->op()) {
- case TExprOpcode::COMPOUND_AND:
- // at least one child can be pushed down
- return std::ranges::any_of(expr->children(), [this](const auto& child)
{
- return _check_expr_can_push_down(child);
- });
- case TExprOpcode::COMPOUND_OR:
- // all children must be pushed down
- return std::ranges::all_of(expr->children(), [this](const auto& child)
{
- return _check_expr_can_push_down(child);
- });
- case TExprOpcode::COMPOUND_NOT:
- DCHECK_EQ(expr->children().size(), 1);
- return _check_expr_can_push_down(expr->children()[0]);
-
- case TExprOpcode::GE:
- case TExprOpcode::GT:
- case TExprOpcode::LE:
- case TExprOpcode::LT:
- case TExprOpcode::EQ:
- case TExprOpcode::NE:
- case TExprOpcode::FILTER_IN:
- case TExprOpcode::FILTER_NOT_IN:
- // can't push down if expr is null aware predicate
- return expr->node_type() != TExprNodeType::NULL_AWARE_BINARY_PRED &&
- expr->node_type() != TExprNodeType::NULL_AWARE_IN_PRED &&
- _check_slot_can_push_down(expr) &&
_check_rest_children_can_push_down(expr);
-
- case TExprOpcode::INVALID_OPCODE:
- if (expr->node_type() == TExprNodeType::FUNCTION_CALL) {
- auto fn_name = expr->fn().name.function_name;
- // only support is_null_pred and is_not_null_pred
- if (fn_name == "is_null_pred" || fn_name == "is_not_null_pred") {
- return _check_slot_can_push_down(expr);
- }
- VLOG_CRITICAL << "Unsupported function [funciton=" << fn_name <<
"]";
- }
- return false;
- default:
- VLOG_CRITICAL << "Unsupported Opcode [OpCode=" << expr->op() << "]";
- return false;
- }
-}
-
-void OrcReader::_build_less_than(const VExprSPtr& expr,
- std::unique_ptr<orc::SearchArgumentBuilder>&
builder) {
- DCHECK(expr->children().size() == 2);
- DCHECK(expr->children()[0]->is_slot_ref());
- DCHECK(expr->children()[1]->is_literal());
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- const auto* literal = static_cast<const
VLiteral*>(expr->children()[1].get());
- DCHECK(_vslot_ref_to_orc_predicate_data_type.contains(slot_ref));
- auto predicate_type = _vslot_ref_to_orc_predicate_data_type[slot_ref];
- DCHECK(_vliteral_to_orc_literal.contains(literal));
- auto orc_literal = _vliteral_to_orc_literal.find(literal)->second;
-
builder->lessThan(_table_info_node_ptr->children_file_column_name(slot_ref->expr_name()),
- predicate_type, orc_literal);
-}
-
-void OrcReader::_build_less_than_equals(const VExprSPtr& expr,
-
std::unique_ptr<orc::SearchArgumentBuilder>& builder) {
- DCHECK(expr->children().size() == 2);
- DCHECK(expr->children()[0]->is_slot_ref());
- DCHECK(expr->children()[1]->is_literal());
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- const auto* literal = static_cast<const
VLiteral*>(expr->children()[1].get());
- DCHECK(_vslot_ref_to_orc_predicate_data_type.contains(slot_ref));
- auto predicate_type = _vslot_ref_to_orc_predicate_data_type[slot_ref];
- DCHECK(_vliteral_to_orc_literal.contains(literal));
- auto orc_literal = _vliteral_to_orc_literal.find(literal)->second;
-
builder->lessThanEquals(_table_info_node_ptr->children_file_column_name(slot_ref->expr_name()),
- predicate_type, orc_literal);
-}
-
-void OrcReader::_build_equals(const VExprSPtr& expr,
- std::unique_ptr<orc::SearchArgumentBuilder>&
builder) {
- DCHECK(expr->children().size() == 2);
- DCHECK(expr->children()[0]->is_slot_ref());
- DCHECK(expr->children()[1]->is_literal());
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- const auto* literal = static_cast<const
VLiteral*>(expr->children()[1].get());
- DCHECK(_vslot_ref_to_orc_predicate_data_type.contains(slot_ref));
- auto predicate_type = _vslot_ref_to_orc_predicate_data_type[slot_ref];
- DCHECK(_vliteral_to_orc_literal.contains(literal));
- auto orc_literal = _vliteral_to_orc_literal.find(literal)->second;
-
builder->equals(_table_info_node_ptr->children_file_column_name(slot_ref->expr_name()),
- predicate_type, orc_literal);
-}
-
-void OrcReader::_build_filter_in(const VExprSPtr& expr,
- std::unique_ptr<orc::SearchArgumentBuilder>&
builder) {
- DCHECK(expr->children().size() >= 2);
- DCHECK(expr->children()[0]->is_slot_ref());
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- std::vector<orc::Literal> literals;
- DCHECK(_vslot_ref_to_orc_predicate_data_type.contains(slot_ref));
- orc::PredicateDataType predicate_type =
_vslot_ref_to_orc_predicate_data_type[slot_ref];
- for (size_t i = 1; i < expr->children().size(); ++i) {
- DCHECK(expr->children()[i]->is_literal());
- const auto* literal = static_cast<const
VLiteral*>(expr->children()[i].get());
- if (_vliteral_to_orc_literal.contains(literal)) {
- auto orc_literal = _vliteral_to_orc_literal.find(literal)->second;
- literals.emplace_back(orc_literal);
- }
- }
- DCHECK(!literals.empty());
- if (literals.size() == 1) {
-
builder->equals(_table_info_node_ptr->children_file_column_name(slot_ref->expr_name()),
- predicate_type, literals[0]);
- } else {
-
builder->in(_table_info_node_ptr->children_file_column_name(slot_ref->expr_name()),
- predicate_type, literals);
- }
-}
-
-void OrcReader::_build_is_null(const VExprSPtr& expr,
- std::unique_ptr<orc::SearchArgumentBuilder>&
builder) {
- DCHECK(expr->children().size() == 1);
- DCHECK(expr->children()[0]->is_slot_ref());
- const auto* slot_ref = static_cast<const
VSlotRef*>(expr->children()[0].get());
- DCHECK(_vslot_ref_to_orc_predicate_data_type.contains(slot_ref));
- auto predicate_type = _vslot_ref_to_orc_predicate_data_type[slot_ref];
-
builder->isNull(_table_info_node_ptr->children_file_column_name(slot_ref->expr_name()),
- predicate_type);
-}
-
-bool OrcReader::_build_search_argument(const VExprSPtr& expr,
-
std::unique_ptr<orc::SearchArgumentBuilder>& builder) {
- // OPTIMIZE: check expr only once
- if (!_check_expr_can_push_down(expr)) {
- return false;
- }
- switch (expr->op()) {
- case TExprOpcode::COMPOUND_AND: {
- builder->startAnd();
- bool at_least_one_can_push_down = false;
- for (const auto& child : expr->children()) {
- if (_build_search_argument(child, builder)) {
- at_least_one_can_push_down = true;
- }
- }
- DCHECK(at_least_one_can_push_down);
- builder->end();
- break;
- }
- case TExprOpcode::COMPOUND_OR: {
- builder->startOr();
- bool all_can_push_down = true;
- for (const auto& child : expr->children()) {
- if (!_build_search_argument(child, builder)) {
- all_can_push_down = false;
- }
- }
- DCHECK(all_can_push_down);
- builder->end();
- break;
- }
- case TExprOpcode::COMPOUND_NOT: {
- DCHECK_EQ(expr->children().size(), 1);
- builder->startNot();
- auto res = _build_search_argument(expr->children()[0], builder);
- DCHECK(res);
- builder->end();
- break;
- }
- case TExprOpcode::GE:
- builder->startNot();
- _build_less_than(expr, builder);
- builder->end();
- break;
- case TExprOpcode::GT:
- builder->startNot();
- _build_less_than_equals(expr, builder);
- builder->end();
- break;
- case TExprOpcode::LE:
- _build_less_than_equals(expr, builder);
- break;
- case TExprOpcode::LT:
- _build_less_than(expr, builder);
- break;
- case TExprOpcode::EQ:
- _build_equals(expr, builder);
- break;
- case TExprOpcode::NE:
- builder->startNot();
- _build_equals(expr, builder);
- builder->end();
- break;
- case TExprOpcode::FILTER_IN:
- _build_filter_in(expr, builder);
- break;
- case TExprOpcode::FILTER_NOT_IN:
- builder->startNot();
- _build_filter_in(expr, builder);
- builder->end();
- break;
- // is null and is not null is represented as function call
- case TExprOpcode::INVALID_OPCODE:
- DCHECK(expr->node_type() == TExprNodeType::FUNCTION_CALL);
- if (expr->fn().name.function_name == "is_null_pred") {
- _build_is_null(expr, builder);
- } else if (expr->fn().name.function_name == "is_not_null_pred") {
- builder->startNot();
- _build_is_null(expr, builder);
- builder->end();
- } else {
- // should not reach here, because _check_expr_can_push_down has
already checked
- __builtin_unreachable();
- }
- break;
-
- default:
- // should not reach here, because _check_expr_can_push_down has
already checked
- __builtin_unreachable();
- }
- return true;
-}
-
-bool OrcReader::_init_search_argument(const VExprSPtrs& exprs) {
- // build search argument, if any expr can not be pushed down, return false
- auto builder = orc::SearchArgumentFactory::newBuilder();
- bool at_least_one_can_push_down = false;
- builder->startAnd();
- for (const auto& expr : exprs) {
- _vslot_ref_to_orc_predicate_data_type.clear();
- _vliteral_to_orc_literal.clear();
- if (_build_search_argument(expr, builder)) {
- at_least_one_can_push_down = true;
- }
- }
- if (!at_least_one_can_push_down) {
- // if all exprs can not be pushed down, builder->end() will throw
exception
- return false;
- }
- builder->end();
-
- auto sargs = builder->build();
- _profile->add_info_string("OrcReader SearchArgument: ", sargs->toString());
- _row_reader_options.searchArgument(std::move(sargs));
- return true;
-}
-
-void OrcReader::_collect_predicate_columns_from_conjuncts(
- std::unordered_map<std::string, std::pair<uint32_t, int>>&
predicate_table_columns) {
- std::function<void(VExpr * expr)> visit_slot = [&](VExpr* expr) {
- if (expr->is_slot_ref()) {
- VSlotRef* slot_ref = static_cast<VSlotRef*>(expr);
- auto expr_name = slot_ref->expr_name();
- predicate_table_columns.emplace(
- expr_name, std::make_pair(slot_ref->column_id(),
slot_ref->slot_id()));
- if (slot_ref->column_id() == 0) {
- _lazy_read_ctx.resize_first_column = false;
- }
- return;
- }
- for (auto& child : expr->children()) {
- visit_slot(child.get());
- }
- };
-
- for (const auto& conjunct : _lazy_read_ctx.conjuncts) {
- auto expr = conjunct->root();
-
- if (expr->is_rf_wrapper()) {
- auto* runtime_filter = static_cast<RuntimeFilterExpr*>(expr.get());
- auto filter_impl = runtime_filter->get_impl();
- visit_slot(filter_impl.get());
-
- if ((runtime_filter->node_type() == TExprNodeType::BINARY_PRED) &&
- (runtime_filter->op() == TExprOpcode::GE ||
- runtime_filter->op() == TExprOpcode::LE)) {
- expr = filter_impl;
- } else if (runtime_filter->node_type() == TExprNodeType::IN_PRED &&
- runtime_filter->op() == TExprOpcode::FILTER_IN) {
- auto* direct_in_predicate =
static_cast<VDirectInPredicate*>(filter_impl.get());
- int max_in_size =
-
_state->query_options().__isset.max_pushdown_conditions_per_column
- ?
_state->query_options().max_pushdown_conditions_per_column
- : 1024;
- if (direct_in_predicate->get_set_func()->size() == 0 ||
- direct_in_predicate->get_set_func()->size() > max_in_size)
{
- continue;
- }
- VExprSPtr new_in_slot = nullptr;
- if (direct_in_predicate->get_slot_in_expr(new_in_slot)) {
- expr = new_in_slot;
- } else {
- continue;
- }
- } else {
- continue;
- }
- } else if (VTopNPred* topn_pred = typeid_cast<VTopNPred*>(expr.get()))
{
- DCHECK(topn_pred->children().size() > 0);
- visit_slot(topn_pred->children()[0].get());
- VExprSPtr binary_expr;
- if (topn_pred->get_binary_expr(binary_expr)) {
- expr = binary_expr;
- } else {
- continue;
- }
- } else {
- visit_slot(expr.get());
- }
-
- if (_check_expr_can_push_down(expr)) {
- _push_down_exprs.emplace_back(expr);
- }
- }
-}
-
-void OrcReader::_classify_columns_for_lazy_read(
- const std::unordered_map<std::string, std::pair<uint32_t, int>>&
predicate_table_columns,
- const std::unordered_map<std::string, std::tuple<std::string, const
SlotDescriptor*>>&
- partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) {
- for (auto& read_table_col : _read_table_cols) {
- _lazy_read_ctx.all_read_columns.emplace_back(read_table_col);
- if (!predicate_table_columns.empty()) {
- auto iter = predicate_table_columns.find(read_table_col);
- if (iter == predicate_table_columns.end()) {
- _lazy_read_ctx.lazy_read_columns.emplace_back(read_table_col);
- } else {
-
_lazy_read_ctx.predicate_columns.first.emplace_back(iter->first);
-
_lazy_read_ctx.predicate_columns.second.emplace_back(iter->second.second);
- _lazy_read_ctx.predicate_orc_columns.emplace_back(
-
_table_info_node_ptr->children_file_column_name(iter->first));
- if (!has_column_optimization(read_table_col,
ColumnOptimizationTypes::LAZY_READ)) {
- // Todo : enable lazy mat where filter iceberg row lineage
column.
- _enable_lazy_mat = false;
- }
- }
- }
- }
-
- for (const auto& kv : partition_columns) {
- auto iter = predicate_table_columns.find(kv.first);
- if (iter == predicate_table_columns.end()) {
- _lazy_read_ctx.partition_columns.emplace(kv.first, kv.second);
- } else {
- _lazy_read_ctx.predicate_partition_columns.emplace(kv.first,
kv.second);
- }
- }
-
- for (const auto& kv : missing_columns) {
- auto iter = predicate_table_columns.find(kv.first);
- if (iter == predicate_table_columns.end()) {
- _lazy_read_ctx.missing_columns.emplace(kv.first, kv.second);
- } else {
- if (_slot_id_to_filter_conjuncts->find(iter->second.second) !=
- _slot_id_to_filter_conjuncts->end()) {
- for (const auto& ctx :
-
_slot_id_to_filter_conjuncts->find(iter->second.second)->second) {
- _filter_conjuncts.emplace_back(ctx);
- }
- }
- _lazy_read_ctx.predicate_missing_columns.emplace(kv.first,
kv.second);
- if (!has_column_optimization(kv.first,
ColumnOptimizationTypes::LAZY_READ)) {
- _enable_lazy_mat = false;
- }
- }
- }
-
- if (_enable_lazy_mat && !_lazy_read_ctx.predicate_columns.first.empty() &&
- !_lazy_read_ctx.lazy_read_columns.empty()) {
- _lazy_read_ctx.can_lazy_read = true;
- }
-}
-
-Status OrcReader::_init_orc_row_reader() {
- try {
- _row_reader_options.range(_range_start_offset, _range_size);
-
_row_reader_options.setTimezoneName(normalized_orc_timezone_name(_ctz));
- if (!_column_ids.empty()) {
- std::list<uint64_t> column_ids_list(_column_ids.begin(),
_column_ids.end());
- _row_reader_options.includeTypes(column_ids_list);
- } else {
- _row_reader_options.include(_read_file_cols);
- }
- _row_reader_options.setEnableLazyDecoding(true);
-
- // Tiny stripe optimization (skip when reading by row id)
- if (!_read_by_rows) {
- uint64_t number_of_stripes = _reader->getNumberOfStripes();
- auto all_stripes_needed =
_reader->getNeedReadStripes(_row_reader_options);
- int64_t range_end_offset = _range_start_offset + _range_size;
-
- bool all_tiny_stripes = true;
- std::vector<io::PrefetchRange> tiny_stripe_ranges;
-
- for (uint64_t i = 0; i < number_of_stripes; i++) {
- std::unique_ptr<orc::StripeInformation> strip_info =
_reader->getStripe(i);
- uint64_t strip_start_offset = strip_info->getOffset();
- uint64_t strip_end_offset = strip_start_offset +
strip_info->getLength();
-
- if (strip_start_offset >= range_end_offset ||
- strip_end_offset < _range_start_offset ||
!all_stripes_needed[i]) {
- continue;
- }
- if (strip_info->getLength() >
_orc_tiny_stripe_threshold_bytes) {
- all_tiny_stripes = false;
- break;
- }
- tiny_stripe_ranges.emplace_back(strip_start_offset,
strip_end_offset);
- }
- if (all_tiny_stripes && number_of_stripes > 0) {
- std::vector<io::PrefetchRange> prefetch_merge_ranges =
-
io::PrefetchRange::merge_adjacent_seq_ranges(tiny_stripe_ranges,
-
_orc_max_merge_distance_bytes,
-
_orc_once_max_read_bytes);
- auto range_finder =
std::make_shared<io::LinearProbeRangeFinder>(
- std::move(prefetch_merge_ranges));
- auto* orc_input_stream_ptr =
static_cast<ORCFileInputStream*>(_reader->getStream());
- orc_input_stream_ptr->set_all_tiny_stripes();
- auto& orc_file_reader =
orc_input_stream_ptr->get_file_reader();
- auto orc_inner_reader =
orc_input_stream_ptr->get_inner_reader();
- orc_file_reader = std::make_shared<io::RangeCacheFileReader>(
- _profile, orc_inner_reader, range_finder);
- }
- }
-
- // Merge predicate partition/missing back if can't lazy read
- if (!_lazy_read_ctx.can_lazy_read) {
- for (auto& kv : _lazy_read_ctx.predicate_partition_columns) {
- _lazy_read_ctx.partition_columns.emplace(kv.first, kv.second);
- }
- for (auto& kv : _lazy_read_ctx.predicate_missing_columns) {
- _lazy_read_ctx.missing_columns.emplace(kv.first, kv.second);
- }
- }
-
- // Create ORC row reader
- if (_lazy_read_ctx.can_lazy_read) {
- _row_reader_options.filter(_lazy_read_ctx.predicate_orc_columns);
- _orc_filter = std::make_unique<ORCFilterImpl>(this);
- }
- if (!_lazy_read_ctx.conjuncts.empty()) {
- _string_dict_filter = std::make_unique<StringDictFilterImpl>(this);
- }
- _row_reader = _reader->createRowReader(_row_reader_options,
_orc_filter.get(),
- _string_dict_filter.get());
Review Comment:
Addressed in 5943432fa85. V2 installs the StringDictFilter callback for safe
STRING/VARCHAR predicates, includes NULL, and respects the metadata-pruning
fence. DictionaryRejectsAbsentValuesWithoutRowDecoding covers an absent value
inside the min/max range and verifies stripe rejection without row decoding.
All 187 ORC tests passed again in the latest ASAN run.
--
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]
