github-actions[bot] commented on code in PR #69723:
URL: https://github.com/apache/doris/pull/69723#discussion_r4399387092
##########
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:
[P2] Reuse predicate columns produced by ORC lazy filtering
This unconditional switch exposes redundant materialization in V2's lazy ORC
path. For `SELECT payload FROM t WHERE length(s) > 0` with mostly surviving
large STRING values and a separate payload column, `_filter_orc_batch` decodes
`predicate_columns` into a local `filter_block`, then discards it after saving
row indices. `get_block` immediately decodes those same predicate columns again
before applying that selection. Each surviving batch therefore performs two
full SerDe conversions/string copies; V1 passed the output Block into its
callback and decoded only follower columns afterward. Retain or transfer the
callback-produced predicate columns so V2 materializes them once, and cover
this with a conversion/copy-count test. This occurs with direct encoding
independently of dictionary pruning.
##########
be/src/format/csv/csv_reader.cpp:
##########
@@ -1,980 +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/csv/csv_reader.h"
-
-#include <fmt/format.h>
-#include <gen_cpp/PlanNodes_types.h>
-#include <gen_cpp/Types_types.h>
-#include <glog/logging.h>
-
-#include <algorithm>
-#include <cstddef>
-#include <map>
-#include <memory>
-#include <numeric>
-#include <ostream>
-#include <regex>
-#include <utility>
-
-#include "common/compiler_util.h" // IWYU pragma: keep
-#include "common/config.h"
-#include "common/consts.h"
-#include "common/status.h"
-#include "core/block/block.h"
-#include "core/block/column_with_type_and_name.h"
-#include "core/column/column_nullable.h"
-#include "core/column/column_string.h"
-#include "core/data_type/data_type_factory.hpp"
-#include "core/data_type_serde/data_type_string_serde.h"
-#include "exec/scan/scanner.h"
-#include "format/file_reader/new_plain_binary_line_reader.h"
-#include "format/file_reader/new_plain_text_line_reader.h"
-#include "format/line_reader.h"
-#include "io/file_factory.h"
-#include "io/fs/broker_file_reader.h"
-#include "io/fs/buffered_reader.h"
-#include "io/fs/file_reader.h"
-#include "io/fs/s3_file_reader.h"
-#include "io/fs/tracing_file_reader.h"
-#include "runtime/descriptors.h"
-#include "runtime/runtime_state.h"
-#include "util/decompressor.h"
-#include "util/string_util.h"
-#include "util/utf8_check.h"
-
-namespace doris {
-class RuntimeProfile;
-class IColumn;
-namespace io {
-struct IOContext;
-enum class FileCachePolicy : uint8_t;
-} // namespace io
-} // namespace doris
-
-namespace doris {
-
-namespace {
-
-size_t columns_byte_size(const std::vector<MutableColumnPtr>& columns) {
- size_t bytes = 0;
- for (const auto& column : columns) {
- DCHECK(column.get() != nullptr);
- bytes += column->byte_size();
- }
- return bytes;
-}
-
-} // namespace
-
-void EncloseCsvTextFieldSplitter::do_split(const Slice& line,
std::vector<Slice>* splitted_values) {
- const char* data = line.data;
- const auto& column_sep_positions =
_text_line_reader_ctx->column_sep_positions();
- size_t value_start_offset = 0;
- for (auto idx : column_sep_positions) {
- process_value_func(data, value_start_offset, idx - value_start_offset,
_trimming_char,
- splitted_values);
- value_start_offset = idx + _value_sep_len;
- }
- if (line.size >= value_start_offset) {
- // process the last column
- process_value_func(data, value_start_offset, line.size -
value_start_offset, _trimming_char,
- splitted_values);
- }
-}
-
-void PlainCsvTextFieldSplitter::_split_field_single_char(const Slice& line,
- std::vector<Slice>*
splitted_values) {
- const char* data = line.data;
- const size_t size = line.size;
- size_t value_start = 0;
- for (size_t i = 0; i < size; ++i) {
- if (data[i] == _value_sep[0]) {
- process_value_func(data, value_start, i - value_start,
_trimming_char, splitted_values);
- value_start = i + _value_sep_len;
- }
- }
- process_value_func(data, value_start, size - value_start, _trimming_char,
splitted_values);
-}
-
-void PlainCsvTextFieldSplitter::_split_field_multi_char(const Slice& line,
- std::vector<Slice>*
splitted_values) {
- size_t start = 0; // point to the start pos of next col value.
- size_t curpos = 0; // point to the start pos of separator matching
sequence.
-
- // value_sep : AAAA
- // line.data : 1234AAAA5678
- // -> 1234,5678
-
- // start start
- // ▼ ▼
- // 1234AAAA5678\0
- // ▲ ▲
- // curpos curpos
-
- //kmp
- std::vector<int> next(_value_sep_len);
- next[0] = -1;
- for (int i = 1, j = -1; i < _value_sep_len; i++) {
- while (j > -1 && _value_sep[i] != _value_sep[j + 1]) {
- j = next[j];
- }
- if (_value_sep[i] == _value_sep[j + 1]) {
- j++;
- }
- next[i] = j;
- }
-
- for (int i = 0, j = -1; i < line.size; i++) {
- // i : line
- // j : _value_sep
- while (j > -1 && line[i] != _value_sep[j + 1]) {
- j = next[j];
- }
- if (line[i] == _value_sep[j + 1]) {
- j++;
- }
- if (j == _value_sep_len - 1) {
- curpos = i - _value_sep_len + 1;
-
- /*
- * column_separator : "xx"
- * data.csv : data1xxxxdata2
- *
- * Parse incorrectly:
- * data1[xx]xxdata2
- * data1x[xx]xdata2
- * data1xx[xx]data2
- * The string "xxxx" is parsed into three "xx" delimiters.
- *
- * Parse correctly:
- * data1[xx]xxdata2
- * data1xx[xx]data2
- */
-
- if (curpos >= start) {
- process_value_func(line.data, start, curpos - start,
_trimming_char,
- splitted_values);
- start = i + 1;
- }
-
- j = next[j];
- }
- }
- process_value_func(line.data, start, line.size - start, _trimming_char,
splitted_values);
-}
-
-void PlainCsvTextFieldSplitter::do_split(const Slice& line,
std::vector<Slice>* splitted_values) {
- if (is_single_char_delim) {
- _split_field_single_char(line, splitted_values);
- } else {
- _split_field_multi_char(line, splitted_values);
- }
-}
-
-CsvReader::CsvReader(RuntimeState* state, RuntimeProfile* profile,
ScannerCounter* counter,
- const TFileScanRangeParams& params, const TFileRangeDesc&
range,
- const std::vector<SlotDescriptor*>& file_slot_descs,
size_t batch_size,
- io::IOContext* io_ctx, std::shared_ptr<io::IOContext>
io_ctx_holder)
- : _profile(profile),
- _params(params),
- _file_reader(nullptr),
- _line_reader(nullptr),
- _decompressor(nullptr),
- _state(state),
- _counter(counter),
- _range(range),
- _file_slot_descs(file_slot_descs),
- _line_reader_eof(false),
- _skip_lines(0),
- _io_ctx(io_ctx),
- _io_ctx_holder(std::move(io_ctx_holder)),
- _batch_size(std::max(batch_size, 1UL)) {
- if (_io_ctx == nullptr && _io_ctx_holder) {
- _io_ctx = _io_ctx_holder.get();
- }
- _file_format_type = _params.format_type;
- _is_proto_format = _file_format_type == TFileFormatType::FORMAT_PROTO;
- if (_range.__isset.compress_type) {
- // for compatibility
- _file_compress_type = _range.compress_type;
- } else {
- _file_compress_type = _params.compress_type;
- }
- _size = _range.size;
-
- _split_values.reserve(_file_slot_descs.size());
- _init_system_properties();
- _init_file_description();
- _serdes = create_data_type_serdes(_file_slot_descs);
-}
-
-void CsvReader::_init_system_properties() {
- if (_range.__isset.file_type) {
- // for compatibility
- _system_properties.system_type = _range.file_type;
- } else {
- _system_properties.system_type = _params.file_type;
- }
- _system_properties.properties = _params.properties;
- _system_properties.hdfs_params = _params.hdfs_params;
- if (_params.__isset.broker_addresses) {
-
_system_properties.broker_addresses.assign(_params.broker_addresses.begin(),
-
_params.broker_addresses.end());
- }
-}
-
-void CsvReader::_init_file_description() {
- _file_description.path = _range.path;
- _file_description.file_size = _range.__isset.file_size ? _range.file_size
: -1;
- if (_range.__isset.fs_name) {
- _file_description.fs_name = _range.fs_name;
- }
- if (_range.__isset.file_cache_admission) {
- _file_description.file_cache_admission = _range.file_cache_admission;
- }
-}
-
-Status CsvReader::init_reader(bool is_load) {
- // set the skip lines and start offset
- _start_offset = _range.start_offset;
- if (_start_offset == 0) {
- // check header typer first
- if (_params.__isset.file_attributes &&
_params.file_attributes.__isset.header_type &&
- !_params.file_attributes.header_type.empty()) {
- std::string header_type =
to_lower(_params.file_attributes.header_type);
- if (header_type == BeConsts::CSV_WITH_NAMES) {
- _skip_lines = 1;
- } else if (header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
- _skip_lines = 2;
- }
- } else if (_params.file_attributes.__isset.skip_lines) {
- _skip_lines = _params.file_attributes.skip_lines;
- }
- } else if (_start_offset != 0) {
- if ((_file_compress_type != TFileCompressType::PLAIN) ||
- (_file_compress_type == TFileCompressType::UNKNOWN &&
- _file_format_type != TFileFormatType::FORMAT_CSV_PLAIN)) {
- return Status::InternalError<false>("For now we do not support
split compressed file");
- }
- // pre-read to promise first line skipped always read
- int64_t pre_read_len = std::min(
-
static_cast<int64_t>(_params.file_attributes.text_params.line_delimiter.size()),
- _start_offset);
- _start_offset -= pre_read_len;
- _size += pre_read_len;
- // not first range will always skip one line
- _skip_lines = 1;
- }
-
- _use_nullable_string_opt.resize(_file_slot_descs.size());
- for (int i = 0; i < _file_slot_descs.size(); ++i) {
- auto data_type_ptr = _file_slot_descs[i]->type();
- if (data_type_ptr->is_nullable() &&
is_string_type(data_type_ptr->get_primitive_type())) {
- _use_nullable_string_opt[i] = 1;
- }
- }
-
- RETURN_IF_ERROR(_init_options());
- RETURN_IF_ERROR(_create_file_reader(false));
- RETURN_IF_ERROR(_create_decompressor());
- RETURN_IF_ERROR(_create_line_reader());
-
- _is_load = is_load;
- if (!_is_load) {
- // For query task, there are 2 slot mapping.
- // One is from file slot to values in line.
- // eg, the file_slot_descs is k1, k3, k5, and values in line are
k1, k2, k3, k4, k5
- // the _col_idxs will save: 0, 2, 4
- // The other is from file slot to columns in output block
- // eg, the file_slot_descs is k1, k3, k5, and columns in block
are p1, k1, k3, k5
- // where "p1" is the partition col which does not exist in file
- // the _file_slot_idx_map will save: 1, 2, 3
- DCHECK(_params.__isset.column_idxs);
- _col_idxs = _params.column_idxs;
- int idx = 0;
- for (const auto& slot_info : _params.required_slots) {
- if (slot_info.is_file_slot) {
- _file_slot_idx_map.push_back(idx);
- }
- idx++;
- }
- } else {
- // For load task, the column order is same as file column order
- int i = 0;
- for (const auto& desc [[maybe_unused]] : _file_slot_descs) {
- _col_idxs.push_back(i++);
- }
- }
-
- _line_reader_eof = false;
- return Status::OK();
-}
-
-// ---- Unified init_reader(ReaderInitContext*) overrides ----
-
-Status CsvReader::_open_file_reader(ReaderInitContext* base_ctx) {
- _start_offset = _range.start_offset;
- if (_start_offset == 0) {
- if (_params.__isset.file_attributes &&
_params.file_attributes.__isset.header_type &&
- !_params.file_attributes.header_type.empty()) {
- std::string header_type =
to_lower(_params.file_attributes.header_type);
- if (header_type == BeConsts::CSV_WITH_NAMES) {
- _skip_lines = 1;
- } else if (header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
- _skip_lines = 2;
- }
- } else if (_params.file_attributes.__isset.skip_lines) {
- _skip_lines = _params.file_attributes.skip_lines;
- }
- } else if (_start_offset != 0) {
- if ((_file_compress_type != TFileCompressType::PLAIN) ||
- (_file_compress_type == TFileCompressType::UNKNOWN &&
- _file_format_type != TFileFormatType::FORMAT_CSV_PLAIN)) {
- return Status::InternalError<false>("For now we do not support
split compressed file");
- }
- int64_t pre_read_len = std::min(
-
static_cast<int64_t>(_params.file_attributes.text_params.line_delimiter.size()),
- _start_offset);
- _start_offset -= pre_read_len;
- _size += pre_read_len;
- _skip_lines = 1;
- }
-
- RETURN_IF_ERROR(_init_options());
- RETURN_IF_ERROR(_create_file_reader(false));
- return Status::OK();
-}
-
-Status CsvReader::_do_init_reader(ReaderInitContext* base_ctx) {
- auto* ctx = checked_context_cast<CsvInitContext>(base_ctx);
- _is_load = ctx->is_load;
-
- _use_nullable_string_opt.resize(_file_slot_descs.size());
- for (int i = 0; i < _file_slot_descs.size(); ++i) {
- auto data_type_ptr = _file_slot_descs[i]->type();
- if (data_type_ptr->is_nullable() &&
is_string_type(data_type_ptr->get_primitive_type())) {
- _use_nullable_string_opt[i] = 1;
- }
- }
-
- RETURN_IF_ERROR(_create_decompressor());
- RETURN_IF_ERROR(_create_line_reader());
-
- if (!_is_load) {
- DCHECK(_params.__isset.column_idxs);
- _col_idxs = _params.column_idxs;
- int idx = 0;
- for (const auto& slot_info : _params.required_slots) {
- if (slot_info.is_file_slot) {
- _file_slot_idx_map.push_back(idx);
- }
- idx++;
- }
- } else {
- int i = 0;
- for (const auto& desc [[maybe_unused]] : _file_slot_descs) {
- _col_idxs.push_back(i++);
- }
- }
- _line_reader_eof = false;
- return Status::OK();
-}
-
-void CsvReader::set_batch_size(size_t batch_size) {
- // 0 means "not set" / "use default" for the row-based readers; we must
- // never let _batch_size be 0 because _do_get_next_block uses it as the
- // upper bound of a `while (rows < _batch_size)` loop and a 0 would make
- // the reader return empty blocks and incorrectly signal EOF.
- _batch_size = std::max(batch_size, 1UL);
-}
-
-// !FIXME: Here we should use MutableBlock
-Status CsvReader::_do_get_next_block(Block* block, size_t* read_rows, bool*
eof) {
- if (_line_reader_eof) {
- *eof = true;
- return Status::OK();
- }
-
- const size_t batch_size = _batch_size;
- const auto max_block_bytes = _state->preferred_block_size_bytes();
- size_t rows = 0;
-
- bool success = false;
- bool is_remove_bom = false;
- if (_range.start_offset != 0 && _skip_lines > 0 && _enclose == 0 &&
- _file_format_type == TFileFormatType::FORMAT_CSV_PLAIN) {
- auto* text_reader =
assert_cast<NewPlainTextLineReader*>(_line_reader.get());
- RETURN_IF_ERROR(text_reader->skip_split_prefix(_range.start_offset,
_line_delimiter,
- &_line_reader_eof,
_io_ctx));
- _skip_lines = 0;
- is_remove_bom = true;
- }
- if (_push_down_agg_type == TPushAggOp::type::COUNT) {
- while (rows < batch_size && !_line_reader_eof) {
- const uint8_t* ptr = nullptr;
- size_t size = 0;
- RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx));
-
- // _skip_lines == 0 means this line is the actual data beginning
line for the entire file
- // is_remove_bom means _remove_bom should only execute once
- if (_skip_lines == 0 && !is_remove_bom) {
- ptr = _remove_bom(ptr, size);
- is_remove_bom = true;
- }
-
- // _skip_lines > 0 means we do not need to remove bom
- if (_skip_lines > 0) {
- _skip_lines--;
- is_remove_bom = true;
- continue;
- }
- if (size == 0) {
- if (!_line_reader_eof) {
- if (_empty_line_as_record() ||
_state->is_read_csv_empty_line_as_null()) {
- ++rows;
- }
- }
- // Read empty line, continue
- continue;
- }
-
- RETURN_IF_ERROR(_validate_line(Slice(ptr, size), &success));
- ++rows;
- }
- auto mutable_columns_guard = block->mutate_columns_scoped();
- auto& mutate_columns = mutable_columns_guard.mutable_columns();
- for (auto& col : mutate_columns) {
- col->resize(rows);
- }
- } else {
- auto columns_guard = block->mutate_columns_scoped();
- auto& columns = columns_guard.mutable_columns();
- _reserve_nullable_string_columns(columns, batch_size);
- while (rows < batch_size && !_line_reader_eof &&
- (columns_byte_size(columns) < max_block_bytes)) {
- const uint8_t* ptr = nullptr;
- size_t size = 0;
- RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx));
-
- // _skip_lines == 0 means this line is the actual data beginning
line for the entire file
- // is_remove_bom means _remove_bom should only execute once
- if (!is_remove_bom && _skip_lines == 0) {
- ptr = _remove_bom(ptr, size);
- is_remove_bom = true;
- }
-
- // _skip_lines > 0 means we do not remove bom
- if (_skip_lines > 0) {
- _skip_lines--;
- is_remove_bom = true;
- continue;
- }
- if (size == 0) {
- if (!_line_reader_eof) {
- if (_empty_line_as_record()) {
- Slice empty_line("", 0);
- RETURN_IF_ERROR(_validate_line(empty_line, &success));
- if (success) {
- RETURN_IF_ERROR(_fill_dest_columns(empty_line,
columns, &rows));
- }
- } else if (_state->is_read_csv_empty_line_as_null()) {
- RETURN_IF_ERROR(_fill_empty_line(columns, &rows));
- }
- }
- // Read empty line, continue
- continue;
- }
-
- RETURN_IF_ERROR(_validate_line(Slice(ptr, size), &success));
- if (!success) {
- continue;
- }
- RETURN_IF_ERROR(_fill_dest_columns(Slice(ptr, size), columns,
&rows));
- }
- }
-
- *eof = (rows == 0);
- *read_rows = rows;
-
- return Status::OK();
-}
-
-Status CsvReader::_get_columns_impl(std::unordered_map<std::string,
DataTypePtr>* name_to_type) {
- for (const auto& slot : _file_slot_descs) {
- name_to_type->emplace(slot->col_name(), slot->type());
- }
- return Status::OK();
-}
-
-// init decompressor, file reader and line reader for parsing schema
-Status CsvReader::init_schema_reader() {
- _start_offset = _range.start_offset;
- if (_start_offset != 0) {
- return Status::InvalidArgument(
- "start offset of TFileRangeDesc must be zero in get parsered
schema");
- }
- if (_params.file_type == TFileType::FILE_BROKER) {
- return Status::InternalError<false>(
- "Getting parsered schema from csv file do not support stream
load and broker "
- "load.");
- }
-
- // csv file without names line and types line.
- _read_line = 1;
- _is_parse_name = false;
-
- if (_params.__isset.file_attributes &&
_params.file_attributes.__isset.header_type &&
- !_params.file_attributes.header_type.empty()) {
- std::string header_type =
to_lower(_params.file_attributes.header_type);
- if (header_type == BeConsts::CSV_WITH_NAMES) {
- _is_parse_name = true;
- } else if (header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
- _read_line = 2;
- _is_parse_name = true;
- }
- }
-
- RETURN_IF_ERROR(_init_options());
- RETURN_IF_ERROR(_create_file_reader(true));
- RETURN_IF_ERROR(_create_decompressor());
- RETURN_IF_ERROR(_create_line_reader());
- return Status::OK();
-}
-
-Status CsvReader::get_parsed_schema(std::vector<std::string>* col_names,
- std::vector<DataTypePtr>* col_types) {
- if (_read_line == 1) {
- if (!_is_parse_name) { //parse csv file without names and types
- size_t col_nums = 0;
- RETURN_IF_ERROR(_parse_col_nums(&col_nums));
- for (size_t i = 0; i < col_nums; ++i) {
- col_names->emplace_back("c" + std::to_string(i + 1));
- }
- } else { // parse csv file with names
- RETURN_IF_ERROR(_parse_col_names(col_names));
- }
-
- for (size_t j = 0; j < col_names->size(); ++j) {
- col_types->emplace_back(
-
DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_STRING, true));
- }
- } else { // parse csv file with names and types
- RETURN_IF_ERROR(_parse_col_names(col_names));
- RETURN_IF_ERROR(_parse_col_types(col_names->size(), col_types));
- }
- return Status::OK();
-}
-
-Status CsvReader::_deserialize_nullable_string(IColumn& column, Slice& slice) {
- // This is the per-row per-column hot path of CSV load (load reads every
column as
- // nullable string). The column type was already verified by the checked
assert_cast
- // in _reserve_nullable_string_columns at the beginning of the batch, so
the casts
- // here can skip the release-build typeid check.
- auto& null_column = assert_cast<ColumnNullable&,
TypeCheckOnRelease::DISABLE>(column);
- auto& string_column = assert_cast<ColumnString&,
TypeCheckOnRelease::DISABLE>(
- null_column.get_nested_column());
- if (_empty_field_as_null && slice.size == 0) {
- string_column.insert_default();
- null_column.get_null_map_data().push_back(1);
- return Status::OK();
- }
- if (_options.null_len > 0 && !(_options.converted_from_string &&
slice.trim_double_quotes())) {
- if (slice.compare(Slice(_options.null_format, _options.null_len)) ==
0) {
- string_column.insert_default();
- null_column.get_null_map_data().push_back(1);
- return Status::OK();
- }
- }
- // Same as DataTypeStringSerDe::deserialize_one_cell_from_csv (which never
fails),
- // written out here to skip the SerDe layer and its per-cell assert_cast.
- if (_options.escape_char != 0) {
- escape_string_for_csv(slice.data, &slice.size, _options.escape_char,
_options.quote_char);
- }
- string_column.insert_data(slice.data, slice.size);
- null_column.get_null_map_data().push_back(0);
- return Status::OK();
-}
-
-Status CsvReader::_init_options() {
- // get column_separator and line_delimiter
- _value_separator = _params.file_attributes.text_params.column_separator;
- _value_separator_length = _value_separator.size();
- _line_delimiter = _params.file_attributes.text_params.line_delimiter;
- _line_delimiter_length = _line_delimiter.size();
- if (_params.file_attributes.text_params.__isset.enclose) {
- _enclose = _params.file_attributes.text_params.enclose;
- }
- if (_params.file_attributes.text_params.__isset.escape) {
- _escape = _params.file_attributes.text_params.escape;
- }
-
- _trim_tailing_spaces =
- (_state != nullptr &&
_state->trim_tailing_spaces_for_external_table_query());
-
- _options.escape_char = _escape;
- _options.quote_char = _enclose;
-
- if (_params.file_attributes.text_params.collection_delimiter.empty()) {
- _options.collection_delim = ',';
- } else {
- _options.collection_delim =
_params.file_attributes.text_params.collection_delimiter[0];
- }
- if (_params.file_attributes.text_params.mapkv_delimiter.empty()) {
- _options.map_key_delim = ':';
- } else {
- _options.map_key_delim =
_params.file_attributes.text_params.mapkv_delimiter[0];
- }
-
- if (_params.file_attributes.text_params.__isset.null_format) {
- _options.null_format =
_params.file_attributes.text_params.null_format.data();
- _options.null_len =
_params.file_attributes.text_params.null_format.length();
- }
-
- if (_params.file_attributes.__isset.trim_double_quotes) {
- _trim_double_quotes = _params.file_attributes.trim_double_quotes;
- }
- _options.converted_from_string = _trim_double_quotes;
-
- if (_state != nullptr) {
- _keep_cr = _state->query_options().keep_carriage_return;
- }
-
- if (_params.file_attributes.text_params.__isset.empty_field_as_null) {
- _empty_field_as_null =
_params.file_attributes.text_params.empty_field_as_null;
- }
- return Status::OK();
-}
-
-Status CsvReader::_create_decompressor() {
- if (_file_compress_type != TFileCompressType::UNKNOWN) {
- RETURN_IF_ERROR(Decompressor::create_decompressor(_file_compress_type,
&_decompressor));
- } else {
- RETURN_IF_ERROR(Decompressor::create_decompressor(_file_format_type,
&_decompressor));
- }
-
- return Status::OK();
-}
-
-Status CsvReader::_create_file_reader(bool need_schema) {
- if (_params.file_type == TFileType::FILE_STREAM) {
- RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id,
&_file_reader, _state,
- need_schema));
- } else {
- _file_description.mtime = _range.__isset.modification_time ?
_range.modification_time : 0;
- io::FileReaderOptions reader_options = FileFactory::get_reader_options(
- _state ? _state->query_options() : _default_query_options,
_file_description);
- io::FileReaderSPtr file_reader;
- if (_io_ctx_holder) {
- file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
- _profile, _system_properties, _file_description,
reader_options,
- io::DelegateReader::AccessMode::SEQUENTIAL,
Review Comment:
[P2] Preserve sequential S3 prefetch in the replacement text readers
The removed CSV/TEXT open path, and its JSON sibling, used
`DelegateReader(SEQUENTIAL)`, which adds asynchronous `PrefetchBufferedReader`
for S3. V2's `DelimitedTextReader::_open_file` and
`JsonReader::_open_file_reader` call `FileFactory` directly and add only
tracing; their line readers consequently wait on synchronous S3 reads. A large
CSV/TEXT/JSON-lines query split with `enable_file_cache=false` that previously
selected V1 now loses concurrent read-ahead and I/O/parsing overlap, especially
with one scanner on a high-latency object store. Preserve the sequential
delegate/prefetch wrapper, including its shared IOContext ownership, in both V2
opening paths before deleting the working implementation.
##########
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:
[P2] Preserve ORC dictionary pruning before retiring this reader
Removing this callback and forcing former V1 queries onto V2 loses
whole-stripe dictionary rejection. For example, with STRING dictionary
`{'aa','zz'}`, no Bloom filter, and row-group bounds spanning `'mm'`, `WHERE s
= 'mm'` previously rejected the stripe after evaluating two dictionary entries.
V2's `_create_row_reader` supplies no `StringDictFilter`; the pinned [ORC
SDK](https://github.com/apache/doris-thirdparty/blob/5860f94768b94406f31698ac66f4cc3d2a781324/c%2B%2B/src/Reader.cc#L1343)
only performs dictionary rejection when that callback is present. V2 therefore
reads/decompresses the row data and materializes/evaluates every predicate row
for scans that previously worked with `enable_file_scanner_v2=false`. Preserve
dictionary pruning in V2 before removing this path, with an
absent-value-inside-minmax fixture that checks avoided decoding.
--
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]
