Gabriel39 commented on code in PR #69723: URL: https://github.com/apache/doris/pull/69723#discussion_r4420465978
########## 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: Addressed in 5943432fa85. Both V2 delimited-text and JSON opening paths retain DelegateReader(SEQUENTIAL) with shared IOContext ownership. CsvV2ReaderTest.S3PrefetchPreservesSplitRowsAndReadStatistics passed again in the latest ASAN run. This restores the wrapper and its lifecycle; no end-to-end performance benchmark is claimed. ########## 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 5943432fa85. The ORC lazy callback retains its materialized predicate block, and get_block reuses those columns after applying the selected row indices. LazyPredicateColumnsAreDecodedOncePerBatch checks that predicate columns are converted once per batch. 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]
