| // 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/parquet/vparquet_reader.h" |
| |
| #include <gen_cpp/Metrics_types.h> |
| #include <gen_cpp/PlanNodes_types.h> |
| #include <gen_cpp/parquet_types.h> |
| #include <glog/logging.h> |
| |
| #include <algorithm> |
| #include <functional> |
| #include <utility> |
| |
| #include "common/config.h" |
| #include "common/status.h" |
| #include "core/block/block.h" |
| #include "core/block/column_with_type_and_name.h" |
| #include "core/column/column.h" |
| #include "core/data_type/define_primitive_type.h" |
| #include "core/typeid_cast.h" |
| #include "core/types.h" |
| #include "exec/scan/file_scanner.h" |
| #include "exprs/runtime_filter_expr.h" |
| #include "exprs/vbloom_predicate.h" |
| #include "exprs/vdirect_in_predicate.h" |
| #include "exprs/vexpr.h" |
| #include "exprs/vexpr_context.h" |
| #include "exprs/vin_predicate.h" |
| #include "exprs/vslot_ref.h" |
| #include "exprs/vtopn_pred.h" |
| #include "format/column_type_convert.h" |
| #include "format/parquet/parquet_block_split_bloom_filter.h" |
| #include "format/parquet/parquet_common.h" |
| #include "format/parquet/parquet_predicate.h" |
| #include "format/parquet/parquet_thrift_util.h" |
| #include "format/parquet/schema_desc.h" |
| #include "format/parquet/vparquet_file_metadata.h" |
| #include "format/parquet/vparquet_group_reader.h" |
| #include "format/parquet/vparquet_page_index.h" |
| #include "information_schema/schema_scanner.h" |
| #include "io/file_factory.h" |
| #include "io/fs/buffered_reader.h" |
| #include "io/fs/file_reader.h" |
| #include "io/fs/file_reader_writer_fwd.h" |
| #include "io/fs/tracing_file_reader.h" |
| #include "runtime/descriptors.h" |
| #include "util/slice.h" |
| #include "util/string_util.h" |
| #include "util/timezone_utils.h" |
| |
| namespace cctz { |
| class time_zone; |
| } // namespace cctz |
| namespace doris { |
| class RowDescriptor; |
| class RuntimeState; |
| class SlotDescriptor; |
| class TupleDescriptor; |
| namespace io { |
| struct IOContext; |
| enum class FileCachePolicy : uint8_t; |
| } // namespace io |
| class Block; |
| } // namespace doris |
| |
| namespace doris { |
| |
| ParquetReader::ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams& params, |
| const TFileRangeDesc& range, size_t batch_size, |
| const cctz::time_zone* ctz, io::IOContext* io_ctx, RuntimeState* state, |
| FileMetaCache* meta_cache, bool enable_lazy_mat) |
| : _profile(profile), |
| _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), |
| _state(state), |
| _enable_lazy_mat(enable_lazy_mat), |
| _enable_filter_by_min_max( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_min_max), |
| _enable_filter_by_bloom_filter( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_bloom_filter) { |
| _meta_cache = meta_cache; |
| _init_profile(); |
| _init_system_properties(); |
| _init_file_description(); |
| } |
| |
| void ParquetReader::set_batch_size(size_t batch_size) { |
| if (_batch_size == batch_size) { |
| return; |
| } |
| _batch_size = batch_size; |
| } |
| |
| ParquetReader::ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams& params, |
| const TFileRangeDesc& range, size_t batch_size, |
| const cctz::time_zone* ctz, |
| std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state, |
| FileMetaCache* meta_cache, bool enable_lazy_mat) |
| : _profile(profile), |
| _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)), |
| _state(state), |
| _enable_lazy_mat(enable_lazy_mat), |
| _enable_filter_by_min_max( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_min_max), |
| _enable_filter_by_bloom_filter( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_bloom_filter) { |
| _meta_cache = meta_cache; |
| _init_profile(); |
| _init_system_properties(); |
| _init_file_description(); |
| } |
| |
| ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range, |
| io::IOContext* io_ctx, RuntimeState* state, FileMetaCache* meta_cache, |
| bool enable_lazy_mat) |
| : _profile(nullptr), |
| _scan_params(params), |
| _scan_range(range), |
| _io_ctx(io_ctx), |
| _state(state), |
| _enable_lazy_mat(enable_lazy_mat), |
| _enable_filter_by_min_max( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_min_max), |
| _enable_filter_by_bloom_filter( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_bloom_filter) { |
| _meta_cache = meta_cache; |
| _init_system_properties(); |
| _init_file_description(); |
| } |
| |
| ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range, |
| std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state, |
| FileMetaCache* meta_cache, bool enable_lazy_mat) |
| : _profile(nullptr), |
| _scan_params(params), |
| _scan_range(range), |
| _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr), |
| _io_ctx_holder(std::move(io_ctx_holder)), |
| _state(state), |
| _enable_lazy_mat(enable_lazy_mat), |
| _enable_filter_by_min_max( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_min_max), |
| _enable_filter_by_bloom_filter( |
| state == nullptr ? true |
| : state->query_options().enable_parquet_filter_by_bloom_filter) { |
| _meta_cache = meta_cache; |
| _init_system_properties(); |
| _init_file_description(); |
| } |
| |
| ParquetReader::~ParquetReader() { |
| _close_internal(); |
| } |
| |
| #ifdef BE_TEST |
| // for unit test |
| void ParquetReader::set_file_reader(io::FileReaderSPtr file_reader) { |
| _file_reader = file_reader; |
| _tracing_file_reader = file_reader; |
| } |
| #endif |
| |
| void ParquetReader::_init_profile() { |
| if (_profile != nullptr) { |
| static const char* parquet_profile = "ParquetReader"; |
| ADD_TIMER_WITH_LEVEL(_profile, parquet_profile, 1); |
| |
| _parquet_profile.filtered_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "RowGroupsFiltered", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.filtered_row_groups_by_min_max = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "RowGroupsFilteredByMinMax", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.filtered_row_groups_by_bloom_filter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "RowGroupsFilteredByBloomFilter", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.to_read_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "RowGroupsReadNum", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.total_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "RowGroupsTotalNum", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.filtered_group_rows = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "FilteredRowsByGroup", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.filtered_page_rows = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "FilteredRowsByPage", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "FilteredRowsByLazyRead", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.filtered_bytes = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "FilteredBytes", TUnit::BYTES, parquet_profile, 1); |
| _parquet_profile.raw_rows_read = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "RawRowsRead", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.column_read_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ColumnReadTime", parquet_profile, 1); |
| _parquet_profile.parse_meta_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ParseMetaTime", parquet_profile, 1); |
| _parquet_profile.parse_footer_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ParseFooterTime", parquet_profile, 1); |
| _parquet_profile.file_reader_create_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "FileReaderCreateTime", parquet_profile, 1); |
| _parquet_profile.open_file_num = |
| ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FileNum", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_index_read_calls = |
| ADD_COUNTER_WITH_LEVEL(_profile, "PageIndexReadCalls", TUnit::UNIT, 1); |
| _parquet_profile.page_index_filter_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexFilterTime", parquet_profile, 1); |
| _parquet_profile.read_page_index_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexReadTime", parquet_profile, 1); |
| _parquet_profile.parse_page_index_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexParseTime", parquet_profile, 1); |
| _parquet_profile.row_group_filter_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "RowGroupFilterTime", parquet_profile, 1); |
| _parquet_profile.file_footer_read_calls = |
| ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterReadCalls", TUnit::UNIT, 1); |
| _parquet_profile.file_footer_hit_cache = |
| ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterHitCache", TUnit::UNIT, 1); |
| _parquet_profile.decompress_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecompressTime", parquet_profile, 1); |
| _parquet_profile.decompress_cnt = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "DecompressCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_read_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageReadCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheWriteCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_compressed_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheCompressedWriteCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_decompressed_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheDecompressedWriteCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheHitCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_missing_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheMissingCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_compressed_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheCompressedHitCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.page_cache_decompressed_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "PageCacheDecompressedHitCount", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.decode_header_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageHeaderDecodeTime", parquet_profile, 1); |
| _parquet_profile.read_page_header_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageHeaderReadTime", parquet_profile, 1); |
| _parquet_profile.decode_value_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeValueTime", parquet_profile, 1); |
| _parquet_profile.decode_dict_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeDictTime", parquet_profile, 1); |
| _parquet_profile.decode_level_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeLevelTime", parquet_profile, 1); |
| _parquet_profile.decode_null_map_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeNullMapTime", parquet_profile, 1); |
| _parquet_profile.skip_page_header_num = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "SkipPageHeaderNum", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.parse_page_header_num = ADD_CHILD_COUNTER_WITH_LEVEL( |
| _profile, "ParsePageHeaderNum", TUnit::UNIT, parquet_profile, 1); |
| _parquet_profile.predicate_filter_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PredicateFilterTime", parquet_profile, 1); |
| _parquet_profile.dict_filter_rewrite_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DictFilterRewriteTime", parquet_profile, 1); |
| _parquet_profile.convert_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ConvertTime", parquet_profile, 1); |
| _parquet_profile.bloom_filter_read_time = |
| ADD_CHILD_TIMER_WITH_LEVEL(_profile, "BloomFilterReadTime", parquet_profile, 1); |
| } |
| } |
| |
| Status ParquetReader::close() { |
| _close_internal(); |
| return Status::OK(); |
| } |
| |
| void ParquetReader::_close_internal() { |
| if (!_closed) { |
| _current_group_reader.reset(); |
| _tracing_file_reader.reset(); |
| _file_reader.reset(); |
| _closed = true; |
| } |
| } |
| |
| Status ParquetReader::_open_file() { |
| if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) { |
| return Status::EndOfFile("stop"); |
| } |
| if (_file_reader == nullptr) { |
| SCOPED_RAW_TIMER(&_reader_statistics.file_reader_create_time); |
| ++_reader_statistics.open_file_num; |
| _file_description.mtime = |
| _scan_range.__isset.modification_time ? _scan_range.modification_time : 0; |
| io::FileReaderOptions reader_options = |
| FileFactory::get_reader_options(_state, _file_description); |
| if (_io_ctx_holder) { |
| _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader( |
| _profile, _system_properties, _file_description, reader_options, |
| io::DelegateReader::AccessMode::RANDOM, |
| std::static_pointer_cast<const io::IOContext>(_io_ctx_holder))); |
| } else { |
| _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader( |
| _profile, _system_properties, _file_description, reader_options, |
| io::DelegateReader::AccessMode::RANDOM, _io_ctx)); |
| } |
| _tracing_file_reader = _io_ctx && _io_ctx->file_reader_stats |
| ? std::make_shared<io::TracingFileReader>( |
| _file_reader, _io_ctx->file_reader_stats) |
| : _file_reader; |
| } |
| |
| if (_file_metadata == nullptr) { |
| SCOPED_RAW_TIMER(&_reader_statistics.parse_footer_time); |
| if (_tracing_file_reader->size() <= sizeof(PARQUET_VERSION_NUMBER)) { |
| // Some system may generate parquet file with only 4 bytes: PAR1 |
| // Should consider it as empty file. |
| return Status::EndOfFile("open file failed, empty parquet file {} with size: {}", |
| _scan_range.path, _tracing_file_reader->size()); |
| } |
| size_t meta_size = 0; |
| bool enable_mapping_varbinary = _scan_params.__isset.enable_mapping_varbinary |
| ? _scan_params.enable_mapping_varbinary |
| : false; |
| bool enable_mapping_timestamp_tz = _scan_params.__isset.enable_mapping_timestamp_tz |
| ? _scan_params.enable_mapping_timestamp_tz |
| : false; |
| if (_meta_cache == nullptr) { |
| // wrap _file_metadata with unique ptr, so that it can be released finally. |
| RETURN_IF_ERROR(parse_thrift_footer(_tracing_file_reader, &_file_metadata_ptr, |
| &meta_size, _io_ctx, enable_mapping_varbinary, |
| enable_mapping_timestamp_tz)); |
| _file_metadata = _file_metadata_ptr.get(); |
| // parse magic number & parse meta data |
| _reader_statistics.file_footer_read_calls += 1; |
| } else { |
| const auto& file_meta_cache_key = |
| FileMetaCache::get_key(_tracing_file_reader, _file_description); |
| if (!_meta_cache->lookup(file_meta_cache_key, &_meta_cache_handle)) { |
| RETURN_IF_ERROR(parse_thrift_footer(_tracing_file_reader, &_file_metadata_ptr, |
| &meta_size, _io_ctx, enable_mapping_varbinary, |
| enable_mapping_timestamp_tz)); |
| // _file_metadata_ptr.release() : move control of _file_metadata to _meta_cache_handle |
| _meta_cache->insert(file_meta_cache_key, _file_metadata_ptr.release(), |
| &_meta_cache_handle); |
| _file_metadata = _meta_cache_handle.data<FileMetaData>(); |
| _reader_statistics.file_footer_read_calls += 1; |
| } else { |
| _reader_statistics.file_footer_hit_cache++; |
| } |
| _file_metadata = _meta_cache_handle.data<FileMetaData>(); |
| } |
| |
| if (_file_metadata == nullptr) { |
| return Status::InternalError("failed to get file meta data: {}", |
| _file_description.path); |
| } |
| } |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::get_file_metadata_schema(const FieldDescriptor** ptr) { |
| RETURN_IF_ERROR(_open_file()); |
| DCHECK(_file_metadata != nullptr); |
| *ptr = &_file_metadata->schema(); |
| return Status::OK(); |
| } |
| |
| void ParquetReader::_init_system_properties() { |
| if (_scan_range.__isset.file_type) { |
| // for compatibility |
| _system_properties.system_type = _scan_range.file_type; |
| } else { |
| _system_properties.system_type = _scan_params.file_type; |
| } |
| _system_properties.properties = _scan_params.properties; |
| _system_properties.hdfs_params = _scan_params.hdfs_params; |
| if (_scan_params.__isset.broker_addresses) { |
| _system_properties.broker_addresses.assign(_scan_params.broker_addresses.begin(), |
| _scan_params.broker_addresses.end()); |
| } |
| } |
| |
| void ParquetReader::_init_file_description() { |
| _file_description.path = _scan_range.path; |
| _file_description.file_size = _scan_range.__isset.file_size ? _scan_range.file_size : -1; |
| if (_scan_range.__isset.fs_name) { |
| _file_description.fs_name = _scan_range.fs_name; |
| } |
| if (_scan_range.__isset.file_cache_admission) { |
| _file_description.file_cache_admission = _scan_range.file_cache_admission; |
| } |
| } |
| |
| Status ParquetReader::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 Parquet file metadata with case-insensitive recursive matching. |
| // File is already opened by init_reader before this hook, so metadata is available. |
| // tuple_descriptor may be null in unit tests that only set column_descs. |
| if (ctx->tuple_descriptor != nullptr) { |
| const FieldDescriptor* field_desc = nullptr; |
| RETURN_IF_ERROR(get_file_metadata_schema(&field_desc)); |
| RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_name( |
| ctx->tuple_descriptor, *field_desc, ctx->table_info_node)); |
| } |
| |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::_open_file_reader(ReaderInitContext* /*ctx*/) { |
| return _open_file(); |
| } |
| |
| Status ParquetReader::_do_init_reader(ReaderInitContext* base_ctx) { |
| auto* ctx = checked_context_cast<ParquetInitContext>(base_ctx); |
| _col_name_to_block_idx = base_ctx->col_name_to_block_idx; |
| _tuple_descriptor = ctx->tuple_descriptor; |
| _row_descriptor = ctx->row_descriptor; |
| _colname_to_slot_id = ctx->colname_to_slot_id; |
| _not_single_slot_filter_conjuncts = ctx->not_single_slot_filter_conjuncts; |
| _slot_id_to_filter_conjuncts = ctx->slot_id_to_filter_conjuncts; |
| _filter_groups = ctx->filter_groups; |
| _table_info_node_ptr = base_ctx->table_info_node; |
| _column_ids = base_ctx->column_ids; |
| _filter_column_ids = base_ctx->filter_column_ids; |
| |
| // _open_file_reader (called by init_reader NVI before hooks) must have opened the file. |
| DCHECK(_file_metadata != nullptr) |
| << "ParquetReader::_do_init_reader called without _open_file_reader"; |
| _t_metadata = &(_file_metadata->to_thrift()); |
| |
| SCOPED_RAW_TIMER(&_reader_statistics.parse_meta_time); |
| _total_groups = _t_metadata->row_groups.size(); |
| if (_total_groups == 0) { |
| return Status::EndOfFile("init reader failed, empty parquet file: " + _scan_range.path); |
| } |
| _current_row_group_index = RowGroupReader::RowGroupIndex {-1, 0, 0}; |
| |
| // 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. |
| if (has_column_descs()) { |
| _fill_missing_cols.clear(); |
| _fill_missing_defaults.clear(); |
| for (const auto& col_name : base_ctx->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 in file-schema order. |
| // _init_read_columns handles both normal path (missing cols populated above) |
| // and standalone path (_fill_missing_cols empty, _table_info_node_ptr may be null). |
| _init_read_columns(base_ctx->column_names); |
| |
| // build column predicates for column lazy read |
| if (ctx->conjuncts != nullptr) { |
| _lazy_read_ctx.conjuncts = *ctx->conjuncts; |
| } |
| if (ctx->slot_id_to_predicates != nullptr) { |
| _lazy_read_ctx.slot_id_to_predicates = *ctx->slot_id_to_predicates; |
| } |
| |
| // ---- 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_columns; |
| _collect_predicate_columns_from_conjuncts(predicate_columns); |
| |
| // 2. Classify read/partition/missing/synthesized columns into lazy vs predicate groups |
| _classify_columns_for_lazy_read(predicate_columns, _fill_partition_values, |
| _fill_missing_defaults); |
| |
| // 3. Populate col_names vectors for ColumnProcessor path |
| for (auto& kv : _lazy_read_ctx.predicate_partition_columns) { |
| _lazy_read_ctx.predicate_partition_col_names.emplace_back(kv.first); |
| } |
| for (auto& kv : _lazy_read_ctx.predicate_missing_columns) { |
| _lazy_read_ctx.predicate_missing_col_names.emplace_back(kv.first); |
| } |
| for (auto& kv : _lazy_read_ctx.partition_columns) { |
| _lazy_read_ctx.partition_col_names.emplace_back(kv.first); |
| } |
| for (auto& kv : _lazy_read_ctx.missing_columns) { |
| _lazy_read_ctx.missing_col_names.emplace_back(kv.first); |
| } |
| |
| if (_filter_groups && (_total_groups == 0 || _t_metadata->num_rows == 0 || _range_size < 0)) { |
| return Status::EndOfFile("No row group to read"); |
| } |
| |
| return Status::OK(); |
| } |
| |
| void ParquetReader::_init_read_columns(const std::vector<std::string>& column_names) { |
| // Build file_col_name → table_col_name map, skipping missing columns. |
| // Must iterate file schema in physical order so that _generate_random_access_ranges |
| // sees monotonically increasing chunk offsets. |
| auto schema_desc = _file_metadata->schema(); |
| std::map<std::string, std::string> required_file_columns; |
| for (const auto& col_name : 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); |
| } |
| required_file_columns[file_col] = col_name; |
| } |
| for (int i = 0; i < schema_desc.size(); ++i) { |
| const auto& name = schema_desc.get_column(i)->name; |
| if (required_file_columns.contains(name)) { |
| _read_file_columns.emplace_back(name); |
| _read_table_columns.emplace_back(required_file_columns[name]); |
| _read_table_columns_set.insert(required_file_columns[name]); |
| } |
| } |
| } |
| |
| bool ParquetReader::_exists_in_file(const std::string& expr_name) const { |
| // `_read_table_columns_set` is used to ensure that only columns actually read are subject to min-max filtering. |
| // This primarily handles cases where partition columns also exist in a file. The reason it's not modified |
| // in `_table_info_node_ptr` is that Iceberg、Hudi has inconsistent requirements for this node; |
| // Iceberg partition evolution need read partition columns from a file. |
| // hudi set `hoodie.datasource.write.drop.partition.columns=false` not need read partition columns from a file. |
| return _table_info_node_ptr->children_column_exists(expr_name) && |
| _read_table_columns_set.contains(expr_name); |
| } |
| |
| bool ParquetReader::_type_matches(const int cid) const { |
| auto* slot = _tuple_descriptor->slots()[cid]; |
| auto table_col_type = remove_nullable(slot->type()); |
| |
| const auto& file_col_name = _table_info_node_ptr->children_file_column_name(slot->col_name()); |
| const auto& file_col_type = |
| remove_nullable(_file_metadata->schema().get_column(file_col_name)->data_type); |
| |
| return (table_col_type->get_primitive_type() == file_col_type->get_primitive_type()) && |
| !is_complex_type(table_col_type->get_primitive_type()); |
| } |
| |
| void ParquetReader::_collect_predicate_columns_from_conjuncts( |
| std::unordered_map<std::string, std::pair<uint32_t, int>>& predicate_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_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()) { |
| RuntimeFilterExpr* runtime_filter = assert_cast<RuntimeFilterExpr*>(expr.get()); |
| auto filter_impl = runtime_filter->get_impl(); |
| visit_slot(filter_impl.get()); |
| } else { |
| visit_slot(expr.get()); |
| } |
| } |
| |
| if (!_lazy_read_ctx.slot_id_to_predicates.empty()) { |
| auto and_pred = AndBlockColumnPredicate::create_unique(); |
| for (const auto& entry : _lazy_read_ctx.slot_id_to_predicates) { |
| for (const auto& pred : entry.second) { |
| // Parquet shares _push_down_predicates for row-group/page min-max pruning and |
| // bloom-filter evaluation, so this flag currently gates both predicate paths. |
| if (!has_column_optimization(pred->col_name(), ColumnOptimizationTypes::MIN_MAX)) { |
| continue; |
| } |
| if (!_exists_in_file(pred->col_name()) || !_type_matches(pred->column_id())) { |
| continue; |
| } |
| and_pred->add_column_predicate( |
| SingleColumnBlockPredicate::create_unique(pred->clone(pred->column_id()))); |
| } |
| } |
| if (and_pred->num_of_column_predicate() > 0) { |
| _push_down_predicates.push_back(std::move(and_pred)); |
| } |
| } |
| } |
| |
| void ParquetReader::_classify_columns_for_lazy_read( |
| const std::unordered_map<std::string, std::pair<uint32_t, int>>& |
| predicate_conjuncts_columns, |
| const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>& |
| partition_columns, |
| const std::unordered_map<std::string, VExprContextSPtr>& missing_columns) { |
| const FieldDescriptor& schema = _file_metadata->schema(); |
| auto predicate_columns = predicate_conjuncts_columns; |
| #ifndef BE_TEST |
| for (const auto& [col_name, _] : _generated_col_handlers) { |
| int slot_id = -1; |
| for (auto slot : _tuple_descriptor->slots()) { |
| if (slot->col_name() == col_name) { |
| slot_id = slot->id(); |
| break; |
| } |
| } |
| DCHECK(slot_id != -1) << "slot id should not be -1 for generated column: " << col_name; |
| auto column_index = _row_descriptor->get_column_id(slot_id); |
| if (column_index == 0) { |
| _lazy_read_ctx.resize_first_column = false; |
| } |
| // assume generated columns are only used for predicate push down. |
| predicate_columns.emplace(col_name, std::make_pair(column_index, slot_id)); |
| } |
| |
| for (const auto& [col_name, _] : _synthesized_col_handlers) { |
| int slot_id = -1; |
| for (auto slot : _tuple_descriptor->slots()) { |
| if (slot->col_name() == col_name) { |
| slot_id = slot->id(); |
| break; |
| } |
| } |
| DCHECK(slot_id != -1) << "slot id should not be -1 for synthesized column: " << col_name; |
| auto column_index = _row_descriptor->get_column_id(slot_id); |
| if (column_index == 0) { |
| _lazy_read_ctx.resize_first_column = false; |
| } |
| // synthesized columns always fill data on first phase. |
| _lazy_read_ctx.all_predicate_col_ids.emplace_back(column_index); |
| } |
| #endif |
| for (auto& read_table_col : _read_table_columns) { |
| _lazy_read_ctx.all_read_columns.emplace_back(read_table_col); |
| |
| auto file_column_name = _table_info_node_ptr->children_file_column_name(read_table_col); |
| PrimitiveType column_type = |
| schema.get_column(file_column_name)->data_type->get_primitive_type(); |
| if (is_complex_type(column_type)) { |
| _lazy_read_ctx.has_complex_type = true; |
| } |
| if (predicate_columns.size() > 0) { |
| auto iter = predicate_columns.find(read_table_col); |
| if (iter == predicate_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.all_predicate_col_ids.emplace_back(iter->second.first); |
| } |
| } |
| } |
| |
| for (auto& kv : partition_columns) { |
| auto iter = predicate_columns.find(kv.first); |
| if (iter == predicate_columns.end()) { |
| _lazy_read_ctx.partition_columns.emplace(kv.first, kv.second); |
| } else { |
| _lazy_read_ctx.predicate_partition_columns.emplace(kv.first, kv.second); |
| _lazy_read_ctx.all_predicate_col_ids.emplace_back(iter->second.first); |
| } |
| } |
| |
| for (auto& kv : missing_columns) { |
| auto iter = predicate_columns.find(kv.first); |
| if (iter != predicate_columns.end()) { |
| //For check missing column : missing column == xx, missing column is null,missing column is not null. |
| if (_slot_id_to_filter_conjuncts->find(iter->second.second) != |
| _slot_id_to_filter_conjuncts->end()) { |
| for (auto& ctx : _slot_id_to_filter_conjuncts->find(iter->second.second)->second) { |
| _lazy_read_ctx.missing_columns_conjuncts.emplace_back(ctx); |
| } |
| } |
| _lazy_read_ctx.predicate_missing_columns.emplace(kv.first, kv.second); |
| _lazy_read_ctx.all_predicate_col_ids.emplace_back(iter->second.first); |
| } else { |
| _lazy_read_ctx.missing_columns.emplace(kv.first, kv.second); |
| } |
| } |
| |
| if (_enable_lazy_mat && _lazy_read_ctx.predicate_columns.first.size() > 0 && |
| _lazy_read_ctx.lazy_read_columns.size() > 0) { |
| _lazy_read_ctx.can_lazy_read = true; |
| } |
| |
| 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); |
| } |
| } |
| } |
| |
| // init file reader and file metadata for parsing schema |
| Status ParquetReader::init_schema_reader() { |
| RETURN_IF_ERROR(_open_file()); |
| _t_metadata = &(_file_metadata->to_thrift()); |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::get_parsed_schema(std::vector<std::string>* col_names, |
| std::vector<DataTypePtr>* col_types) { |
| _total_groups = _t_metadata->row_groups.size(); |
| auto schema_desc = _file_metadata->schema(); |
| for (int i = 0; i < schema_desc.size(); ++i) { |
| // Get the Column Reader for the boolean column |
| col_names->emplace_back(schema_desc.get_column(i)->name); |
| col_types->emplace_back(make_nullable(schema_desc.get_column(i)->data_type)); |
| } |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::_get_columns_impl( |
| std::unordered_map<std::string, DataTypePtr>* name_to_type) { |
| const auto& schema_desc = _file_metadata->schema(); |
| std::unordered_set<std::string> column_names; |
| schema_desc.get_column_names(&column_names); |
| for (auto& name : column_names) { |
| auto field = schema_desc.get_column(name); |
| name_to_type->emplace(name, field->data_type); |
| } |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::_do_get_next_block(Block* block, size_t* read_rows, bool* eof) { |
| if (_current_group_reader == nullptr || _row_group_eof) { |
| Status st = _next_row_group_reader(); |
| if (!st.ok() && !st.is<ErrorCode::END_OF_FILE>()) { |
| return st; |
| } |
| if (_current_group_reader == nullptr || _row_group_eof || st.is<ErrorCode::END_OF_FILE>()) { |
| _current_group_reader.reset(nullptr); |
| _row_group_eof = true; |
| *read_rows = 0; |
| *eof = true; |
| return Status::OK(); |
| } |
| } |
| |
| // Limit memory per batch for load paths. |
| // _load_bytes_per_row is updated after each batch so the *next* call pre-shrinks _batch_size |
| // before reading, ensuring the current batch is already within the limit (from call 2 onward). |
| const int64_t max_block_bytes = |
| (_state != nullptr && _state->query_type() == TQueryType::LOAD && |
| config::load_reader_max_block_bytes > 0) |
| ? config::load_reader_max_block_bytes |
| : 0; |
| if (max_block_bytes > 0 && _load_bytes_per_row > 0) { |
| _batch_size = std::max((size_t)1, |
| (size_t)((int64_t)max_block_bytes / (int64_t)_load_bytes_per_row)); |
| } |
| |
| SCOPED_RAW_TIMER(&_reader_statistics.column_read_time); |
| Status batch_st = |
| _current_group_reader->next_batch(block, _batch_size, read_rows, &_row_group_eof); |
| if (batch_st.is<ErrorCode::END_OF_FILE>()) { |
| block->clear_column_data(); |
| _current_group_reader.reset(nullptr); |
| *read_rows = 0; |
| *eof = true; |
| return Status::OK(); |
| } |
| |
| if (!batch_st.ok()) { |
| return Status::InternalError("Read parquet file {} failed, reason = {}", _scan_range.path, |
| batch_st.to_string()); |
| } |
| |
| if (max_block_bytes > 0 && *read_rows > 0) { |
| _load_bytes_per_row = block->bytes() / *read_rows; |
| } |
| |
| if (_row_group_eof) { |
| auto column_st = _current_group_reader->merged_column_statistics(); |
| _column_statistics.merge(column_st); |
| _reader_statistics.lazy_read_filtered_rows += |
| _current_group_reader->lazy_read_filtered_rows(); |
| _reader_statistics.predicate_filter_time += _current_group_reader->predicate_filter_time(); |
| _reader_statistics.dict_filter_rewrite_time += |
| _current_group_reader->dict_filter_rewrite_time(); |
| if (_io_ctx) { |
| _io_ctx->condition_cache_filtered_rows += |
| _current_group_reader->condition_cache_filtered_rows(); |
| } |
| |
| if (_current_row_group_index.row_group_id + 1 == _total_groups) { |
| *eof = true; |
| } else { |
| *eof = false; |
| } |
| } |
| return Status::OK(); |
| } |
| |
| RowGroupReader::PositionDeleteContext ParquetReader::_get_position_delete_ctx( |
| const tparquet::RowGroup& row_group, const RowGroupReader::RowGroupIndex& row_group_index) { |
| if (_delete_rows == nullptr) { |
| return RowGroupReader::PositionDeleteContext(row_group.num_rows, row_group_index.first_row); |
| } |
| const int64_t* delete_rows = &(*_delete_rows)[0]; |
| const int64_t* delete_rows_end = delete_rows + _delete_rows->size(); |
| const int64_t* start_pos = std::lower_bound(delete_rows + _delete_rows_index, delete_rows_end, |
| row_group_index.first_row); |
| int64_t start_index = start_pos - delete_rows; |
| const int64_t* end_pos = std::lower_bound(start_pos, delete_rows_end, row_group_index.last_row); |
| int64_t end_index = end_pos - delete_rows; |
| _delete_rows_index = end_index; |
| return RowGroupReader::PositionDeleteContext(*_delete_rows, row_group.num_rows, |
| row_group_index.first_row, start_index, end_index); |
| } |
| |
| Status ParquetReader::_next_row_group_reader() { |
| if (_current_group_reader != nullptr) { |
| _current_group_reader->collect_profile_before_close(); |
| } |
| |
| RowRanges candidate_row_ranges; |
| while (++_current_row_group_index.row_group_id < _total_groups) { |
| const auto& row_group = _t_metadata->row_groups[_current_row_group_index.row_group_id]; |
| _current_row_group_index.first_row = _current_row_group_index.last_row; |
| _current_row_group_index.last_row = _current_row_group_index.last_row + row_group.num_rows; |
| |
| if (_filter_groups && _is_misaligned_range_group(row_group)) { |
| continue; |
| } |
| |
| candidate_row_ranges.clear(); |
| // The range of lines to be read is determined by the push down predicate. |
| RETURN_IF_ERROR(_process_min_max_bloom_filter( |
| _current_row_group_index, row_group, _push_down_predicates, &candidate_row_ranges)); |
| |
| std::function<int64_t(const FieldSchema*)> column_compressed_size = |
| [&row_group, &column_compressed_size](const FieldSchema* field) -> int64_t { |
| if (field->physical_column_index >= 0) { |
| int parquet_col_id = field->physical_column_index; |
| if (row_group.columns[parquet_col_id].__isset.meta_data) { |
| return row_group.columns[parquet_col_id].meta_data.total_compressed_size; |
| } |
| return 0; |
| } |
| int64_t size = 0; |
| for (const FieldSchema& child : field->children) { |
| size += column_compressed_size(&child); |
| } |
| return size; |
| }; |
| int64_t group_size = 0; // only calculate the needed columns |
| for (auto& read_col : _read_file_columns) { |
| const FieldSchema* field = _file_metadata->schema().get_column(read_col); |
| group_size += column_compressed_size(field); |
| } |
| |
| _reader_statistics.read_rows += candidate_row_ranges.count(); |
| if (_io_ctx && _io_ctx->file_reader_stats) { |
| _io_ctx->file_reader_stats->read_rows += candidate_row_ranges.count(); |
| } |
| |
| if (candidate_row_ranges.count() != 0) { |
| // need read this row group. |
| _reader_statistics.read_row_groups++; |
| _reader_statistics.filtered_page_rows += |
| row_group.num_rows - candidate_row_ranges.count(); |
| break; |
| } else { |
| // this row group be filtered. |
| _reader_statistics.filtered_row_groups++; |
| _reader_statistics.filtered_bytes += group_size; |
| _reader_statistics.filtered_group_rows += row_group.num_rows; |
| } |
| } |
| |
| if (_current_row_group_index.row_group_id == _total_groups) { |
| _row_group_eof = true; |
| _current_group_reader.reset(nullptr); |
| return Status::EndOfFile("No next RowGroupReader"); |
| } |
| |
| // process page index and generate the ranges to read |
| auto& row_group = _t_metadata->row_groups[_current_row_group_index.row_group_id]; |
| |
| RowGroupReader::PositionDeleteContext position_delete_ctx = |
| _get_position_delete_ctx(row_group, _current_row_group_index); |
| io::FileReaderSPtr group_file_reader; |
| if (typeid_cast<io::InMemoryFileReader*>(_file_reader.get())) { |
| // InMemoryFileReader has the ability to merge small IO |
| group_file_reader = _file_reader; |
| } else { |
| size_t avg_io_size = 0; |
| const std::vector<io::PrefetchRange> io_ranges = |
| _generate_random_access_ranges(_current_row_group_index, &avg_io_size); |
| int64_t merged_read_slice_size = -1; |
| if (_state != nullptr && _state->query_options().__isset.merge_read_slice_size) { |
| merged_read_slice_size = _state->query_options().merge_read_slice_size; |
| } |
| // The underlying page reader will prefetch data in column. |
| // Using both MergeRangeFileReader and BufferedStreamReader simultaneously would waste a lot of memory. |
| group_file_reader = |
| avg_io_size < io::MergeRangeFileReader::SMALL_IO |
| ? std::make_shared<io::MergeRangeFileReader>( |
| _profile, _file_reader, io_ranges, merged_read_slice_size) |
| : _file_reader; |
| } |
| _current_group_reader.reset(new RowGroupReader( |
| _io_ctx && _io_ctx->file_reader_stats |
| ? std::make_shared<io::TracingFileReader>(group_file_reader, |
| _io_ctx->file_reader_stats) |
| : group_file_reader, |
| _read_table_columns, _current_row_group_index.row_group_id, row_group, _ctz, _io_ctx, |
| position_delete_ctx, _lazy_read_ctx, _state, _column_ids, _filter_column_ids)); |
| _row_group_eof = false; |
| |
| _current_group_reader->set_current_row_group_idx(_current_row_group_index); |
| _current_group_reader->set_col_name_to_block_idx(_col_name_to_block_idx); |
| if (_condition_cache_ctx) { |
| _current_group_reader->set_condition_cache_context(_condition_cache_ctx); |
| } |
| _current_group_reader->set_table_format_reader(this); |
| |
| _current_group_reader->_table_info_node_ptr = _table_info_node_ptr; |
| return _current_group_reader->init(_file_metadata->schema(), candidate_row_ranges, _col_offsets, |
| _tuple_descriptor, _row_descriptor, _colname_to_slot_id, |
| _not_single_slot_filter_conjuncts, |
| _slot_id_to_filter_conjuncts); |
| } |
| |
| std::vector<io::PrefetchRange> ParquetReader::_generate_random_access_ranges( |
| const RowGroupReader::RowGroupIndex& group, size_t* avg_io_size) { |
| std::vector<io::PrefetchRange> result; |
| int64_t last_chunk_end = -1; |
| size_t total_io_size = 0; |
| std::function<void(const FieldSchema*, const tparquet::RowGroup&)> scalar_range = |
| [&](const FieldSchema* field, const tparquet::RowGroup& row_group) { |
| if (_column_ids.empty() || |
| _column_ids.find(field->get_column_id()) != _column_ids.end()) { |
| if (field->data_type->get_primitive_type() == TYPE_ARRAY) { |
| scalar_range(&field->children[0], row_group); |
| } else if (field->data_type->get_primitive_type() == TYPE_MAP) { |
| scalar_range(&field->children[0], row_group); |
| scalar_range(&field->children[1], row_group); |
| } else if (field->data_type->get_primitive_type() == TYPE_STRUCT) { |
| for (int i = 0; i < field->children.size(); ++i) { |
| scalar_range(&field->children[i], row_group); |
| } |
| } else { |
| const tparquet::ColumnChunk& chunk = |
| row_group.columns[field->physical_column_index]; |
| auto& chunk_meta = chunk.meta_data; |
| int64_t chunk_start = has_dict_page(chunk_meta) |
| ? chunk_meta.dictionary_page_offset |
| : chunk_meta.data_page_offset; |
| int64_t chunk_end = chunk_start + chunk_meta.total_compressed_size; |
| DCHECK_GE(chunk_start, last_chunk_end); |
| result.emplace_back(chunk_start, chunk_end); |
| total_io_size += chunk_meta.total_compressed_size; |
| last_chunk_end = chunk_end; |
| } |
| } |
| }; |
| const tparquet::RowGroup& row_group = _t_metadata->row_groups[group.row_group_id]; |
| for (const auto& read_col : _read_file_columns) { |
| const FieldSchema* field = _file_metadata->schema().get_column(read_col); |
| scalar_range(field, row_group); |
| } |
| if (!result.empty()) { |
| *avg_io_size = total_io_size / result.size(); |
| } |
| return result; |
| } |
| |
| bool ParquetReader::_is_misaligned_range_group(const tparquet::RowGroup& row_group) const { |
| int64_t start_offset = _get_column_start_offset(row_group.columns[0].meta_data); |
| |
| auto& last_column = row_group.columns[row_group.columns.size() - 1].meta_data; |
| int64_t end_offset = _get_column_start_offset(last_column) + last_column.total_compressed_size; |
| |
| int64_t row_group_mid = start_offset + (end_offset - start_offset) / 2; |
| if (!(row_group_mid >= _range_start_offset && |
| row_group_mid < _range_start_offset + _range_size)) { |
| return true; |
| } |
| return false; |
| } |
| |
| int64_t ParquetReader::get_total_rows() const { |
| if (!_t_metadata) return 0; |
| if (!_filter_groups) return _t_metadata->num_rows; |
| int64_t total = 0; |
| for (const auto& rg : _t_metadata->row_groups) { |
| if (!_is_misaligned_range_group(rg)) { |
| total += rg.num_rows; |
| } |
| } |
| return total; |
| } |
| |
| void ParquetReader::set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) { |
| _condition_cache_ctx = std::move(ctx); |
| if (!_condition_cache_ctx || !_t_metadata || !_filter_groups) { |
| return; |
| } |
| // Find the first assigned row group to compute base_granule. |
| int64_t first_row = 0; |
| for (const auto& rg : _t_metadata->row_groups) { |
| if (!_is_misaligned_range_group(rg)) { |
| _condition_cache_ctx->base_granule = first_row / ConditionCacheContext::GRANULE_SIZE; |
| return; |
| } |
| first_row += rg.num_rows; |
| } |
| } |
| |
| Status ParquetReader::_process_page_index_filter( |
| const tparquet::RowGroup& row_group, const RowGroupReader::RowGroupIndex& row_group_index, |
| const std::vector<std::unique_ptr<MutilColumnBlockPredicate>>& push_down_pred, |
| RowRanges* candidate_row_ranges) { |
| if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) { |
| return Status::EndOfFile("stop"); |
| } |
| |
| std::function<void()> read_whole_row_group = [&]() { |
| candidate_row_ranges->add(RowRange {0, row_group.num_rows}); |
| }; |
| |
| // Check if the page index is available and if it exists. |
| PageIndex page_index; |
| if (!config::enable_parquet_page_index || _colname_to_slot_id == nullptr || |
| !page_index.check_and_get_page_index_ranges(row_group.columns)) { |
| read_whole_row_group(); |
| return Status::OK(); |
| } |
| |
| std::vector<int> parquet_col_ids; |
| for (size_t idx = 0; idx < _read_table_columns.size(); idx++) { |
| const auto& read_table_col = _read_table_columns[idx]; |
| const auto& read_file_col = _read_file_columns[idx]; |
| if (!_colname_to_slot_id->contains(read_table_col)) { |
| continue; |
| } |
| auto* field = _file_metadata->schema().get_column(read_file_col); |
| |
| std::function<void(FieldSchema * field)> f = [&](FieldSchema* field) { |
| if (!_column_ids.empty() && |
| _column_ids.find(field->get_column_id()) == _column_ids.end()) { |
| return; |
| } |
| |
| if (field->data_type->get_primitive_type() == TYPE_ARRAY) { |
| f(&field->children[0]); |
| } else if (field->data_type->get_primitive_type() == TYPE_MAP) { |
| f(&field->children[0]); |
| f(&field->children[1]); |
| } else if (field->data_type->get_primitive_type() == TYPE_STRUCT) { |
| for (int i = 0; i < field->children.size(); ++i) { |
| f(&field->children[i]); |
| } |
| } else { |
| int parquet_col_id = field->physical_column_index; |
| if (parquet_col_id >= 0) { |
| parquet_col_ids.push_back(parquet_col_id); |
| } |
| } |
| }; |
| |
| f(field); |
| } |
| |
| auto parse_offset_index = [&]() -> Status { |
| std::vector<uint8_t> off_index_buff(page_index._offset_index_size); |
| Slice res(off_index_buff.data(), page_index._offset_index_size); |
| size_t bytes_read = 0; |
| { |
| SCOPED_RAW_TIMER(&_reader_statistics.read_page_index_time); |
| RETURN_IF_ERROR(_tracing_file_reader->read_at(page_index._offset_index_start, res, |
| &bytes_read, _io_ctx)); |
| } |
| _column_statistics.page_index_read_calls++; |
| _col_offsets.clear(); |
| |
| for (auto parquet_col_id : parquet_col_ids) { |
| auto& chunk = row_group.columns[parquet_col_id]; |
| if (chunk.offset_index_length == 0) [[unlikely]] { |
| continue; |
| } |
| tparquet::OffsetIndex offset_index; |
| SCOPED_RAW_TIMER(&_reader_statistics.parse_page_index_time); |
| RETURN_IF_ERROR( |
| page_index.parse_offset_index(chunk, off_index_buff.data(), &offset_index)); |
| _col_offsets[parquet_col_id] = offset_index; |
| } |
| return Status::OK(); |
| }; |
| |
| // from https://github.com/apache/doris/pull/55795 |
| RETURN_IF_ERROR(parse_offset_index()); |
| |
| // Check if page index is needed for min-max filter. |
| if (!_enable_filter_by_min_max || push_down_pred.empty()) { |
| read_whole_row_group(); |
| return Status::OK(); |
| } |
| |
| // read column index. |
| std::vector<uint8_t> col_index_buff(page_index._column_index_size); |
| size_t bytes_read = 0; |
| Slice result(col_index_buff.data(), page_index._column_index_size); |
| { |
| SCOPED_RAW_TIMER(&_reader_statistics.read_page_index_time); |
| RETURN_IF_ERROR(_tracing_file_reader->read_at(page_index._column_index_start, result, |
| &bytes_read, _io_ctx)); |
| } |
| _column_statistics.page_index_read_calls++; |
| |
| SCOPED_RAW_TIMER(&_reader_statistics.page_index_filter_time); |
| |
| // Construct a cacheable page index structure to avoid repeatedly reading the page index of the same column. |
| ParquetPredicate::CachedPageIndexStat cached_page_index; |
| cached_page_index.ctz = _ctz; |
| std::function<bool(ParquetPredicate::PageIndexStat**, int)> get_stat_func = |
| [&](ParquetPredicate::PageIndexStat** ans, const int cid) -> bool { |
| if (cached_page_index.stats.contains(cid)) { |
| *ans = &cached_page_index.stats[cid]; |
| return (*ans)->available; |
| } |
| cached_page_index.stats.emplace(cid, ParquetPredicate::PageIndexStat {}); |
| auto& sig_stat = cached_page_index.stats[cid]; |
| |
| auto* slot = _tuple_descriptor->slots()[cid]; |
| if (!_table_info_node_ptr->children_column_exists(slot->col_name())) { |
| // table column not exist in file, may be schema change. |
| return false; |
| } |
| |
| const auto& file_col_name = |
| _table_info_node_ptr->children_file_column_name(slot->col_name()); |
| const FieldSchema* col_schema = _file_metadata->schema().get_column(file_col_name); |
| int parquet_col_id = col_schema->physical_column_index; |
| |
| if (parquet_col_id < 0) { |
| // complex type, not support page index yet. |
| return false; |
| } |
| if (!_col_offsets.contains(parquet_col_id)) { |
| // If the file contains partition columns and the query applies filters on those |
| // partition columns, then reading the page index is unnecessary. |
| return false; |
| } |
| |
| auto& column_chunk = row_group.columns[parquet_col_id]; |
| if (column_chunk.column_index_length == 0 || column_chunk.offset_index_length == 0) { |
| // column no page index. |
| return false; |
| } |
| |
| tparquet::ColumnIndex column_index; |
| { |
| SCOPED_RAW_TIMER(&_reader_statistics.parse_page_index_time); |
| RETURN_IF_ERROR(page_index.parse_column_index(column_chunk, col_index_buff.data(), |
| &column_index)); |
| } |
| const int64_t num_of_pages = column_index.null_pages.size(); |
| if (num_of_pages <= 0) [[unlikely]] { |
| // no page. (maybe this row group no data.) |
| return false; |
| } |
| DCHECK_EQ(column_index.min_values.size(), column_index.max_values.size()); |
| if (!column_index.__isset.null_counts) { |
| // not set null or null counts; |
| return false; |
| } |
| |
| auto& offset_index = _col_offsets[parquet_col_id]; |
| const auto& page_locations = offset_index.page_locations; |
| |
| sig_stat.col_schema = col_schema; |
| sig_stat.num_of_pages = num_of_pages; |
| sig_stat.encoded_min_value = column_index.min_values; |
| sig_stat.encoded_max_value = column_index.max_values; |
| sig_stat.is_all_null.resize(num_of_pages); |
| sig_stat.has_null.resize(num_of_pages); |
| sig_stat.ranges.resize(num_of_pages); |
| |
| for (int page_id = 0; page_id < num_of_pages; page_id++) { |
| sig_stat.is_all_null[page_id] = column_index.null_pages[page_id]; |
| sig_stat.has_null[page_id] = column_index.null_counts[page_id] > 0; |
| |
| int64_t from = page_locations[page_id].first_row_index; |
| int64_t to = 0; |
| if (page_id == page_locations.size() - 1) { |
| to = row_group_index.last_row; |
| } else { |
| to = page_locations[page_id + 1].first_row_index; |
| } |
| sig_stat.ranges[page_id] = RowRange {from, to}; |
| } |
| |
| sig_stat.available = true; |
| *ans = &sig_stat; |
| return true; |
| }; |
| cached_page_index.row_group_range = {0, row_group.num_rows}; |
| cached_page_index.get_stat_func = get_stat_func; |
| |
| candidate_row_ranges->add({0, row_group.num_rows}); |
| for (const auto& predicate : push_down_pred) { |
| RowRanges tmp_row_range; |
| if (!predicate->evaluate_and(&cached_page_index, &tmp_row_range)) { |
| // no need read this row group. |
| candidate_row_ranges->clear(); |
| return Status::OK(); |
| } |
| RowRanges::ranges_intersection(*candidate_row_ranges, tmp_row_range, candidate_row_ranges); |
| } |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::_process_min_max_bloom_filter( |
| const RowGroupReader::RowGroupIndex& row_group_index, const tparquet::RowGroup& row_group, |
| const std::vector<std::unique_ptr<MutilColumnBlockPredicate>>& push_down_pred, |
| RowRanges* row_ranges) { |
| SCOPED_RAW_TIMER(&_reader_statistics.row_group_filter_time); |
| if (!_filter_groups) { |
| // No row group filtering is needed; |
| // for example, Iceberg reads position delete files. |
| row_ranges->add({0, row_group.num_rows}); |
| return Status::OK(); |
| } |
| |
| if (_read_by_rows) { |
| auto group_start = row_group_index.first_row; |
| auto group_end = row_group_index.last_row; |
| |
| while (!_row_ids.empty()) { |
| auto v = _row_ids.front(); |
| if (v < group_start) { |
| continue; |
| } else if (v < group_end) { |
| row_ranges->add(RowRange {v - group_start, v - group_start + 1}); |
| _row_ids.pop_front(); |
| } else { |
| break; |
| } |
| } |
| } else { |
| bool filter_this_row_group = false; |
| bool filtered_by_min_max = false; |
| bool filtered_by_bloom_filter = false; |
| RETURN_IF_ERROR(_process_column_stat_filter(row_group, push_down_pred, |
| &filter_this_row_group, &filtered_by_min_max, |
| &filtered_by_bloom_filter)); |
| // Update statistics based on filter type |
| if (filter_this_row_group) { |
| if (filtered_by_min_max) { |
| _reader_statistics.filtered_row_groups_by_min_max++; |
| } |
| if (filtered_by_bloom_filter) { |
| _reader_statistics.filtered_row_groups_by_bloom_filter++; |
| } |
| } |
| |
| if (!filter_this_row_group) { |
| RETURN_IF_ERROR(_process_page_index_filter(row_group, row_group_index, push_down_pred, |
| row_ranges)); |
| } |
| } |
| |
| return Status::OK(); |
| } |
| |
| Status ParquetReader::_process_column_stat_filter( |
| const tparquet::RowGroup& row_group, |
| const std::vector<std::unique_ptr<MutilColumnBlockPredicate>>& push_down_pred, |
| bool* filter_group, bool* filtered_by_min_max, bool* filtered_by_bloom_filter) { |
| // If both filters are disabled, skip filtering |
| if (!_enable_filter_by_min_max && !_enable_filter_by_bloom_filter) { |
| return Status::OK(); |
| } |
| |
| // Cache bloom filters for each column to avoid reading the same bloom filter multiple times |
| // when there are multiple predicates on the same column |
| std::unordered_map<int, std::unique_ptr<ParquetBlockSplitBloomFilter>> bloom_filter_cache; |
| |
| // Initialize output parameters |
| *filtered_by_min_max = false; |
| *filtered_by_bloom_filter = false; |
| |
| for (const auto& predicate : _push_down_predicates) { |
| std::function<bool(ParquetPredicate::ColumnStat*, int)> get_stat_func = |
| [&](ParquetPredicate::ColumnStat* stat, const int cid) { |
| // Check if min-max filter is enabled |
| if (!_enable_filter_by_min_max) { |
| return false; |
| } |
| auto* slot = _tuple_descriptor->slots()[cid]; |
| if (!_table_info_node_ptr->children_column_exists(slot->col_name())) { |
| return false; |
| } |
| const auto& file_col_name = |
| _table_info_node_ptr->children_file_column_name(slot->col_name()); |
| const FieldSchema* col_schema = |
| _file_metadata->schema().get_column(file_col_name); |
| int parquet_col_id = col_schema->physical_column_index; |
| auto meta_data = row_group.columns[parquet_col_id].meta_data; |
| stat->col_schema = col_schema; |
| return ParquetPredicate::read_column_stats(col_schema, meta_data, |
| &_ignored_stats, |
| _t_metadata->created_by, stat) |
| .ok(); |
| }; |
| std::function<bool(ParquetPredicate::ColumnStat*, int)> get_bloom_filter_func = |
| [&](ParquetPredicate::ColumnStat* stat, const int cid) { |
| auto* slot = _tuple_descriptor->slots()[cid]; |
| if (!_table_info_node_ptr->children_column_exists(slot->col_name())) { |
| return false; |
| } |
| const auto& file_col_name = |
| _table_info_node_ptr->children_file_column_name(slot->col_name()); |
| const FieldSchema* col_schema = |
| _file_metadata->schema().get_column(file_col_name); |
| int parquet_col_id = col_schema->physical_column_index; |
| auto meta_data = row_group.columns[parquet_col_id].meta_data; |
| if (!meta_data.__isset.bloom_filter_offset) { |
| return false; |
| } |
| auto primitive_type = |
| remove_nullable(col_schema->data_type)->get_primitive_type(); |
| if (!ParquetPredicate::bloom_filter_supported(primitive_type)) { |
| return false; |
| } |
| |
| // Check if bloom filter is enabled |
| if (!_enable_filter_by_bloom_filter) { |
| return false; |
| } |
| |
| // Check cache first |
| auto cache_iter = bloom_filter_cache.find(parquet_col_id); |
| if (cache_iter != bloom_filter_cache.end()) { |
| // Bloom filter already loaded for this column, reuse it |
| stat->bloom_filter = std::move(cache_iter->second); |
| bloom_filter_cache.erase(cache_iter); |
| return stat->bloom_filter != nullptr; |
| } |
| |
| if (!stat->bloom_filter) { |
| SCOPED_RAW_TIMER(&_reader_statistics.bloom_filter_read_time); |
| auto st = ParquetPredicate::read_bloom_filter( |
| meta_data, _tracing_file_reader, _io_ctx, stat); |
| if (!st.ok()) { |
| LOG(WARNING) << "Failed to read bloom filter for column " |
| << col_schema->name << " in file " << _scan_range.path |
| << ", status: " << st.to_string(); |
| stat->bloom_filter.reset(); |
| return false; |
| } |
| } |
| return stat->bloom_filter != nullptr; |
| }; |
| ParquetPredicate::ColumnStat stat; |
| stat.ctz = _ctz; |
| stat.get_stat_func = &get_stat_func; |
| stat.get_bloom_filter_func = &get_bloom_filter_func; |
| |
| if (!predicate->evaluate_and(&stat)) { |
| *filter_group = true; |
| |
| // Track which filter was used for filtering |
| // If bloom filter was loaded, it means bloom filter was used |
| if (stat.bloom_filter) { |
| *filtered_by_bloom_filter = true; |
| } |
| // If col_schema was set but no bloom filter, it means min-max stats were used |
| if (stat.col_schema && !stat.bloom_filter) { |
| *filtered_by_min_max = true; |
| } |
| |
| return Status::OK(); |
| } |
| |
| // After evaluating, if the bloom filter was used, cache it for subsequent predicates |
| if (stat.bloom_filter) { |
| // Find the column id for caching |
| for (auto* slot : _tuple_descriptor->slots()) { |
| if (_table_info_node_ptr->children_column_exists(slot->col_name())) { |
| const auto& file_col_name = |
| _table_info_node_ptr->children_file_column_name(slot->col_name()); |
| const FieldSchema* col_schema = |
| _file_metadata->schema().get_column(file_col_name); |
| int parquet_col_id = col_schema->physical_column_index; |
| if (stat.col_schema == col_schema) { |
| bloom_filter_cache[parquet_col_id] = std::move(stat.bloom_filter); |
| break; |
| } |
| } |
| } |
| } |
| } |
| |
| // Update filter statistics if this row group was not filtered |
| // The statistics will be updated in _init_row_groups when filter_group is true |
| return Status::OK(); |
| } |
| |
| int64_t ParquetReader::_get_column_start_offset(const tparquet::ColumnMetaData& column) const { |
| return has_dict_page(column) ? column.dictionary_page_offset : column.data_page_offset; |
| } |
| |
| void ParquetReader::_collect_profile() { |
| if (_profile == nullptr) { |
| return; |
| } |
| |
| if (_current_group_reader != nullptr) { |
| _current_group_reader->collect_profile_before_close(); |
| } |
| COUNTER_UPDATE(_parquet_profile.filtered_row_groups, _reader_statistics.filtered_row_groups); |
| COUNTER_UPDATE(_parquet_profile.filtered_row_groups_by_min_max, |
| _reader_statistics.filtered_row_groups_by_min_max); |
| COUNTER_UPDATE(_parquet_profile.filtered_row_groups_by_bloom_filter, |
| _reader_statistics.filtered_row_groups_by_bloom_filter); |
| COUNTER_UPDATE(_parquet_profile.to_read_row_groups, _reader_statistics.read_row_groups); |
| COUNTER_UPDATE(_parquet_profile.total_row_groups, _total_groups); |
| COUNTER_UPDATE(_parquet_profile.filtered_group_rows, _reader_statistics.filtered_group_rows); |
| COUNTER_UPDATE(_parquet_profile.filtered_page_rows, _reader_statistics.filtered_page_rows); |
| COUNTER_UPDATE(_parquet_profile.lazy_read_filtered_rows, |
| _reader_statistics.lazy_read_filtered_rows); |
| COUNTER_UPDATE(_parquet_profile.filtered_bytes, _reader_statistics.filtered_bytes); |
| COUNTER_UPDATE(_parquet_profile.raw_rows_read, _reader_statistics.read_rows); |
| COUNTER_UPDATE(_parquet_profile.column_read_time, _reader_statistics.column_read_time); |
| COUNTER_UPDATE(_parquet_profile.parse_meta_time, _reader_statistics.parse_meta_time); |
| COUNTER_UPDATE(_parquet_profile.parse_footer_time, _reader_statistics.parse_footer_time); |
| COUNTER_UPDATE(_parquet_profile.file_reader_create_time, |
| _reader_statistics.file_reader_create_time); |
| COUNTER_UPDATE(_parquet_profile.open_file_num, _reader_statistics.open_file_num); |
| COUNTER_UPDATE(_parquet_profile.page_index_filter_time, |
| _reader_statistics.page_index_filter_time); |
| COUNTER_UPDATE(_parquet_profile.read_page_index_time, _reader_statistics.read_page_index_time); |
| COUNTER_UPDATE(_parquet_profile.parse_page_index_time, |
| _reader_statistics.parse_page_index_time); |
| COUNTER_UPDATE(_parquet_profile.row_group_filter_time, |
| _reader_statistics.row_group_filter_time); |
| COUNTER_UPDATE(_parquet_profile.file_footer_read_calls, |
| _reader_statistics.file_footer_read_calls); |
| COUNTER_UPDATE(_parquet_profile.file_footer_hit_cache, |
| _reader_statistics.file_footer_hit_cache); |
| |
| COUNTER_UPDATE(_parquet_profile.skip_page_header_num, _column_statistics.skip_page_header_num); |
| COUNTER_UPDATE(_parquet_profile.parse_page_header_num, |
| _column_statistics.parse_page_header_num); |
| COUNTER_UPDATE(_parquet_profile.predicate_filter_time, |
| _reader_statistics.predicate_filter_time); |
| COUNTER_UPDATE(_parquet_profile.dict_filter_rewrite_time, |
| _reader_statistics.dict_filter_rewrite_time); |
| COUNTER_UPDATE(_parquet_profile.convert_time, _column_statistics.convert_time); |
| COUNTER_UPDATE(_parquet_profile.bloom_filter_read_time, |
| _reader_statistics.bloom_filter_read_time); |
| COUNTER_UPDATE(_parquet_profile.page_index_read_calls, |
| _column_statistics.page_index_read_calls); |
| COUNTER_UPDATE(_parquet_profile.decompress_time, _column_statistics.decompress_time); |
| COUNTER_UPDATE(_parquet_profile.decompress_cnt, _column_statistics.decompress_cnt); |
| COUNTER_UPDATE(_parquet_profile.page_read_counter, _column_statistics.page_read_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_write_counter, |
| _column_statistics.page_cache_write_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_compressed_write_counter, |
| _column_statistics.page_cache_compressed_write_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_decompressed_write_counter, |
| _column_statistics.page_cache_decompressed_write_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_hit_counter, |
| _column_statistics.page_cache_hit_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_missing_counter, |
| _column_statistics.page_cache_missing_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_compressed_hit_counter, |
| _column_statistics.page_cache_compressed_hit_counter); |
| COUNTER_UPDATE(_parquet_profile.page_cache_decompressed_hit_counter, |
| _column_statistics.page_cache_decompressed_hit_counter); |
| COUNTER_UPDATE(_parquet_profile.decode_header_time, _column_statistics.decode_header_time); |
| COUNTER_UPDATE(_parquet_profile.read_page_header_time, |
| _column_statistics.read_page_header_time); |
| COUNTER_UPDATE(_parquet_profile.decode_value_time, _column_statistics.decode_value_time); |
| COUNTER_UPDATE(_parquet_profile.decode_dict_time, _column_statistics.decode_dict_time); |
| COUNTER_UPDATE(_parquet_profile.decode_level_time, _column_statistics.decode_level_time); |
| COUNTER_UPDATE(_parquet_profile.decode_null_map_time, _column_statistics.decode_null_map_time); |
| } |
| |
| void ParquetReader::_collect_profile_before_close() { |
| _collect_profile(); |
| } |
| |
| } // namespace doris |
