github-actions[bot] commented on code in PR #65674: URL: https://github.com/apache/doris/pull/65674#discussion_r3597122842
########## be/src/format_v2/parquet/reader/native/decoder.h: ########## @@ -0,0 +1,172 @@ +// 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. + +#pragma once + +#include <gen_cpp/parquet_types.h> +#include <glog/logging.h> + +#include <cstddef> +#include <cstdint> +#include <cstring> +#include <memory> +#include <ostream> +#include <vector> + +#include "common/status.h" +#include "core/custom_allocator.h" +#include "core/data_type_serde/parquet_decode_source.h" +#include "core/types.h" +#include "util/rle_encoding.h" +#include "util/slice.h" + +namespace doris::format::parquet::native { +class Decoder : public ParquetDecodeSource { +public: + Decoder() = default; + virtual ~Decoder() = default; + + static Status get_decoder(tparquet::Type::type type, tparquet::Encoding::type encoding, + std::unique_ptr<Decoder>& decoder); + + // The type with fix length + void set_type_length(int32_t type_length) { _type_length = type_length; } + + // Set the data to be decoded + virtual Status set_data(Slice* data) { + _data = data; + _offset = 0; + return Status::OK(); + } + + Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& consumer) override { + return Status::NotSupported("Fixed values are not supported by this Parquet decoder"); + } + + Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer& consumer) override { + return Status::NotSupported("Binary values are not supported by this Parquet decoder"); + } + + Status skip_values(size_t num_values) override = 0; + + virtual Status set_dict(DorisUniqueBufferPtr<uint8_t>& dict, int32_t length, + size_t num_values) { + return Status::NotSupported("set_dict is not supported"); + } + +protected: + int32_t _type_length; + Slice* _data = nullptr; + uint32_t _offset = 0; +}; + +class BaseDictDecoder : public Decoder { +public: + BaseDictDecoder() = default; + ~BaseDictDecoder() override = default; + + // Set the data to be decoded + Status set_data(Slice* data) override { + if (UNLIKELY(data == nullptr || data->size == 0)) { + return Status::Corruption("Parquet dictionary index stream is empty"); + } + _data = data; + _offset = 0; + uint8_t bit_width = *data->data; Review Comment: [P1] Reject dictionary bit widths above 32 before constructing this decoder. This byte comes directly from the RLE_DICTIONARY page payload; with bit_width=33, a repeated run calls GetBytes<uint32_t>(5, &repeated_value_), whose size guard is only a DCHECK, so a release build copies five bytes into four-byte state. Literal runs also receive an invalid width. Return corruption here and add repeated/literal tests for widths above 32. ########## be/src/format_v2/parquet/reader/native/bool_rle_decoder.cpp: ########## @@ -0,0 +1,84 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/parquet/reader/native/bool_rle_decoder.h" + +#include <glog/logging.h> + +#include <algorithm> +#include <cstring> +#include <ostream> +#include <string> + +#include "core/column/column_vector.h" +#include "core/types.h" +#include "format/parquet/parquet_common.h" +#include "util/coding.h" +#include "util/slice.h" + +namespace doris::format::parquet::native { +Status BoolRLEDecoder::set_data(Slice* slice) { + _data = slice; + _num_bytes = slice->size; + _offset = 0; + if (_num_bytes < 4) { + return Status::IOError("Received invalid length : " + std::to_string(_num_bytes) + + " (corrupt data page?)"); + } + // Load the first 4 bytes in little-endian, which indicates the length + const auto* data = reinterpret_cast<const uint8_t*>(_data->data); + uint32_t num_bytes = decode_fixed32_le(data); + if (num_bytes > static_cast<uint32_t>(_num_bytes - 4)) { + return Status::IOError("Received invalid number of bytes : " + std::to_string(num_bytes) + + " (corrupt data page?)"); + } + _num_bytes = num_bytes; + auto decoder_data = data + 4; + _decoder = RleDecoder<uint8_t>(decoder_data, num_bytes, 1); + return Status::OK(); +} + +Status BoolRLEDecoder::skip_values(size_t num_values) { + _decoder.Skip(num_values); + return Status::OK(); +} + +Status BoolRLEDecoder::decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& consumer) { + _values.resize(num_values); + if (!_decoder.get_values(_values.data(), num_values)) { Review Comment: [P1] Require get_values() to return the requested count. It returns size_t, not bool: if a truncated stream decodes one of two advertised values it returns 1, passes this condition, and the whole two-slot vector (including a zero or retained tail) is materialized. The selected path below has the same bug, and truncated RLE bodies are currently guarded only by DCHECKs. Propagate body-read failure, compare against num_values/total_values, and return corruption on every short read. ########## be/src/format_v2/parquet/reader/native/page_reader.h: ########## @@ -0,0 +1,261 @@ +// 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. + +#pragma once + +#include <gen_cpp/parquet_types.h> +#include <stdint.h> + +#include <memory> +#include <string> + +#include "common/cast_set.h" +#include "common/config.h" +#include "common/status.h" +#include "format/parquet/parquet_common.h" +#include "storage/cache/page_cache.h" +#include "util/block_compression.h" +namespace doris { +class BlockCompressionCodec; + +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io + +} // namespace doris + +namespace doris { +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io +struct Slice; +} // namespace doris + +namespace doris::format::parquet::native { +/** + * Use to deserialize parquet page header, and get the page data in iterator interface. + */ + +// Session-level options for parquet page reading/caching. +struct ParquetPageReadContext { + bool enable_parquet_file_page_cache = true; + ParquetPageReadContext() = default; + ParquetPageReadContext(bool enable_parquet_file_page_cache) + : enable_parquet_file_page_cache(enable_parquet_file_page_cache) {} +}; + +inline bool should_cache_decompressed(const tparquet::PageHeader* header, + const tparquet::ColumnMetaData& metadata) { + if (header->compressed_page_size <= 0) return true; + if (metadata.codec == tparquet::CompressionCodec::UNCOMPRESSED) return true; + if (header->uncompressed_page_size == 0) return true; + + double ratio = static_cast<double>(header->uncompressed_page_size) / + static_cast<double>(header->compressed_page_size); + return ratio <= config::parquet_page_cache_decompress_threshold; +} + +class ParquetPageCacheKeyBuilder { +public: + void init(const std::string& path, int64_t mtime); + StoragePageCache::CacheKey make_key(uint64_t end_offset, int64_t offset) const { + return StoragePageCache::CacheKey(_file_key_prefix, end_offset, offset); + } + +private: + std::string _file_key_prefix; +}; + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +class PageReader { +public: + struct PageStatistics { + int64_t decode_header_time = 0; + int64_t skip_page_header_num = 0; + int64_t parse_page_header_num = 0; + int64_t read_page_header_time = 0; + int64_t page_cache_hit_counter = 0; + int64_t page_cache_missing_counter = 0; + int64_t page_cache_compressed_hit_counter = 0; + int64_t page_cache_decompressed_hit_counter = 0; + int64_t page_cache_write_counter = 0; + int64_t page_cache_compressed_write_counter = 0; + int64_t page_cache_decompressed_write_counter = 0; + int64_t page_read_counter = 0; + }; + + PageReader(io::BufferedStreamReader* reader, io::IOContext* io_ctx, uint64_t offset, + uint64_t length, size_t total_rows, const tparquet::ColumnMetaData& metadata, + const ParquetPageReadContext& page_read_ctx, + const tparquet::OffsetIndex* offset_index = nullptr); + ~PageReader() = default; + + bool has_next_page() const { + if constexpr (OFFSET_INDEX) { + return _page_index + 1 != _offset_index->page_locations.size(); Review Comment: [P1] Reject or fall back from an empty OffsetIndex before selecting the indexed reader. Arrow copies an empty page_locations list and load_native_offset_indexes() stores every non-null result; here 1 != 0 reports a next page, after which next_page() increments and dereferences page_locations[1]. A pruned scan can derive ranges from one column and hit this on a projected sibling. Validate the imported index and guard next_page() itself. ########## be/src/format_v2/parquet/reader/native_column_reader.cpp: ########## @@ -0,0 +1,629 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/parquet/reader/native_column_reader.h" + +#include <algorithm> +#include <cstddef> +#include <limits> +#include <ranges> +#include <string> +#include <utility> + +#include "common/cast_set.h" +#include "common/config.h" +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.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_struct.h" +#include "format/parquet/vparquet_file_metadata.h" +#include "format_v2/column_data.h" +#include "format_v2/parquet/parquet_column_schema.h" +#include "runtime/runtime_state.h" + +namespace doris::format::parquet { +namespace { + +constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20; + +DataTypePtr projected_type(const ParquetColumnSchema& schema, + const format::LocalColumnIndex* projection) { + if (!format::is_partial_projection(projection)) { + return schema.type; + } + switch (schema.kind) { + case ParquetColumnSchemaKind::PRIMITIVE: + return schema.type; + case ParquetColumnSchemaKind::STRUCT: { + DataTypes child_types; + Strings child_names; + child_types.reserve(projection->children.size()); + child_names.reserve(projection->children.size()); + for (const auto& child_projection : projection->children) { + const auto child_it = std::ranges::find_if(schema.children, [&](const auto& child) { + return child->local_id == child_projection.local_id(); + }); + DORIS_CHECK(child_it != schema.children.end()); + child_types.push_back(make_nullable(projected_type(**child_it, &child_projection))); + child_names.push_back((*child_it)->name); + } + DataTypePtr type = std::make_shared<DataTypeStruct>(child_types, child_names); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + case ParquetColumnSchemaKind::LIST: { + DORIS_CHECK(schema.children.size() == 1); + const auto* child_projection = + format::find_child_projection(projection, schema.children[0]->local_id); + DORIS_CHECK(child_projection != nullptr); + DataTypePtr type = std::make_shared<DataTypeArray>( + projected_type(*schema.children[0], child_projection)); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + case ParquetColumnSchemaKind::MAP: { + DORIS_CHECK(schema.children.size() == 2); + const auto* value_projection = + format::find_child_projection(projection, schema.children[1]->local_id); + DORIS_CHECK(value_projection != nullptr); + DataTypePtr type = std::make_shared<DataTypeMap>( + make_nullable(schema.children[0]->type), + make_nullable(projected_type(*schema.children[1], value_projection))); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + } + DORIS_CHECK(false); + return nullptr; +} + +const FieldSchema* find_child_field(const FieldSchema& parent, const ParquetColumnSchema& child) { + auto field_it = std::ranges::find_if(parent.children, [&](const FieldSchema& field) { + return (child.parquet_field_id >= 0 && field.field_id == child.parquet_field_id) || + field.name == child.name; + }); + return field_it == parent.children.end() ? nullptr : &*field_it; +} + +void collect_projected_ids(const ParquetColumnSchema& schema, + const format::LocalColumnIndex* projection, + const FieldSchema& native_field, std::set<uint64_t>* ids) { + DORIS_CHECK(ids != nullptr); + if (!format::is_partial_projection(projection)) { + return; + } + for (const auto& child_projection : projection->children) { + const auto schema_it = std::ranges::find_if(schema.children, [&](const auto& child) { + return child->local_id == child_projection.local_id(); + }); + DORIS_CHECK(schema_it != schema.children.end()); + const FieldSchema* child_field = find_child_field(native_field, **schema_it); + DORIS_CHECK(child_field != nullptr); + ids->insert(child_field->get_column_id()); + collect_projected_ids(**schema_it, &child_projection, *child_field, ids); + } + if (schema.kind == ParquetColumnSchemaKind::MAP) { + DORIS_CHECK(!native_field.children.empty()); + // MAP entry existence and offsets are owned by the key stream even for value-only + // projections. Keep the key reader live so the native complex reader can validate + // key/value entry alignment while constructing offsets. + ids->insert(native_field.children[0].get_column_id()); + } +} + +Status append_non_null_dictionary_values(MutableColumnPtr& target, MutableColumnPtr values) { + DORIS_CHECK(target); + DORIS_CHECK(values); + const size_t value_count = values->size(); + if (auto* nullable = check_and_get_column<ColumnNullable>(*target); nullable != nullptr) { + nullable->get_nested_column().insert_range_from(*values, 0, value_count); + auto& null_map = nullable->get_null_map_data(); + null_map.resize_fill(null_map.size() + value_count, 0); + return Status::OK(); + } + target->insert_range_from(*values, 0, value_count); + return Status::OK(); +} + +} // namespace + +NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema, + DataTypePtr projected_type, + ParquetColumnReaderProfile profile) + : ParquetColumnReader(schema, std::move(projected_type), profile), + _nested(schema.kind != ParquetColumnSchemaKind::PRIMITIVE) {} + +NativeColumnReader::~NativeColumnReader() { + (void)sync_native_profile(); +} + +Status NativeColumnReader::create( + const ParquetColumnSchema& column_schema, const format::LocalColumnIndex* projection, + io::FileReaderSPtr file, const FileMetaData* metadata, int row_group_id, + const std::vector<RowRange>& selected_ranges, + const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes, + const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, + bool enable_page_cache, bool enable_dictionary_filter, ParquetColumnReaderProfile profile, + std::unique_ptr<ParquetColumnReader>* reader) { + if (reader == nullptr) { + return Status::InvalidArgument("Native parquet reader result is null"); + } + if (file == nullptr || metadata == nullptr) { + return Status::InvalidArgument("Native parquet file context is not initialized"); + } + if (row_group_id < 0 || + row_group_id >= static_cast<int>(metadata->to_thrift().row_groups.size())) { + return Status::InvalidArgument("Invalid native parquet row group {}", row_group_id); + } + const auto& native_schema = metadata->schema(); + if (column_schema.local_id < 0 || column_schema.local_id >= native_schema.size()) { + return Status::InvalidArgument("Invalid native parquet top-level column id {} for {}", + column_schema.local_id, column_schema.name); + } + auto* field = const_cast<FieldSchema*>(native_schema.get_column(column_schema.local_id)); + DORIS_CHECK(field != nullptr); + if (field->name != column_schema.name && + !(field->field_id >= 0 && field->field_id == column_schema.parquet_field_id)) { + return Status::Corruption( + "Native/metadata parquet schema mismatch at column {}: native={}, arrow={}", + column_schema.local_id, field->name, column_schema.name); + } + + auto type = projected_type(column_schema, projection); + std::shared_ptr<TableSchemaChangeHelper::Node> schema_node; + RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_name(type, *field, + schema_node)); + std::set<uint64_t> projected_ids; + collect_projected_ids(column_schema, projection, *field, &projected_ids); + + auto native_reader = std::unique_ptr<NativeColumnReader>( + new NativeColumnReader(column_schema, std::move(type), profile)); + RETURN_IF_ERROR(native_reader->init( + std::move(file), metadata, row_group_id, field, std::move(schema_node), + std::move(projected_ids), selected_ranges, offset_indexes, timezone, io_ctx, + runtime_state, enable_page_cache, enable_dictionary_filter)); + *reader = std::move(native_reader); + return Status::OK(); +} + +Status NativeColumnReader::init( + io::FileReaderSPtr file, const FileMetaData* metadata, int row_group_id, FieldSchema* field, + std::shared_ptr<TableSchemaChangeHelper::Node> schema_node, + std::set<uint64_t> projected_column_ids, const std::vector<RowRange>& selected_ranges, + const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes, + const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, + bool enable_page_cache, bool enable_dictionary_filter) { + DORIS_CHECK(file != nullptr); + DORIS_CHECK(metadata != nullptr); + DORIS_CHECK(field != nullptr); + DORIS_CHECK(schema_node != nullptr); + const auto& row_group = metadata->to_thrift().row_groups[row_group_id]; + DORIS_CHECK(row_group.num_rows > 0); + _row_group_rows = row_group.num_rows; + _selected_ranges = selected_ranges; + DORIS_CHECK(!_selected_ranges.empty()); + for (const auto& range : _selected_ranges) { + DORIS_CHECK(range.start >= 0); + DORIS_CHECK(range.length > 0); + DORIS_CHECK(range.start + range.length <= _row_group_rows); + _row_ranges.add(::doris::RowRange(range.start, range.start + range.length)); + } + _offset_indexes = offset_indexes; + _schema_node = std::move(schema_node); + _projected_column_ids = std::move(projected_column_ids); + _dictionary_filter_enabled = enable_dictionary_filter; + + const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb << 20; + const size_t max_column_buffer = config::parquet_column_max_buffer_mb << 20; + const size_t max_buffer_size = std::min(max_group_buffer, max_column_buffer); + RuntimeState* native_runtime_state = runtime_state; + const bool runtime_page_cache_enabled = + runtime_state == nullptr || + runtime_state->query_options().enable_parquet_file_page_cache; + if (runtime_page_cache_enabled != enable_page_cache) { + TQueryOptions query_options = + runtime_state == nullptr ? TQueryOptions() : runtime_state->query_options(); + query_options.__set_enable_parquet_file_page_cache(enable_page_cache); + _page_cache_runtime_state = RuntimeState::create_unique(query_options, TQueryGlobals()); + native_runtime_state = _page_cache_runtime_state.get(); + } + RETURN_IF_ERROR(native::ColumnReader::create(std::move(file), field, row_group, _row_ranges, + timezone, io_ctx, _native_reader, max_buffer_size, + _offset_indexes, native_runtime_state, false, + _projected_column_ids, _filter_column_ids)); + DORIS_CHECK(_native_reader != nullptr); + _skip_column = _type->create_column(); + return Status::OK(); +} + +Status NativeColumnReader::read_with_filter(int64_t rows, const uint8_t* filter_data, + bool filter_all, MutableColumnPtr& column, + const DataTypePtr& output_type, bool dictionary_ids, + int64_t* rows_read) { + DORIS_CHECK(rows >= 0); + DORIS_CHECK(column); + DORIS_CHECK(output_type != nullptr); + DORIS_CHECK(rows_read != nullptr); + *rows_read = 0; + if (rows == 0) { + return Status::OK(); + } + + ::doris::FilterMap filter; + RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows), filter_all)); + _native_reader->reset_filter_map_index(); + ColumnPtr native_column(std::move(column)); + bool eof = false; + int64_t native_calls = 0; + int64_t consecutive_empty_calls = 0; + while (*rows_read < rows && !eof) { + ++native_calls; + size_t loop_rows = 0; + RETURN_IF_ERROR(_native_reader->read_column_data( + native_column, output_type, _schema_node, filter, + static_cast<size_t>(rows - *rows_read), &loop_rows, &eof, dictionary_ids)); + if (loop_rows == 0 && !eof) { + // A selected RowRanges plan may reject the current data page completely. V1 advances + // the page cursor and deliberately returns zero rows so the caller can request the + // next page. Bound consecutive empty transitions by the Row Group row count to retain + // a deterministic corruption exit if a decoder ever stops advancing. + if (++consecutive_empty_calls > _row_group_rows + 1) { + column = IColumn::mutate(std::move(native_column)); + return Status::Corruption("Native parquet reader made no progress for column {}", + _name); + } + continue; + } + consecutive_empty_calls = 0; + *rows_read += static_cast<int64_t>(loop_rows); + } + column = IColumn::mutate(std::move(native_column)); + if (_profile.native_read_calls != nullptr) { + COUNTER_UPDATE(_profile.native_read_calls, native_calls); + } + if (_nested && _profile.nested_batches != nullptr) { + COUNTER_UPDATE(_profile.nested_batches, 1); + } + _native_reader->release_batch_scratch(MAX_RETAINED_BATCH_SCRATCH_BYTES); + if (*rows_read != rows) { + return Status::Corruption("Native parquet reader returned {} rows, expected {} for {}", + *rows_read, rows, _name); + } + return Status::OK(); +} + +Status NativeColumnReader::validate_selected_span(int64_t rows) { + DORIS_CHECK(rows >= 0); + while (_selected_range_idx < _selected_ranges.size()) { + const auto& range = _selected_ranges[_selected_range_idx]; + const int64_t range_end = range.start + range.length; + if (_logical_row_position < range_end) { + break; + } + ++_selected_range_idx; + } + if (_selected_range_idx >= _selected_ranges.size()) { + return Status::Corruption("Native parquet read past selected ranges for column {}", _name); + } + const auto& range = _selected_ranges[_selected_range_idx]; + if (_logical_row_position < range.start || + rows > range.start + range.length - _logical_row_position) { + return Status::Corruption( + "Native parquet read [{}, {}) crosses selected range [{}, {}) for column {}", + _logical_row_position, _logical_row_position + rows, range.start, + range.start + range.length, _name); + } + return Status::OK(); +} + +void NativeColumnReader::advance_selected_span(int64_t rows) { + _logical_row_position += rows; + while (_selected_range_idx < _selected_ranges.size() && + _logical_row_position >= _selected_ranges[_selected_range_idx].start + + _selected_ranges[_selected_range_idx].length) { + ++_selected_range_idx; + } +} + +Status NativeColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) { + RETURN_IF_ERROR(validate_selected_span(rows)); + RETURN_IF_ERROR(read_with_filter(rows, nullptr, false, column, _type, false, rows_read)); + advance_selected_span(*rows_read); + update_reader_read_rows(*rows_read); + record_page_fragments(sync_native_profile()); + return Status::OK(); +} + +Status NativeColumnReader::skip(int64_t rows) { + if (rows <= 0) { + return Status::OK(); + } + DORIS_CHECK(_logical_row_position <= _row_group_rows - rows); + int64_t remaining = rows; + int64_t native_skipped_rows = 0; + while (remaining > 0) { + while (_selected_range_idx < _selected_ranges.size() && + _logical_row_position >= _selected_ranges[_selected_range_idx].start + + _selected_ranges[_selected_range_idx].length) { + ++_selected_range_idx; + } + if (_selected_range_idx >= _selected_ranges.size()) { + _logical_row_position += remaining; + break; + } + const auto& range = _selected_ranges[_selected_range_idx]; + if (_logical_row_position < range.start) { + const int64_t gap = std::min(remaining, range.start - _logical_row_position); + _logical_row_position += gap; + remaining -= gap; + continue; + } + const int64_t selected_rows = + std::min(remaining, range.start + range.length - _logical_row_position); + _skip_column->clear(); + _filter_scratch.resize(static_cast<size_t>(selected_rows)); Review Comment: [P1] Clear this scratch buffer before using it for an all-filtered skip. select() leaves survivor bits set and resize() preserves that prefix; nested gen_filter_map() ignores FilterMap::filter_all() and copies the raw bytes, so stale ones materialize values into _skip_column and the empty-column check below terminates the BE. This is reachable after a partial nested batch followed by a pending lazy skip. Zero the buffer and/or make nested filtering honor filter_all, with a select -> skip -> read test. ########## be/src/format_v2/parquet/reader/native_column_reader.cpp: ########## @@ -0,0 +1,629 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/parquet/reader/native_column_reader.h" + +#include <algorithm> +#include <cstddef> +#include <limits> +#include <ranges> +#include <string> +#include <utility> + +#include "common/cast_set.h" +#include "common/config.h" +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.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_struct.h" +#include "format/parquet/vparquet_file_metadata.h" +#include "format_v2/column_data.h" +#include "format_v2/parquet/parquet_column_schema.h" +#include "runtime/runtime_state.h" + +namespace doris::format::parquet { +namespace { + +constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20; + +DataTypePtr projected_type(const ParquetColumnSchema& schema, + const format::LocalColumnIndex* projection) { + if (!format::is_partial_projection(projection)) { + return schema.type; + } + switch (schema.kind) { + case ParquetColumnSchemaKind::PRIMITIVE: + return schema.type; + case ParquetColumnSchemaKind::STRUCT: { + DataTypes child_types; + Strings child_names; + child_types.reserve(projection->children.size()); + child_names.reserve(projection->children.size()); + for (const auto& child_projection : projection->children) { + const auto child_it = std::ranges::find_if(schema.children, [&](const auto& child) { + return child->local_id == child_projection.local_id(); + }); + DORIS_CHECK(child_it != schema.children.end()); + child_types.push_back(make_nullable(projected_type(**child_it, &child_projection))); + child_names.push_back((*child_it)->name); + } + DataTypePtr type = std::make_shared<DataTypeStruct>(child_types, child_names); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + case ParquetColumnSchemaKind::LIST: { + DORIS_CHECK(schema.children.size() == 1); + const auto* child_projection = + format::find_child_projection(projection, schema.children[0]->local_id); + DORIS_CHECK(child_projection != nullptr); + DataTypePtr type = std::make_shared<DataTypeArray>( + projected_type(*schema.children[0], child_projection)); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + case ParquetColumnSchemaKind::MAP: { + DORIS_CHECK(schema.children.size() == 2); + const auto* value_projection = + format::find_child_projection(projection, schema.children[1]->local_id); + DORIS_CHECK(value_projection != nullptr); + DataTypePtr type = std::make_shared<DataTypeMap>( + make_nullable(schema.children[0]->type), + make_nullable(projected_type(*schema.children[1], value_projection))); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + } + DORIS_CHECK(false); + return nullptr; +} + +const FieldSchema* find_child_field(const FieldSchema& parent, const ParquetColumnSchema& child) { + auto field_it = std::ranges::find_if(parent.children, [&](const FieldSchema& field) { + return (child.parquet_field_id >= 0 && field.field_id == child.parquet_field_id) || + field.name == child.name; + }); + return field_it == parent.children.end() ? nullptr : &*field_it; +} + +void collect_projected_ids(const ParquetColumnSchema& schema, + const format::LocalColumnIndex* projection, + const FieldSchema& native_field, std::set<uint64_t>* ids) { + DORIS_CHECK(ids != nullptr); + if (!format::is_partial_projection(projection)) { + return; + } + for (const auto& child_projection : projection->children) { + const auto schema_it = std::ranges::find_if(schema.children, [&](const auto& child) { + return child->local_id == child_projection.local_id(); + }); + DORIS_CHECK(schema_it != schema.children.end()); + const FieldSchema* child_field = find_child_field(native_field, **schema_it); + DORIS_CHECK(child_field != nullptr); + ids->insert(child_field->get_column_id()); + collect_projected_ids(**schema_it, &child_projection, *child_field, ids); + } + if (schema.kind == ParquetColumnSchemaKind::MAP) { + DORIS_CHECK(!native_field.children.empty()); + // MAP entry existence and offsets are owned by the key stream even for value-only + // projections. Keep the key reader live so the native complex reader can validate + // key/value entry alignment while constructing offsets. + ids->insert(native_field.children[0].get_column_id()); + } +} + +Status append_non_null_dictionary_values(MutableColumnPtr& target, MutableColumnPtr values) { + DORIS_CHECK(target); + DORIS_CHECK(values); + const size_t value_count = values->size(); + if (auto* nullable = check_and_get_column<ColumnNullable>(*target); nullable != nullptr) { + nullable->get_nested_column().insert_range_from(*values, 0, value_count); + auto& null_map = nullable->get_null_map_data(); + null_map.resize_fill(null_map.size() + value_count, 0); + return Status::OK(); + } + target->insert_range_from(*values, 0, value_count); + return Status::OK(); +} + +} // namespace + +NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema, + DataTypePtr projected_type, + ParquetColumnReaderProfile profile) + : ParquetColumnReader(schema, std::move(projected_type), profile), + _nested(schema.kind != ParquetColumnSchemaKind::PRIMITIVE) {} + +NativeColumnReader::~NativeColumnReader() { + (void)sync_native_profile(); +} + +Status NativeColumnReader::create( + const ParquetColumnSchema& column_schema, const format::LocalColumnIndex* projection, + io::FileReaderSPtr file, const FileMetaData* metadata, int row_group_id, + const std::vector<RowRange>& selected_ranges, + const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes, + const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, + bool enable_page_cache, bool enable_dictionary_filter, ParquetColumnReaderProfile profile, + std::unique_ptr<ParquetColumnReader>* reader) { + if (reader == nullptr) { + return Status::InvalidArgument("Native parquet reader result is null"); + } + if (file == nullptr || metadata == nullptr) { + return Status::InvalidArgument("Native parquet file context is not initialized"); + } + if (row_group_id < 0 || + row_group_id >= static_cast<int>(metadata->to_thrift().row_groups.size())) { + return Status::InvalidArgument("Invalid native parquet row group {}", row_group_id); + } + const auto& native_schema = metadata->schema(); + if (column_schema.local_id < 0 || column_schema.local_id >= native_schema.size()) { + return Status::InvalidArgument("Invalid native parquet top-level column id {} for {}", + column_schema.local_id, column_schema.name); + } + auto* field = const_cast<FieldSchema*>(native_schema.get_column(column_schema.local_id)); + DORIS_CHECK(field != nullptr); + if (field->name != column_schema.name && + !(field->field_id >= 0 && field->field_id == column_schema.parquet_field_id)) { + return Status::Corruption( + "Native/metadata parquet schema mismatch at column {}: native={}, arrow={}", + column_schema.local_id, field->name, column_schema.name); + } + + auto type = projected_type(column_schema, projection); + std::shared_ptr<TableSchemaChangeHelper::Node> schema_node; + RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_name(type, *field, + schema_node)); + std::set<uint64_t> projected_ids; + collect_projected_ids(column_schema, projection, *field, &projected_ids); + + auto native_reader = std::unique_ptr<NativeColumnReader>( + new NativeColumnReader(column_schema, std::move(type), profile)); + RETURN_IF_ERROR(native_reader->init( + std::move(file), metadata, row_group_id, field, std::move(schema_node), + std::move(projected_ids), selected_ranges, offset_indexes, timezone, io_ctx, + runtime_state, enable_page_cache, enable_dictionary_filter)); + *reader = std::move(native_reader); + return Status::OK(); +} + +Status NativeColumnReader::init( + io::FileReaderSPtr file, const FileMetaData* metadata, int row_group_id, FieldSchema* field, + std::shared_ptr<TableSchemaChangeHelper::Node> schema_node, + std::set<uint64_t> projected_column_ids, const std::vector<RowRange>& selected_ranges, + const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes, + const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, + bool enable_page_cache, bool enable_dictionary_filter) { + DORIS_CHECK(file != nullptr); + DORIS_CHECK(metadata != nullptr); + DORIS_CHECK(field != nullptr); + DORIS_CHECK(schema_node != nullptr); + const auto& row_group = metadata->to_thrift().row_groups[row_group_id]; + DORIS_CHECK(row_group.num_rows > 0); + _row_group_rows = row_group.num_rows; + _selected_ranges = selected_ranges; + DORIS_CHECK(!_selected_ranges.empty()); + for (const auto& range : _selected_ranges) { + DORIS_CHECK(range.start >= 0); + DORIS_CHECK(range.length > 0); + DORIS_CHECK(range.start + range.length <= _row_group_rows); + _row_ranges.add(::doris::RowRange(range.start, range.start + range.length)); + } + _offset_indexes = offset_indexes; + _schema_node = std::move(schema_node); + _projected_column_ids = std::move(projected_column_ids); + _dictionary_filter_enabled = enable_dictionary_filter; + + const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb << 20; + const size_t max_column_buffer = config::parquet_column_max_buffer_mb << 20; + const size_t max_buffer_size = std::min(max_group_buffer, max_column_buffer); + RuntimeState* native_runtime_state = runtime_state; + const bool runtime_page_cache_enabled = + runtime_state == nullptr || + runtime_state->query_options().enable_parquet_file_page_cache; + if (runtime_page_cache_enabled != enable_page_cache) { + TQueryOptions query_options = + runtime_state == nullptr ? TQueryOptions() : runtime_state->query_options(); + query_options.__set_enable_parquet_file_page_cache(enable_page_cache); + _page_cache_runtime_state = RuntimeState::create_unique(query_options, TQueryGlobals()); + native_runtime_state = _page_cache_runtime_state.get(); + } + RETURN_IF_ERROR(native::ColumnReader::create(std::move(file), field, row_group, _row_ranges, + timezone, io_ctx, _native_reader, max_buffer_size, + _offset_indexes, native_runtime_state, false, + _projected_column_ids, _filter_column_ids)); + DORIS_CHECK(_native_reader != nullptr); + _skip_column = _type->create_column(); + return Status::OK(); +} + +Status NativeColumnReader::read_with_filter(int64_t rows, const uint8_t* filter_data, + bool filter_all, MutableColumnPtr& column, + const DataTypePtr& output_type, bool dictionary_ids, + int64_t* rows_read) { + DORIS_CHECK(rows >= 0); + DORIS_CHECK(column); + DORIS_CHECK(output_type != nullptr); + DORIS_CHECK(rows_read != nullptr); + *rows_read = 0; + if (rows == 0) { + return Status::OK(); + } + + ::doris::FilterMap filter; + RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows), filter_all)); + _native_reader->reset_filter_map_index(); + ColumnPtr native_column(std::move(column)); + bool eof = false; + int64_t native_calls = 0; + int64_t consecutive_empty_calls = 0; + while (*rows_read < rows && !eof) { + ++native_calls; + size_t loop_rows = 0; + RETURN_IF_ERROR(_native_reader->read_column_data( + native_column, output_type, _schema_node, filter, + static_cast<size_t>(rows - *rows_read), &loop_rows, &eof, dictionary_ids)); + if (loop_rows == 0 && !eof) { + // A selected RowRanges plan may reject the current data page completely. V1 advances + // the page cursor and deliberately returns zero rows so the caller can request the + // next page. Bound consecutive empty transitions by the Row Group row count to retain + // a deterministic corruption exit if a decoder ever stops advancing. + if (++consecutive_empty_calls > _row_group_rows + 1) { + column = IColumn::mutate(std::move(native_column)); + return Status::Corruption("Native parquet reader made no progress for column {}", + _name); + } + continue; + } + consecutive_empty_calls = 0; + *rows_read += static_cast<int64_t>(loop_rows); + } + column = IColumn::mutate(std::move(native_column)); + if (_profile.native_read_calls != nullptr) { + COUNTER_UPDATE(_profile.native_read_calls, native_calls); + } + if (_nested && _profile.nested_batches != nullptr) { + COUNTER_UPDATE(_profile.nested_batches, 1); + } + _native_reader->release_batch_scratch(MAX_RETAINED_BATCH_SCRATCH_BYTES); + if (*rows_read != rows) { + return Status::Corruption("Native parquet reader returned {} rows, expected {} for {}", + *rows_read, rows, _name); + } + return Status::OK(); +} + +Status NativeColumnReader::validate_selected_span(int64_t rows) { + DORIS_CHECK(rows >= 0); + while (_selected_range_idx < _selected_ranges.size()) { + const auto& range = _selected_ranges[_selected_range_idx]; + const int64_t range_end = range.start + range.length; + if (_logical_row_position < range_end) { + break; + } + ++_selected_range_idx; + } + if (_selected_range_idx >= _selected_ranges.size()) { + return Status::Corruption("Native parquet read past selected ranges for column {}", _name); + } + const auto& range = _selected_ranges[_selected_range_idx]; + if (_logical_row_position < range.start || + rows > range.start + range.length - _logical_row_position) { + return Status::Corruption( + "Native parquet read [{}, {}) crosses selected range [{}, {}) for column {}", + _logical_row_position, _logical_row_position + rows, range.start, + range.start + range.length, _name); + } + return Status::OK(); +} + +void NativeColumnReader::advance_selected_span(int64_t rows) { + _logical_row_position += rows; + while (_selected_range_idx < _selected_ranges.size() && + _logical_row_position >= _selected_ranges[_selected_range_idx].start + + _selected_ranges[_selected_range_idx].length) { + ++_selected_range_idx; + } +} + +Status NativeColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) { + RETURN_IF_ERROR(validate_selected_span(rows)); + RETURN_IF_ERROR(read_with_filter(rows, nullptr, false, column, _type, false, rows_read)); + advance_selected_span(*rows_read); + update_reader_read_rows(*rows_read); + record_page_fragments(sync_native_profile()); + return Status::OK(); +} + +Status NativeColumnReader::skip(int64_t rows) { + if (rows <= 0) { + return Status::OK(); + } + DORIS_CHECK(_logical_row_position <= _row_group_rows - rows); + int64_t remaining = rows; + int64_t native_skipped_rows = 0; + while (remaining > 0) { + while (_selected_range_idx < _selected_ranges.size() && + _logical_row_position >= _selected_ranges[_selected_range_idx].start + + _selected_ranges[_selected_range_idx].length) { + ++_selected_range_idx; + } + if (_selected_range_idx >= _selected_ranges.size()) { + _logical_row_position += remaining; + break; + } + const auto& range = _selected_ranges[_selected_range_idx]; + if (_logical_row_position < range.start) { + const int64_t gap = std::min(remaining, range.start - _logical_row_position); + _logical_row_position += gap; + remaining -= gap; + continue; + } + const int64_t selected_rows = + std::min(remaining, range.start + range.length - _logical_row_position); + _skip_column->clear(); + _filter_scratch.resize(static_cast<size_t>(selected_rows)); + int64_t rows_read = 0; + RETURN_IF_ERROR(read_with_filter(selected_rows, _filter_scratch.data(), true, _skip_column, + _type, false, &rows_read)); + DORIS_CHECK(_skip_column->empty()); + DORIS_CHECK(rows_read == selected_rows); + _logical_row_position += rows_read; + native_skipped_rows += rows_read; + remaining -= rows_read; + } + update_reader_skip_rows(native_skipped_rows); + record_page_fragments(sync_native_profile()); + return Status::OK(); +} + +Status NativeColumnReader::select(const SelectionVector& selection, uint16_t selected_rows, + int64_t batch_rows, MutableColumnPtr& column) { + RETURN_IF_ERROR(selection.verify(selected_rows, batch_rows)); + RETURN_IF_ERROR(validate_selected_span(batch_rows)); + _filter_scratch.assign(static_cast<size_t>(batch_rows), 0); + for (uint16_t idx = 0; idx < selected_rows; ++idx) { + _filter_scratch[selection.get_index(idx)] = 1; + } + const size_t old_size = column->size(); + int64_t rows_read = 0; + RETURN_IF_ERROR(read_with_filter(batch_rows, _filter_scratch.data(), selected_rows == 0, column, + _type, false, &rows_read)); + advance_selected_span(rows_read); + if (column->size() != old_size + selected_rows) { + return Status::Corruption( + "Native parquet selection appended {} rows, expected {} for column {}", + column->size() - old_size, selected_rows, _name); + } + if (_profile.reader_select_rows != nullptr) { + COUNTER_UPDATE(_profile.reader_select_rows, selected_rows); + } + update_reader_read_rows(selected_rows); + update_reader_skip_rows(batch_rows - selected_rows); + record_page_fragments(sync_native_profile()); + return Status::OK(); +} + +Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& selection, + uint16_t selected_rows, int64_t batch_rows, + const IColumn::Filter& dictionary_filter, + MutableColumnPtr& column, + IColumn::Filter* row_filter, + bool* used_filter) { + DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(used_filter != nullptr); + RETURN_IF_ERROR(selection.verify(selected_rows, batch_rows)); + RETURN_IF_ERROR(validate_selected_span(batch_rows)); + *used_filter = false; + row_filter->clear(); + if (!_dictionary_filter_enabled) { + return Status::OK(); + } + *used_filter = true; + + _filter_scratch.assign(static_cast<size_t>(batch_rows), 0); + for (uint16_t idx = 0; idx < selected_rows; ++idx) { + _filter_scratch[selection.get_index(idx)] = 1; + } + const bool nullable = _type->is_nullable(); + DataTypePtr id_type = std::make_shared<DataTypeInt32>(); + if (nullable) { + id_type = make_nullable(id_type); + } + if (!_dictionary_id_column) { + _dictionary_id_column = id_type->create_column(); + } + _dictionary_id_column->clear(); + int64_t rows_read = 0; + RETURN_IF_ERROR(read_with_filter(batch_rows, _filter_scratch.data(), selected_rows == 0, + _dictionary_id_column, id_type, true, &rows_read)); + advance_selected_span(rows_read); + if (_dictionary_id_column->size() != selected_rows) { + return Status::Corruption( + "Native parquet dictionary reader appended {} rows, expected {} for {}", + _dictionary_id_column->size(), selected_rows, _name); + } + + const ColumnInt32* ids = nullptr; + const NullMap* null_map = nullptr; + if (const auto* nullable_ids = check_and_get_column<ColumnNullable>(*_dictionary_id_column); + nullable_ids != nullptr) { + ids = check_and_get_column<ColumnInt32>(nullable_ids->get_nested_column()); + null_map = &nullable_ids->get_null_map_data(); + } else { + ids = check_and_get_column<ColumnInt32>(*_dictionary_id_column); + } + DORIS_CHECK(ids != nullptr); + + if (!_matched_dictionary_ids) { + _matched_dictionary_ids = ColumnInt32::create(); + } + _matched_dictionary_ids->clear(); + auto& matched_ids = assert_cast<ColumnInt32&>(*_matched_dictionary_ids).get_data(); + row_filter->reserve(selected_rows); + const auto& id_data = ids->get_data(); + for (size_t row = 0; row < selected_rows; ++row) { + bool keep = false; + if (null_map == nullptr || (*null_map)[row] == 0) { + const int32_t dictionary_id = id_data[row]; + if (dictionary_id < 0 || + static_cast<size_t>(dictionary_id) >= dictionary_filter.size()) { + return Status::Corruption( + "Invalid parquet dictionary id {} for column {} with {} entries", + dictionary_id, _name, dictionary_filter.size()); + } + keep = dictionary_filter[static_cast<size_t>(dictionary_id)] != 0; + if (keep) { + matched_ids.push_back(dictionary_id); + } + } + row_filter->push_back(keep ? 1 : 0); + } + + const auto* matched_id_column = check_and_get_column<ColumnInt32>(*_matched_dictionary_ids); + DORIS_CHECK(matched_id_column != nullptr); + auto string_values = + DORIS_TRY(_native_reader->convert_dict_column_to_string_column(matched_id_column)); + RETURN_IF_ERROR(append_non_null_dictionary_values(column, std::move(string_values))); + if (_profile.reader_select_rows != nullptr) { + COUNTER_UPDATE(_profile.reader_select_rows, selected_rows); + } + update_reader_read_rows(cast_set<int64_t>(matched_ids.size())); + update_reader_skip_rows(batch_rows - cast_set<int64_t>(matched_ids.size())); + record_page_fragments(sync_native_profile()); + return Status::OK(); +} + +void NativeColumnReader::record_page_fragments(int64_t page_fragments) { + if (_profile.native_page_fragments != nullptr) { + COUNTER_UPDATE(_profile.native_page_fragments, page_fragments); + } + if (page_fragments > 1 && _profile.page_crossing_batches != nullptr) { Review Comment: [P2] Do not infer a page-crossing batch from the sum of child page reads. MAP/STRUCT column_statistics() adds each leaf's page_read_counter, so one page from each of two siblings yields page_fragments=2 and increments this counter even though neither leaf crossed a boundary. Track a per-leaf crossing (or an explicit crossing delta) so PageCrossingBatches remains meaningful for complex columns. ########## be/src/format_v2/parquet/reader/native_column_reader.cpp: ########## @@ -0,0 +1,629 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/parquet/reader/native_column_reader.h" + +#include <algorithm> +#include <cstddef> +#include <limits> +#include <ranges> +#include <string> +#include <utility> + +#include "common/cast_set.h" +#include "common/config.h" +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.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_struct.h" +#include "format/parquet/vparquet_file_metadata.h" +#include "format_v2/column_data.h" +#include "format_v2/parquet/parquet_column_schema.h" +#include "runtime/runtime_state.h" + +namespace doris::format::parquet { +namespace { + +constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20; + +DataTypePtr projected_type(const ParquetColumnSchema& schema, + const format::LocalColumnIndex* projection) { + if (!format::is_partial_projection(projection)) { + return schema.type; + } + switch (schema.kind) { + case ParquetColumnSchemaKind::PRIMITIVE: + return schema.type; + case ParquetColumnSchemaKind::STRUCT: { + DataTypes child_types; + Strings child_names; + child_types.reserve(projection->children.size()); + child_names.reserve(projection->children.size()); + for (const auto& child_projection : projection->children) { + const auto child_it = std::ranges::find_if(schema.children, [&](const auto& child) { + return child->local_id == child_projection.local_id(); + }); + DORIS_CHECK(child_it != schema.children.end()); + child_types.push_back(make_nullable(projected_type(**child_it, &child_projection))); + child_names.push_back((*child_it)->name); + } + DataTypePtr type = std::make_shared<DataTypeStruct>(child_types, child_names); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + case ParquetColumnSchemaKind::LIST: { + DORIS_CHECK(schema.children.size() == 1); + const auto* child_projection = + format::find_child_projection(projection, schema.children[0]->local_id); + DORIS_CHECK(child_projection != nullptr); + DataTypePtr type = std::make_shared<DataTypeArray>( + projected_type(*schema.children[0], child_projection)); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + case ParquetColumnSchemaKind::MAP: { + DORIS_CHECK(schema.children.size() == 2); + const auto* value_projection = + format::find_child_projection(projection, schema.children[1]->local_id); + DORIS_CHECK(value_projection != nullptr); + DataTypePtr type = std::make_shared<DataTypeMap>( + make_nullable(schema.children[0]->type), + make_nullable(projected_type(*schema.children[1], value_projection))); + return schema.type->is_nullable() ? make_nullable(type) : type; + } + } + DORIS_CHECK(false); + return nullptr; +} + +const FieldSchema* find_child_field(const FieldSchema& parent, const ParquetColumnSchema& child) { + auto field_it = std::ranges::find_if(parent.children, [&](const FieldSchema& field) { + return (child.parquet_field_id >= 0 && field.field_id == child.parquet_field_id) || + field.name == child.name; + }); + return field_it == parent.children.end() ? nullptr : &*field_it; +} + +void collect_projected_ids(const ParquetColumnSchema& schema, + const format::LocalColumnIndex* projection, + const FieldSchema& native_field, std::set<uint64_t>* ids) { + DORIS_CHECK(ids != nullptr); + if (!format::is_partial_projection(projection)) { + return; + } + for (const auto& child_projection : projection->children) { + const auto schema_it = std::ranges::find_if(schema.children, [&](const auto& child) { + return child->local_id == child_projection.local_id(); + }); + DORIS_CHECK(schema_it != schema.children.end()); + const FieldSchema* child_field = find_child_field(native_field, **schema_it); + DORIS_CHECK(child_field != nullptr); + ids->insert(child_field->get_column_id()); + collect_projected_ids(**schema_it, &child_projection, *child_field, ids); + } + if (schema.kind == ParquetColumnSchemaKind::MAP) { + DORIS_CHECK(!native_field.children.empty()); + // MAP entry existence and offsets are owned by the key stream even for value-only + // projections. Keep the key reader live so the native complex reader can validate + // key/value entry alignment while constructing offsets. + ids->insert(native_field.children[0].get_column_id()); + } +} + +Status append_non_null_dictionary_values(MutableColumnPtr& target, MutableColumnPtr values) { + DORIS_CHECK(target); + DORIS_CHECK(values); + const size_t value_count = values->size(); + if (auto* nullable = check_and_get_column<ColumnNullable>(*target); nullable != nullptr) { + nullable->get_nested_column().insert_range_from(*values, 0, value_count); + auto& null_map = nullable->get_null_map_data(); + null_map.resize_fill(null_map.size() + value_count, 0); + return Status::OK(); + } + target->insert_range_from(*values, 0, value_count); + return Status::OK(); +} + +} // namespace + +NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema, + DataTypePtr projected_type, + ParquetColumnReaderProfile profile) + : ParquetColumnReader(schema, std::move(projected_type), profile), + _nested(schema.kind != ParquetColumnSchemaKind::PRIMITIVE) {} + +NativeColumnReader::~NativeColumnReader() { + (void)sync_native_profile(); +} + +Status NativeColumnReader::create( + const ParquetColumnSchema& column_schema, const format::LocalColumnIndex* projection, + io::FileReaderSPtr file, const FileMetaData* metadata, int row_group_id, + const std::vector<RowRange>& selected_ranges, + const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes, + const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, + bool enable_page_cache, bool enable_dictionary_filter, ParquetColumnReaderProfile profile, + std::unique_ptr<ParquetColumnReader>* reader) { + if (reader == nullptr) { + return Status::InvalidArgument("Native parquet reader result is null"); + } + if (file == nullptr || metadata == nullptr) { + return Status::InvalidArgument("Native parquet file context is not initialized"); + } + if (row_group_id < 0 || + row_group_id >= static_cast<int>(metadata->to_thrift().row_groups.size())) { + return Status::InvalidArgument("Invalid native parquet row group {}", row_group_id); + } + const auto& native_schema = metadata->schema(); + if (column_schema.local_id < 0 || column_schema.local_id >= native_schema.size()) { + return Status::InvalidArgument("Invalid native parquet top-level column id {} for {}", + column_schema.local_id, column_schema.name); + } + auto* field = const_cast<FieldSchema*>(native_schema.get_column(column_schema.local_id)); + DORIS_CHECK(field != nullptr); + if (field->name != column_schema.name && + !(field->field_id >= 0 && field->field_id == column_schema.parquet_field_id)) { + return Status::Corruption( + "Native/metadata parquet schema mismatch at column {}: native={}, arrow={}", + column_schema.local_id, field->name, column_schema.name); + } + + auto type = projected_type(column_schema, projection); + std::shared_ptr<TableSchemaChangeHelper::Node> schema_node; + RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_name(type, *field, + schema_node)); + std::set<uint64_t> projected_ids; + collect_projected_ids(column_schema, projection, *field, &projected_ids); + + auto native_reader = std::unique_ptr<NativeColumnReader>( + new NativeColumnReader(column_schema, std::move(type), profile)); + RETURN_IF_ERROR(native_reader->init( + std::move(file), metadata, row_group_id, field, std::move(schema_node), + std::move(projected_ids), selected_ranges, offset_indexes, timezone, io_ctx, + runtime_state, enable_page_cache, enable_dictionary_filter)); + *reader = std::move(native_reader); + return Status::OK(); +} + +Status NativeColumnReader::init( + io::FileReaderSPtr file, const FileMetaData* metadata, int row_group_id, FieldSchema* field, + std::shared_ptr<TableSchemaChangeHelper::Node> schema_node, + std::set<uint64_t> projected_column_ids, const std::vector<RowRange>& selected_ranges, + const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes, + const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* runtime_state, + bool enable_page_cache, bool enable_dictionary_filter) { + DORIS_CHECK(file != nullptr); + DORIS_CHECK(metadata != nullptr); + DORIS_CHECK(field != nullptr); + DORIS_CHECK(schema_node != nullptr); + const auto& row_group = metadata->to_thrift().row_groups[row_group_id]; + DORIS_CHECK(row_group.num_rows > 0); + _row_group_rows = row_group.num_rows; + _selected_ranges = selected_ranges; + DORIS_CHECK(!_selected_ranges.empty()); + for (const auto& range : _selected_ranges) { + DORIS_CHECK(range.start >= 0); + DORIS_CHECK(range.length > 0); + DORIS_CHECK(range.start + range.length <= _row_group_rows); + _row_ranges.add(::doris::RowRange(range.start, range.start + range.length)); + } + _offset_indexes = offset_indexes; Review Comment: [P1] Avoid copying the full Row Group OffsetIndex map into every projected reader. open_next_row_group() loads indexes for all projected leaves and passes that same map to each top-level NativeColumnReader; this assignment deep-copies all page_locations each time even though a reader uses only its own subtree. With N scalar columns and P pages this retains O(N^2*P) locations and can turn a legal wide filtered scan into a memory-limit failure. Share immutable Row Group metadata or retain only this projection's entries. ########## be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.h: ########## @@ -0,0 +1,523 @@ +// 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. + +#pragma once + +#include <gen_cpp/parquet_types.h> +#include <glog/logging.h> + +#include <cstddef> +#include <cstdint> +#include <cstring> +#include <limits> +#include <memory> +#include <ostream> +#include <string> +#include <type_traits> +#include <utility> +#include <vector> + +#include "common/status.h" +#include "core/column/column_fixed_length_object.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type.h" +#include "exec/common/arithmetic_overflow.h" +#include "format/parquet/parquet_common.h" +#include "format_v2/parquet/reader/native/decoder.h" +#include "util/bit_stream_utils.h" +#include "util/bit_stream_utils.inline.h" +#include "util/slice.h" + +namespace doris::format::parquet::native { +class DeltaDecoder : public Decoder { +public: + DeltaDecoder() = default; + ~DeltaDecoder() override = default; +}; + +/** + * Format + * [header] [block 1] [block 2] ... [block N] + * Header + * [block size] [_mini_blocks_per_block] [_total_value_count] [first value] + * Block + * [min delta] [list of bitwidths of the mini blocks] [miniblocks] + */ +template <typename T> +class DeltaBitPackDecoder final : public DeltaDecoder { +public: + using UT = std::make_unsigned_t<T>; + + DeltaBitPackDecoder() = default; + ~DeltaBitPackDecoder() override = default; + + Status skip_values(size_t num_values) override { + _values.resize(num_values); + uint32_t num_valid_values; + return _get_internal(_values.data(), cast_set<int32_t>(num_values), &num_valid_values); + } + + Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& consumer) override { + _values.resize(num_values); + uint32_t decoded_count = 0; + RETURN_IF_ERROR( + _get_internal(_values.data(), cast_set<uint32_t>(num_values), &decoded_count)); + if (UNLIKELY(decoded_count != num_values)) { + return Status::IOError("Expected {} Parquet delta values, decoded {}", num_values, + decoded_count); + } + return consumer.consume(reinterpret_cast<const uint8_t*>(_values.data()), num_values, + sizeof(T)); + } + + Status decode_selected_fixed_values(const ParquetSelection& selection, + ParquetFixedValueConsumer& consumer) override { + _values.resize(selection.total_values); + uint32_t decoded_count = 0; + RETURN_IF_ERROR(_get_internal(_values.data(), cast_set<uint32_t>(selection.total_values), + &decoded_count)); + if (UNLIKELY(decoded_count != selection.total_values)) { + return Status::IOError("Expected {} Parquet delta values, decoded {}", + selection.total_values, decoded_count); + } + size_t output = 0; + for (const auto& range : selection.ranges) { + memmove(_values.data() + output, _values.data() + range.first, range.count * sizeof(T)); + output += range.count; + } + DORIS_CHECK_EQ(output, selection.selected_values); + return consumer.consume(reinterpret_cast<const uint8_t*>(_values.data()), output, + sizeof(T)); + } + + Status decode(T* buffer, uint32_t num_values, uint32_t* out_num_values) { + return _get_internal(buffer, num_values, out_num_values); + } + + uint32_t valid_values_count() { + // _total_value_count in header ignores of null values + return _total_values_remaining; + } + + Status set_data(Slice* slice) override { + _bit_reader.reset( + new BitReader((const uint8_t*)slice->data, cast_set<uint32_t>(slice->size))); + RETURN_IF_ERROR(_init_header()); + _data = slice; + _offset = 0; + return Status::OK(); + } + + // Set BitReader which is already initialized by DeltaLengthByteArrayDecoder or + // DeltaByteArrayDecoder + Status set_bit_reader(std::shared_ptr<BitReader> bit_reader) { + _bit_reader = std::move(bit_reader); + RETURN_IF_ERROR(_init_header()); + return Status::OK(); + } + +private: + static constexpr int kMaxDeltaBitWidth = static_cast<int>(sizeof(T) * 8); + Status _init_header(); + Status _init_block(); + Status _init_mini_block(int bit_width); + Status _get_internal(T* buffer, uint32_t max_values, uint32_t* out_num_values); + + std::vector<T> _values; + + std::shared_ptr<BitReader> _bit_reader; + uint32_t _values_per_block; + uint32_t _mini_blocks_per_block; + uint32_t _values_per_mini_block; + uint32_t _total_value_count; + + T _min_delta; + T _last_value; + + uint32_t _mini_block_idx; + std::vector<uint8_t> _delta_bit_widths; + int _delta_bit_width; + // If the page doesn't contain any block, `_block_initialized` will + // always be false. Otherwise, it will be true when first block initialized. + bool _block_initialized; + + uint32_t _total_values_remaining; + // Remaining values in current mini block. If the current block is the last mini block, + // _values_remaining_current_mini_block may greater than _total_values_remaining. + uint32_t _values_remaining_current_mini_block; +}; + +class DeltaLengthByteArrayDecoder final : public DeltaDecoder { +public: + explicit DeltaLengthByteArrayDecoder() + : _len_decoder(), _buffered_length(0), _buffered_data(0) {} + + Status skip_values(size_t num_values) override { + _values.resize(num_values); + int num_valid_values; + return _get_internal(_values.data(), cast_set<int32_t>(num_values), &num_valid_values); + } + + Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer& consumer) override { + _values.resize(num_values); + int decoded_count = 0; + RETURN_IF_ERROR( + _get_internal(_values.data(), cast_set<int32_t>(num_values), &decoded_count)); + if (UNLIKELY(decoded_count != num_values)) { + return Status::IOError("Expected {} Parquet delta-length values, decoded {}", + num_values, decoded_count); + } + _string_refs.resize(num_values); + for (size_t row = 0; row < num_values; ++row) { + _string_refs[row] = StringRef(_values[row].data, _values[row].size); + } + return consumer.consume(_string_refs.data(), _string_refs.size()); + } + + Status decode_selected_binary_values(const ParquetSelection& selection, + ParquetBinaryValueConsumer& consumer) override { + _values.resize(selection.total_values); + int decoded_count = 0; + RETURN_IF_ERROR(_get_internal(_values.data(), cast_set<int32_t>(selection.total_values), + &decoded_count)); + if (UNLIKELY(decoded_count != selection.total_values)) { + return Status::IOError("Expected {} Parquet delta-length values, decoded {}", + selection.total_values, decoded_count); + } + _string_refs.resize(selection.selected_values); + size_t output = 0; + for (const auto& range : selection.ranges) { + for (size_t row = 0; row < range.count; ++row) { + const auto& value = _values[range.first + row]; + _string_refs[output++] = StringRef(value.data, value.size); + } + } + DORIS_CHECK_EQ(output, selection.selected_values); + return consumer.consume(_string_refs.data(), _string_refs.size()); + } + + Status decode(Slice* buffer, int num_values, int* out_num_values) { + return _get_internal(buffer, num_values, out_num_values); + } + + Status set_data(Slice* slice) override { + if (slice->size == 0) { + return Status::OK(); + } + _bit_reader = std::make_shared<BitReader>((const uint8_t*)slice->data, slice->size); + _data = slice; + _offset = 0; + RETURN_IF_ERROR(_decode_lengths()); + return Status::OK(); + } + + Status set_bit_reader(std::shared_ptr<BitReader> bit_reader) { + _bit_reader = std::move(bit_reader); + RETURN_IF_ERROR(_decode_lengths()); + return Status::OK(); + } + +private: + // Decode all the encoded lengths. The decoder_ will be at the start of the encoded data + // after that. + Status _decode_lengths(); + Status _get_internal(Slice* buffer, int max_values, int* out_num_values); + + std::vector<Slice> _values; + std::vector<StringRef> _string_refs; + std::shared_ptr<BitReader> _bit_reader; + DeltaBitPackDecoder<int32_t> _len_decoder; + + int _num_valid_values; + uint32_t _length_idx; + std::vector<int32_t> _buffered_length; + std::vector<char> _buffered_data; +}; + +class DeltaByteArrayDecoder : public DeltaDecoder { +public: + explicit DeltaByteArrayDecoder() : _buffered_prefix_length(0), _buffered_data(0) {} + + Status skip_values(size_t num_values) override { Review Comment: [P2] Validate every reconstructed FIXED_LEN_BYTE_ARRAY value before selection or skip. skip_values() calls _get_internal() without checking _type_length, and decode_selected_fixed_values() checks widths only inside selected ranges, so a three-byte value in a width-four column is accepted when filtered but rejected when selected. Exact decoded-count checks do not fix this. Validate all consumed slices and add selected-good/filtered-bad plus decoder-skip tests. ########## be/src/format_v2/parquet/reader/native/page_reader.h: ########## @@ -0,0 +1,261 @@ +// 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. + +#pragma once + +#include <gen_cpp/parquet_types.h> +#include <stdint.h> + +#include <memory> +#include <string> + +#include "common/cast_set.h" +#include "common/config.h" +#include "common/status.h" +#include "format/parquet/parquet_common.h" +#include "storage/cache/page_cache.h" +#include "util/block_compression.h" +namespace doris { +class BlockCompressionCodec; + +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io + +} // namespace doris + +namespace doris { +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io +struct Slice; +} // namespace doris + +namespace doris::format::parquet::native { +/** + * Use to deserialize parquet page header, and get the page data in iterator interface. + */ + +// Session-level options for parquet page reading/caching. +struct ParquetPageReadContext { + bool enable_parquet_file_page_cache = true; + ParquetPageReadContext() = default; + ParquetPageReadContext(bool enable_parquet_file_page_cache) + : enable_parquet_file_page_cache(enable_parquet_file_page_cache) {} +}; + +inline bool should_cache_decompressed(const tparquet::PageHeader* header, + const tparquet::ColumnMetaData& metadata) { + if (header->compressed_page_size <= 0) return true; Review Comment: [P1] Treat Data Page V2 with is_compressed=false as an uncompressed cache payload. The cold path correctly skips decompression and stores these bytes, but on a warm hit this helper ignores is_compressed; for a SNAPPY chunk with ratio 1 and a threshold below 1 it returns false, so load_page_data() sends already-uncompressed bytes to the SNAPPY decompressor. A valid file therefore succeeds cold and fails warm. Record the representation explicitly or honor the V2 flag, and add a compressed-codec/uncompressed-V2 cache-hit test. ########## be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.cpp: ########## @@ -0,0 +1,133 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/parquet/reader/native/delta_bit_pack_decoder.h" + +namespace doris::format::parquet::native { +Status DeltaLengthByteArrayDecoder::_decode_lengths() { + RETURN_IF_ERROR(_len_decoder.set_bit_reader(_bit_reader)); + // get the number of encoded lengths + int num_length = _len_decoder.valid_values_count(); + _buffered_length.resize(num_length); Review Comment: [P1] Bound delta counts before resizing from them. valid_values_count() comes from the payload's ULEB32 header; INT_MAX here makes this tiny malformed page request roughly 8 GiB before any block bytes are checked (and values above INT_MAX first narrow to a negative int). Decoded lengths and miniblock counts have the same allocate-before-remaining-bytes pattern. Validate counts against the page's expected non-null values and aggregate lengths against remaining payload bytes, using checked unsigned types, before allocating. -- 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]
