github-actions[bot] commented on code in PR #65674: URL: https://github.com/apache/doris/pull/65674#discussion_r3600435759
########## be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp: ########## @@ -0,0 +1,1015 @@ +// 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_chunk_reader.h" + +#include <gen_cpp/parquet_types.h> +#include <glog/logging.h> +#include <string.h> + +#include <cstdint> +#include <memory> +#include <utility> + +#include "common/compiler_util.h" // IWYU pragma: keep +#include "core/column/column.h" +#include "core/custom_allocator.h" +#include "core/data_type_serde/data_type_serde.h" +#include "format/parquet/schema_desc.h" +#include "format_v2/parquet/reader/native/decoder.h" +#include "format_v2/parquet/reader/native/level_decoder.h" +#include "format_v2/parquet/reader/native/page_reader.h" +#include "io/fs/buffered_reader.h" +#include "runtime/runtime_profile.h" +#include "storage/cache/page_cache.h" +#include "util/bit_util.h" +#include "util/block_compression.h" + +namespace cctz { +class time_zone; +} // namespace cctz +namespace doris { +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io +} // namespace doris + +namespace doris::format::parquet::native { +namespace { + +Status translate_value_encoding(tparquet::Encoding::type encoding, + ParquetValueEncoding* translated) { + DORIS_CHECK(translated != nullptr); + switch (encoding) { + case tparquet::Encoding::PLAIN: + *translated = ParquetValueEncoding::PLAIN; + return Status::OK(); + case tparquet::Encoding::RLE_DICTIONARY: + case tparquet::Encoding::PLAIN_DICTIONARY: + *translated = ParquetValueEncoding::DICTIONARY; + return Status::OK(); + case tparquet::Encoding::RLE: + *translated = ParquetValueEncoding::RLE; + return Status::OK(); + case tparquet::Encoding::BIT_PACKED: + *translated = ParquetValueEncoding::BIT_PACKED; + return Status::OK(); + case tparquet::Encoding::DELTA_BINARY_PACKED: + *translated = ParquetValueEncoding::DELTA_BINARY_PACKED; + return Status::OK(); + case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY: + *translated = ParquetValueEncoding::DELTA_LENGTH_BYTE_ARRAY; + return Status::OK(); + case tparquet::Encoding::DELTA_BYTE_ARRAY: + *translated = ParquetValueEncoding::DELTA_BYTE_ARRAY; + return Status::OK(); + case tparquet::Encoding::BYTE_STREAM_SPLIT: + *translated = ParquetValueEncoding::BYTE_STREAM_SPLIT; + return Status::OK(); + default: + return Status::NotSupported("Unsupported Parquet encoding {}", + tparquet::to_string(encoding)); + } +} + +template <bool HAS_FILTER> +Status decode_selected_values(IColumn& column, const DataTypeSerDe& serde, Decoder& decoder, + const ParquetDecodeContext& context, + ParquetMaterializationState& state, ColumnSelectVector& select_vector, + int64_t* materialization_time) { + SCOPED_RAW_TIMER(materialization_time); + ColumnSelectVector::DataReadType read_type; + while (const size_t run_length = select_vector.get_next_run<HAS_FILTER>(&read_type)) { + switch (read_type) { + case ColumnSelectVector::CONTENT: + RETURN_IF_ERROR( + serde.read_column_from_parquet(column, decoder, context, run_length, state)); + break; + case ColumnSelectVector::NULL_DATA: + column.insert_many_defaults(run_length); + break; + case ColumnSelectVector::FILTERED_CONTENT: + RETURN_IF_ERROR(decoder.skip_values(run_length)); + break; + case ColumnSelectVector::FILTERED_NULL: + break; + } + } + return Status::OK(); +} + +// Presents one sparse page request as an ordinary sequential source to DataTypeSerDe. SerDe is +// entered once per page fragment; the concrete decoder decides whether to gather selected spans, +// batch-decode and compact, or use the cursor-preserving range fallback. +class SelectedDecodeSource final : public ParquetDecodeSource { +public: + SelectedDecodeSource(Decoder& decoder, const ParquetSelection& selection) + : _decoder(decoder), _selection(selection) {} + + Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& consumer) override { + DORIS_CHECK_EQ(num_values, _selection.selected_values); + return _decoder.decode_selected_fixed_values(_selection, consumer); + } + + Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer& consumer) override { + DORIS_CHECK_EQ(num_values, _selection.selected_values); + return _decoder.decode_selected_binary_values(_selection, consumer); + } + + Status skip_values(size_t num_values) override { + return Status::NotSupported("Selected Parquet source cannot be skipped, values={}", + num_values); + } + + bool has_dictionary() const override { return _decoder.has_dictionary(); } + uint64_t dictionary_generation() const override { return _decoder.dictionary_generation(); } + size_t dictionary_size() const override { return _decoder.dictionary_size(); } + + Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer, + ParquetBinaryValueConsumer& binary_consumer) override { + return _decoder.decode_dictionary(fixed_consumer, binary_consumer); + } + + Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>* indices) override { + DORIS_CHECK_EQ(num_values, _selection.selected_values); + return _decoder.decode_selected_dictionary_indices(_selection, indices); + } + +private: + Decoder& _decoder; + const ParquetSelection& _selection; +}; + +Status decode_selected_non_null_values(IColumn& column, const DataTypeSerDe& serde, + Decoder& decoder, const ParquetDecodeContext& context, + ParquetMaterializationState& state, + ColumnSelectVector& select_vector, + int64_t* materialization_time) { + auto& selection = state.selection; + selection.ranges.clear(); + selection.total_values = select_vector.num_values(); + selection.selected_values = 0; + + size_t cursor = 0; + ColumnSelectVector::DataReadType read_type; + while (const size_t run_length = select_vector.get_next_run<true>(&read_type)) { + DORIS_CHECK(read_type == ColumnSelectVector::CONTENT || + read_type == ColumnSelectVector::FILTERED_CONTENT); + if (read_type == ColumnSelectVector::CONTENT) { + selection.ranges.push_back({.first = cursor, .count = run_length}); + selection.selected_values += run_length; + } + cursor += run_length; + } + DORIS_CHECK_EQ(cursor, selection.total_values); + if (selection.selected_values == 0) { + return decoder.skip_values(selection.total_values); + } + + SCOPED_RAW_TIMER(materialization_time); + SelectedDecodeSource selected_source(decoder, selection); + return serde.read_column_from_parquet(column, selected_source, context, + selection.selected_values, state); +} + +} // namespace + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::ColumnChunkReader( + io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk, + FieldSchema* field_schema, const tparquet::OffsetIndex* offset_index, size_t total_rows, + io::IOContext* io_ctx, const ParquetPageReadContext& page_read_ctx) + : _field_schema(field_schema), + _max_rep_level(field_schema->repetition_level), + _max_def_level(field_schema->definition_level), + _stream_reader(reader), + _metadata(column_chunk->meta_data), + _offset_index(offset_index), + _total_rows(total_rows), + _io_ctx(io_ctx), + _page_read_ctx(page_read_ctx) {} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() { + size_t start_offset = has_dict_page(_metadata) ? _metadata.dictionary_page_offset + : _metadata.data_page_offset; + size_t chunk_size = _metadata.total_compressed_size; + // create page reader + _page_reader = create_page_reader<IN_COLLECTION, OFFSET_INDEX>( + _stream_reader, _io_ctx, start_offset, chunk_size, _total_rows, _metadata, + _page_read_ctx, _offset_index); + // get the block compression codec + RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec, &_block_compress_codec)); + _state = INITIALIZED; + RETURN_IF_ERROR(_parse_first_page_header()); + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_nested_values( + const std::vector<level_t>& def_levels, size_t start_index) { + size_t no_value_cnt = 0; + size_t value_cnt = 0; + + DORIS_CHECK(start_index <= def_levels.size()); + for (size_t idx = start_index; idx < def_levels.size(); idx++) { + level_t def_level = def_levels[idx]; + if (IN_COLLECTION && def_level < _field_schema->repeated_parent_def_level) { + no_value_cnt++; + } else if (def_level < _field_schema->definition_level) { + no_value_cnt++; + } else { + value_cnt++; + } + } + + RETURN_IF_ERROR(skip_values(value_cnt, true)); + RETURN_IF_ERROR(skip_values(no_value_cnt, false)); + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::read_levels( + size_t num_values, std::vector<level_t>* rep_levels, std::vector<level_t>* def_levels) { + DORIS_CHECK(rep_levels != nullptr); + DORIS_CHECK(def_levels != nullptr); + if (_remaining_num_values < num_values || _remaining_rep_nums < num_values || + _remaining_def_nums < num_values) { + return Status::Corruption( + "Parquet level reader requested {} slots with only {}/{}/{} remaining", num_values, + _remaining_num_values, _remaining_rep_nums, _remaining_def_nums); + } + + const size_t start_index = def_levels->size(); + rep_levels->resize(rep_levels->size() + num_values, 0); + def_levels->resize(def_levels->size() + num_values, 0); + if (_max_rep_level > 0) { + const size_t decoded = _rep_level_decoder.get_levels( + rep_levels->data() + rep_levels->size() - num_values, num_values); + if (decoded != num_values) { + return Status::Corruption("Parquet repetition level stream ended after {} of {} slots", + decoded, num_values); + } + } + if (_max_def_level > 0) { + const size_t decoded = _def_level_decoder.get_levels( + def_levels->data() + def_levels->size() - num_values, num_values); + if (decoded != num_values) { + return Status::Corruption("Parquet definition level stream ended after {} of {} slots", + decoded, num_values); + } + } + _remaining_rep_nums -= num_values; + _remaining_def_nums -= num_values; + return skip_nested_values(*def_levels, start_index); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_parse_first_page_header() { + RETURN_IF_ERROR(parse_page_header()); + + const tparquet::PageHeader* header = nullptr; + RETURN_IF_ERROR(_page_reader->get_page_header(&header)); + if (header->type == tparquet::PageType::DICTIONARY_PAGE) { + // the first page maybe directory page even if _metadata.__isset.dictionary_page_offset == false, + // so we should parse the directory page in next_page() + RETURN_IF_ERROR(_decode_dict_page()); + // parse the real first data page + RETURN_IF_ERROR(_page_reader->dict_next_page()); + _state = INITIALIZED; + } + + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() { + if (_state == HEADER_PARSED || _state == DATA_LOADED) { + return Status::OK(); + } + RETURN_IF_ERROR(_page_reader->parse_page_header()); + + const tparquet::PageHeader* header = nullptr; + RETURN_IF_ERROR(_page_reader->get_page_header(&header)); + int32_t page_num_values = _page_reader->is_header_v2() ? header->data_page_header_v2.num_values Review Comment: [P2] Skip non-data pages before initializing the data decoders. `INDEX_PAGE` is a valid Parquet page type, and the Arrow reader this path replaces explicitly loops past unknown/non-data pages. Here any non-V2 page that is not the first dictionary page is treated as `data_page_header`; for a nested sequential scan, `seek_to_nested_row()` then feeds the INDEX_PAGE payload into the repetition/definition/value decoders, so a previously readable file fails or is misdecoded. Loop over page headers until DATA_PAGE/DATA_PAGE_V2 (while still handling the dictionary page), and add a nested column chunk containing an INDEX_PAGE. ########## be/src/format_v2/parquet/parquet_file_context.cpp: ########## @@ -568,6 +606,51 @@ Status ParquetFileContext::open(io::FileReaderSPtr input_file_reader, io::IOCont return Status::OK(); } +Status ParquetFileContext::load_native_offset_indexes( + int row_group_id, const std::unordered_set<int>& leaf_column_ids, + std::unordered_map<int, tparquet::OffsetIndex>* offset_indexes) const { + DORIS_CHECK(offset_indexes != nullptr); + offset_indexes->clear(); + if (leaf_column_ids.empty() || file_reader == nullptr) { + return Status::OK(); + } + try { + auto page_index_reader = file_reader->GetPageIndexReader(); + if (page_index_reader == nullptr) { + return Status::OK(); + } + auto row_group_reader = page_index_reader->RowGroup(row_group_id); + if (row_group_reader == nullptr) { + return Status::OK(); + } + for (const int leaf_column_id : leaf_column_ids) { + auto arrow_index = row_group_reader->GetOffsetIndex(leaf_column_id); + if (arrow_index == nullptr || arrow_index->page_locations().empty()) { + // An empty optional index is equivalent to no index. Publishing it would select + // the indexed PageReader even though there is no first page to dereference. + continue; + } + tparquet::OffsetIndex native_index; + native_index.page_locations.reserve(arrow_index->page_locations().size()); + for (const auto& arrow_location : arrow_index->page_locations()) { Review Comment: [P1] Validate every nonempty OffsetIndex before selecting the indexed reader. Arrow only deserializes these PageLocations; it does not enforce their required row/offset ordering or chunk bounds. For example, locations `(page1_offset, 0)` and `(page1_offset, K)` make `next_page()` label page 1 as rows K onward, so a later selected range silently returns the first page's values again while the in-chunk header check still passes. Require first row 0, strictly increasing in-range row indexes and offsets, positive sizes, and page ranges within the owning chunk/file; discard the whole optional index or return corruption on failure, with a backward/repeated-offset test. ########## be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp: ########## @@ -0,0 +1,1015 @@ +// 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_chunk_reader.h" + +#include <gen_cpp/parquet_types.h> +#include <glog/logging.h> +#include <string.h> + +#include <cstdint> +#include <memory> +#include <utility> + +#include "common/compiler_util.h" // IWYU pragma: keep +#include "core/column/column.h" +#include "core/custom_allocator.h" +#include "core/data_type_serde/data_type_serde.h" +#include "format/parquet/schema_desc.h" +#include "format_v2/parquet/reader/native/decoder.h" +#include "format_v2/parquet/reader/native/level_decoder.h" +#include "format_v2/parquet/reader/native/page_reader.h" +#include "io/fs/buffered_reader.h" +#include "runtime/runtime_profile.h" +#include "storage/cache/page_cache.h" +#include "util/bit_util.h" +#include "util/block_compression.h" + +namespace cctz { +class time_zone; +} // namespace cctz +namespace doris { +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io +} // namespace doris + +namespace doris::format::parquet::native { +namespace { + +Status translate_value_encoding(tparquet::Encoding::type encoding, + ParquetValueEncoding* translated) { + DORIS_CHECK(translated != nullptr); + switch (encoding) { + case tparquet::Encoding::PLAIN: + *translated = ParquetValueEncoding::PLAIN; + return Status::OK(); + case tparquet::Encoding::RLE_DICTIONARY: + case tparquet::Encoding::PLAIN_DICTIONARY: + *translated = ParquetValueEncoding::DICTIONARY; + return Status::OK(); + case tparquet::Encoding::RLE: + *translated = ParquetValueEncoding::RLE; + return Status::OK(); + case tparquet::Encoding::BIT_PACKED: + *translated = ParquetValueEncoding::BIT_PACKED; + return Status::OK(); + case tparquet::Encoding::DELTA_BINARY_PACKED: + *translated = ParquetValueEncoding::DELTA_BINARY_PACKED; + return Status::OK(); + case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY: + *translated = ParquetValueEncoding::DELTA_LENGTH_BYTE_ARRAY; + return Status::OK(); + case tparquet::Encoding::DELTA_BYTE_ARRAY: + *translated = ParquetValueEncoding::DELTA_BYTE_ARRAY; + return Status::OK(); + case tparquet::Encoding::BYTE_STREAM_SPLIT: + *translated = ParquetValueEncoding::BYTE_STREAM_SPLIT; + return Status::OK(); + default: + return Status::NotSupported("Unsupported Parquet encoding {}", + tparquet::to_string(encoding)); + } +} + +template <bool HAS_FILTER> +Status decode_selected_values(IColumn& column, const DataTypeSerDe& serde, Decoder& decoder, + const ParquetDecodeContext& context, + ParquetMaterializationState& state, ColumnSelectVector& select_vector, + int64_t* materialization_time) { + SCOPED_RAW_TIMER(materialization_time); + ColumnSelectVector::DataReadType read_type; + while (const size_t run_length = select_vector.get_next_run<HAS_FILTER>(&read_type)) { + switch (read_type) { + case ColumnSelectVector::CONTENT: + RETURN_IF_ERROR( + serde.read_column_from_parquet(column, decoder, context, run_length, state)); + break; + case ColumnSelectVector::NULL_DATA: + column.insert_many_defaults(run_length); + break; + case ColumnSelectVector::FILTERED_CONTENT: + RETURN_IF_ERROR(decoder.skip_values(run_length)); + break; + case ColumnSelectVector::FILTERED_NULL: + break; + } + } + return Status::OK(); +} + +// Presents one sparse page request as an ordinary sequential source to DataTypeSerDe. SerDe is +// entered once per page fragment; the concrete decoder decides whether to gather selected spans, +// batch-decode and compact, or use the cursor-preserving range fallback. +class SelectedDecodeSource final : public ParquetDecodeSource { +public: + SelectedDecodeSource(Decoder& decoder, const ParquetSelection& selection) + : _decoder(decoder), _selection(selection) {} + + Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& consumer) override { + DORIS_CHECK_EQ(num_values, _selection.selected_values); + return _decoder.decode_selected_fixed_values(_selection, consumer); + } + + Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer& consumer) override { + DORIS_CHECK_EQ(num_values, _selection.selected_values); + return _decoder.decode_selected_binary_values(_selection, consumer); + } + + Status skip_values(size_t num_values) override { + return Status::NotSupported("Selected Parquet source cannot be skipped, values={}", + num_values); + } + + bool has_dictionary() const override { return _decoder.has_dictionary(); } + uint64_t dictionary_generation() const override { return _decoder.dictionary_generation(); } + size_t dictionary_size() const override { return _decoder.dictionary_size(); } + + Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer, + ParquetBinaryValueConsumer& binary_consumer) override { + return _decoder.decode_dictionary(fixed_consumer, binary_consumer); + } + + Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>* indices) override { + DORIS_CHECK_EQ(num_values, _selection.selected_values); + return _decoder.decode_selected_dictionary_indices(_selection, indices); + } + +private: + Decoder& _decoder; + const ParquetSelection& _selection; +}; + +Status decode_selected_non_null_values(IColumn& column, const DataTypeSerDe& serde, + Decoder& decoder, const ParquetDecodeContext& context, + ParquetMaterializationState& state, + ColumnSelectVector& select_vector, + int64_t* materialization_time) { + auto& selection = state.selection; + selection.ranges.clear(); + selection.total_values = select_vector.num_values(); + selection.selected_values = 0; + + size_t cursor = 0; + ColumnSelectVector::DataReadType read_type; + while (const size_t run_length = select_vector.get_next_run<true>(&read_type)) { + DORIS_CHECK(read_type == ColumnSelectVector::CONTENT || + read_type == ColumnSelectVector::FILTERED_CONTENT); + if (read_type == ColumnSelectVector::CONTENT) { + selection.ranges.push_back({.first = cursor, .count = run_length}); + selection.selected_values += run_length; + } + cursor += run_length; + } + DORIS_CHECK_EQ(cursor, selection.total_values); + if (selection.selected_values == 0) { + return decoder.skip_values(selection.total_values); + } + + SCOPED_RAW_TIMER(materialization_time); + SelectedDecodeSource selected_source(decoder, selection); + return serde.read_column_from_parquet(column, selected_source, context, + selection.selected_values, state); +} + +} // namespace + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::ColumnChunkReader( + io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk, + FieldSchema* field_schema, const tparquet::OffsetIndex* offset_index, size_t total_rows, + io::IOContext* io_ctx, const ParquetPageReadContext& page_read_ctx) + : _field_schema(field_schema), + _max_rep_level(field_schema->repetition_level), + _max_def_level(field_schema->definition_level), + _stream_reader(reader), + _metadata(column_chunk->meta_data), + _offset_index(offset_index), + _total_rows(total_rows), + _io_ctx(io_ctx), + _page_read_ctx(page_read_ctx) {} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() { + size_t start_offset = has_dict_page(_metadata) ? _metadata.dictionary_page_offset + : _metadata.data_page_offset; + size_t chunk_size = _metadata.total_compressed_size; + // create page reader + _page_reader = create_page_reader<IN_COLLECTION, OFFSET_INDEX>( + _stream_reader, _io_ctx, start_offset, chunk_size, _total_rows, _metadata, + _page_read_ctx, _offset_index); + // get the block compression codec + RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec, &_block_compress_codec)); + _state = INITIALIZED; + RETURN_IF_ERROR(_parse_first_page_header()); + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_nested_values( + const std::vector<level_t>& def_levels, size_t start_index) { + size_t no_value_cnt = 0; + size_t value_cnt = 0; + + DORIS_CHECK(start_index <= def_levels.size()); + for (size_t idx = start_index; idx < def_levels.size(); idx++) { + level_t def_level = def_levels[idx]; + if (IN_COLLECTION && def_level < _field_schema->repeated_parent_def_level) { + no_value_cnt++; + } else if (def_level < _field_schema->definition_level) { + no_value_cnt++; + } else { + value_cnt++; + } + } + + RETURN_IF_ERROR(skip_values(value_cnt, true)); + RETURN_IF_ERROR(skip_values(no_value_cnt, false)); + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::read_levels( + size_t num_values, std::vector<level_t>* rep_levels, std::vector<level_t>* def_levels) { + DORIS_CHECK(rep_levels != nullptr); + DORIS_CHECK(def_levels != nullptr); + if (_remaining_num_values < num_values || _remaining_rep_nums < num_values || + _remaining_def_nums < num_values) { + return Status::Corruption( + "Parquet level reader requested {} slots with only {}/{}/{} remaining", num_values, + _remaining_num_values, _remaining_rep_nums, _remaining_def_nums); + } + + const size_t start_index = def_levels->size(); + rep_levels->resize(rep_levels->size() + num_values, 0); + def_levels->resize(def_levels->size() + num_values, 0); + if (_max_rep_level > 0) { + const size_t decoded = _rep_level_decoder.get_levels( + rep_levels->data() + rep_levels->size() - num_values, num_values); + if (decoded != num_values) { + return Status::Corruption("Parquet repetition level stream ended after {} of {} slots", + decoded, num_values); + } + } + if (_max_def_level > 0) { + const size_t decoded = _def_level_decoder.get_levels( + def_levels->data() + def_levels->size() - num_values, num_values); + if (decoded != num_values) { + return Status::Corruption("Parquet definition level stream ended after {} of {} slots", + decoded, num_values); + } + } + _remaining_rep_nums -= num_values; + _remaining_def_nums -= num_values; + return skip_nested_values(*def_levels, start_index); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_parse_first_page_header() { + RETURN_IF_ERROR(parse_page_header()); + + const tparquet::PageHeader* header = nullptr; + RETURN_IF_ERROR(_page_reader->get_page_header(&header)); + if (header->type == tparquet::PageType::DICTIONARY_PAGE) { + // the first page maybe directory page even if _metadata.__isset.dictionary_page_offset == false, + // so we should parse the directory page in next_page() + RETURN_IF_ERROR(_decode_dict_page()); + // parse the real first data page + RETURN_IF_ERROR(_page_reader->dict_next_page()); + _state = INITIALIZED; + } + + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() { + if (_state == HEADER_PARSED || _state == DATA_LOADED) { + return Status::OK(); + } + RETURN_IF_ERROR(_page_reader->parse_page_header()); + + const tparquet::PageHeader* header = nullptr; + RETURN_IF_ERROR(_page_reader->get_page_header(&header)); + int32_t page_num_values = _page_reader->is_header_v2() ? header->data_page_header_v2.num_values + : header->data_page_header.num_values; + _remaining_rep_nums = page_num_values; + _remaining_def_nums = page_num_values; + _remaining_num_values = page_num_values; + + // no offset will parse all header. + if constexpr (OFFSET_INDEX == false) { + _chunk_parsed_values += _remaining_num_values; + } + _state = HEADER_PARSED; + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() { + _state = INITIALIZED; + RETURN_IF_ERROR(_page_reader->next_page()); + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_get_uncompressed_levels( + const tparquet::DataPageHeaderV2& page_v2, Slice& page_data) { + const size_t rl = page_v2.repetition_levels_byte_length; + const size_t dl = page_v2.definition_levels_byte_length; + if (UNLIKELY(rl > page_data.size || dl > page_data.size - rl)) { + // Validate the physical slice again because a cached entry may itself be truncated. + return Status::Corruption("Parquet data page v2 level bytes exceed available payload"); + } + _v2_rep_levels = Slice(page_data.data, rl); + _v2_def_levels = Slice(page_data.data + rl, dl); + page_data.data += dl + rl; + page_data.size -= dl + rl; + return Status::OK(); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() { + if (_state == DATA_LOADED) { + return Status::OK(); + } + if (UNLIKELY(_state != HEADER_PARSED)) { + return Status::Corruption("Should parse page header"); + } + + const tparquet::PageHeader* header = nullptr; + RETURN_IF_ERROR(_page_reader->get_page_header(&header)); + int32_t uncompressed_size = header->uncompressed_page_size; + bool page_loaded = false; + + // First, try to reuse a cache handle previously discovered by PageReader + // (header-only lookup) to avoid a second lookup here. + if (_page_read_ctx.enable_parquet_file_page_cache && !config::disable_storage_page_cache && + StoragePageCache::instance() != nullptr) { + if (_page_reader->has_page_cache_handle()) { + const PageCacheHandle& handle = _page_reader->page_cache_handle(); + Slice cached = handle.data(); + size_t header_size = _page_reader->header_bytes().size(); + size_t levels_size = 0; + if (header->__isset.data_page_header_v2) { + const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2; + size_t rl = header_v2.repetition_levels_byte_length; + size_t dl = header_v2.definition_levels_byte_length; + levels_size = rl + dl; + if (UNLIKELY(header_size > cached.size || + levels_size > cached.size - header_size)) { + return Status::Corruption("Cached Parquet page is shorter than its v2 levels"); + } + _v2_rep_levels = + Slice(reinterpret_cast<const uint8_t*>(cached.data) + header_size, rl); + _v2_def_levels = + Slice(reinterpret_cast<const uint8_t*>(cached.data) + header_size + rl, dl); + } + // payload_slice points to the bytes after header and levels + if (UNLIKELY(header_size + levels_size > cached.size)) { + return Status::Corruption("Cached Parquet page is shorter than its header"); + } + Slice payload_slice(cached.data + header_size + levels_size, + cached.size - header_size - levels_size); + + bool cache_payload_is_decompressed = _page_reader->is_cache_payload_decompressed(); + const size_t expected_payload_size = + cache_payload_is_decompressed + ? static_cast<size_t>(header->uncompressed_page_size) - levels_size + : static_cast<size_t>(header->compressed_page_size) - levels_size; + if (UNLIKELY(payload_slice.size != expected_payload_size)) { + return Status::Corruption("Cached Parquet page payload has size {}, expected {}", + payload_slice.size, expected_payload_size); + } + + if (cache_payload_is_decompressed) { + // Cached payload is already uncompressed + _page_data = payload_slice; + } else { + CHECK(_block_compress_codec); + // Decompress cached payload into _decompress_buf for decoding + size_t uncompressed_payload_size = + header->__isset.data_page_header_v2 + ? static_cast<size_t>(header->uncompressed_page_size) - levels_size + : static_cast<size_t>(header->uncompressed_page_size); + _reserve_decompress_buf(uncompressed_payload_size); + _page_data = Slice(_decompress_buf.get(), uncompressed_payload_size); + SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time); + _chunk_statistics.decompress_cnt++; + RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &_page_data)); + if (UNLIKELY(_page_data.size != uncompressed_payload_size)) { + return Status::Corruption("Parquet page decompressed to {} bytes, expected {}", + _page_data.size, uncompressed_payload_size); + } + } + // page cache counters were incremented when PageReader did the header-only + // cache lookup. Do not increment again to avoid double-counting. + page_loaded = true; + } + } + + if (!page_loaded) { + if (_block_compress_codec != nullptr) { + Slice compressed_data; + RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data)); + std::vector<uint8_t> level_bytes; + if (header->__isset.data_page_header_v2) { + const tparquet::DataPageHeaderV2& header_v2 = header->data_page_header_v2; + // uncompressed_size = rl + dl + uncompressed_data_size + // compressed_size = rl + dl + compressed_data_size + uncompressed_size -= header_v2.repetition_levels_byte_length + + header_v2.definition_levels_byte_length; + // copy level bytes (rl + dl) so that we can cache header + levels + uncompressed payload + size_t rl = header_v2.repetition_levels_byte_length; + size_t dl = header_v2.definition_levels_byte_length; + size_t level_sz = rl + dl; + if (level_sz > 0) { + level_bytes.resize(level_sz); + memcpy(level_bytes.data(), compressed_data.data, level_sz); + } + // now remove levels from compressed_data for decompression + RETURN_IF_ERROR(_get_uncompressed_levels(header_v2, compressed_data)); + } + bool is_v2_compressed = header->__isset.data_page_header_v2 && + header->data_page_header_v2.is_compressed; Review Comment: [P1] Preserve the pre-Arrow-3 DataPageV2 compression override. The Arrow page reader this replaces checks the file's writer version and forces decompression for writers older than `PARQUET_CPP_10353_FIXED_VERSION`, because those Arrow versions always set `is_compressed=false` even when the V2 value payload was compressed. This native path has no writer-version override and therefore sends those compressed bytes directly to the value decoder (and classifies them as decompressed for cache purposes), breaking every compressed V2 page from the affected writers. Thread the legacy `always_compressed` decision through the native cold/cache paths and add an old-Arrow fixture. ########## be/src/format_v2/parquet/reader/native/column_reader.cpp: ########## @@ -0,0 +1,1529 @@ +// 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 <gen_cpp/parquet_types.h> +#include <limits.h> +#include <sys/types.h> + +#include <algorithm> +#include <utility> + +#include "common/cast_set.h" +#include "common/status.h" +#include "core/column/column.h" +#include "core/column/column_array.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.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_struct.h" +#include "core/data_type/define_primitive_type.h" +#include "format/parquet/schema_desc.h" +#include "format_v2/parquet/reader/native/column_chunk_reader.h" +#include "format_v2/parquet/reader/native/level_decoder.h" +#include "io/fs/tracing_file_reader.h" +#include "runtime/runtime_profile.h" + +namespace doris::format::parquet::native { +namespace { + +ParquetTimeUnit parquet_time_unit(const tparquet::TimeUnit& unit) { + if (unit.__isset.MILLIS) { + return ParquetTimeUnit::MILLIS; + } + if (unit.__isset.MICROS) { + return ParquetTimeUnit::MICROS; + } + if (unit.__isset.NANOS) { + return ParquetTimeUnit::NANOS; + } + return ParquetTimeUnit::UNKNOWN; +} + +bool is_direct_integer_type(PrimitiveType type) { + switch (type) { + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + return true; + default: + return false; + } +} + +bool is_direct_decimal_type(PrimitiveType type) { + switch (type) { + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + return true; + default: + return false; + } +} + +template <typename T> +bool release_vector_if_oversized(std::vector<T>* values, size_t max_retained_bytes) { + DORIS_CHECK(values != nullptr); + if (values->capacity() * sizeof(T) <= max_retained_bytes) { + return false; + } + std::vector<T>().swap(*values); + return true; +} + +size_t retained_set_bytes(const std::unordered_set<size_t>& values) { + return values.bucket_count() * sizeof(void*) + values.size() * sizeof(size_t); +} + +bool is_direct_binary_type(PrimitiveType type) { + return is_string_type(type) || type == TYPE_VARBINARY; +} + +// The target SerDe can fuse physical decode with these logical type changes. Less common schema +// changes retain the generic file-format converter as a compatibility path: the decoder still +// exposes raw spans, but the source SerDe first materializes a reusable source column before the +// generic logical cast. Ordinary scans and numeric widening never allocate that source column. +bool serde_can_materialize_directly(const DataTypePtr& source_type, + const DataTypePtr& target_type) { + const auto source = remove_nullable(source_type)->get_primitive_type(); + const auto target = remove_nullable(target_type)->get_primitive_type(); + return source == target || (is_direct_integer_type(source) && is_direct_integer_type(target)) || Review Comment: [P1] Preserve non-strict nullable conversion failures in the direct path. The removed reader passed `enable_strict_mode` through `DecodedColumnView`, and `DataTypeNullableSerDe` let the numeric/date/time/decimal SerDes mark an out-of-range value NULL when strict mode was off. This path unwraps nullable, routes all integer conversions directly to consumers that always return `DataQualityError`, and carries neither the strict flag nor a failure null-map; the scheduler's `_enable_strict_mode` is now write-only. For example, Nullable<TINYINT> reading Parquet INT32 value 200 changes from NULL to a failed scan in non-strict mode. Thread conversion-failure state through native materialization and cover strict/non-strict nullable overflow (plus date/time/decimal). ########## be/src/format_v2/parquet/reader/native/column_reader.cpp: ########## @@ -0,0 +1,1529 @@ +// 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 <gen_cpp/parquet_types.h> +#include <limits.h> +#include <sys/types.h> + +#include <algorithm> +#include <utility> + +#include "common/cast_set.h" +#include "common/status.h" +#include "core/column/column.h" +#include "core/column/column_array.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.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_struct.h" +#include "core/data_type/define_primitive_type.h" +#include "format/parquet/schema_desc.h" +#include "format_v2/parquet/reader/native/column_chunk_reader.h" +#include "format_v2/parquet/reader/native/level_decoder.h" +#include "io/fs/tracing_file_reader.h" +#include "runtime/runtime_profile.h" + +namespace doris::format::parquet::native { +namespace { + +ParquetTimeUnit parquet_time_unit(const tparquet::TimeUnit& unit) { + if (unit.__isset.MILLIS) { + return ParquetTimeUnit::MILLIS; + } + if (unit.__isset.MICROS) { + return ParquetTimeUnit::MICROS; + } + if (unit.__isset.NANOS) { + return ParquetTimeUnit::NANOS; + } + return ParquetTimeUnit::UNKNOWN; +} + +bool is_direct_integer_type(PrimitiveType type) { + switch (type) { + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + return true; + default: + return false; + } +} + +bool is_direct_decimal_type(PrimitiveType type) { + switch (type) { + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + return true; + default: + return false; + } +} + +template <typename T> +bool release_vector_if_oversized(std::vector<T>* values, size_t max_retained_bytes) { + DORIS_CHECK(values != nullptr); + if (values->capacity() * sizeof(T) <= max_retained_bytes) { + return false; + } + std::vector<T>().swap(*values); + return true; +} + +size_t retained_set_bytes(const std::unordered_set<size_t>& values) { + return values.bucket_count() * sizeof(void*) + values.size() * sizeof(size_t); +} + +bool is_direct_binary_type(PrimitiveType type) { + return is_string_type(type) || type == TYPE_VARBINARY; +} + +// The target SerDe can fuse physical decode with these logical type changes. Less common schema +// changes retain the generic file-format converter as a compatibility path: the decoder still +// exposes raw spans, but the source SerDe first materializes a reusable source column before the +// generic logical cast. Ordinary scans and numeric widening never allocate that source column. +bool serde_can_materialize_directly(const DataTypePtr& source_type, + const DataTypePtr& target_type) { + const auto source = remove_nullable(source_type)->get_primitive_type(); + const auto target = remove_nullable(target_type)->get_primitive_type(); + return source == target || (is_direct_integer_type(source) && is_direct_integer_type(target)) || + (source == TYPE_FLOAT && target == TYPE_DOUBLE) || + (is_direct_decimal_type(source) && is_direct_decimal_type(target)) || + // Parquet STRING and VARBINARY share BYTE_ARRAY bytes. Materializing through the target + // SerDe preserves those bytes and avoids a converter whose scratch column uses the v1 + // String representation instead of the native ColumnVarbinary representation. + (is_direct_binary_type(source) && is_direct_binary_type(target)); +} + +Status init_decode_context(const FieldSchema& field, const cctz::time_zone* ctz, + ParquetDecodeContext* context) { + DORIS_CHECK(context != nullptr); + switch (field.physical_type) { + case tparquet::Type::BOOLEAN: + context->physical_type = ParquetPhysicalType::BOOLEAN; + break; + case tparquet::Type::INT32: + context->physical_type = ParquetPhysicalType::INT32; + break; + case tparquet::Type::INT64: + context->physical_type = ParquetPhysicalType::INT64; + break; + case tparquet::Type::INT96: + context->physical_type = ParquetPhysicalType::INT96; + break; + case tparquet::Type::FLOAT: + context->physical_type = ParquetPhysicalType::FLOAT; + break; + case tparquet::Type::DOUBLE: + context->physical_type = ParquetPhysicalType::DOUBLE; + break; + case tparquet::Type::BYTE_ARRAY: + context->physical_type = ParquetPhysicalType::BYTE_ARRAY; + break; + case tparquet::Type::FIXED_LEN_BYTE_ARRAY: + context->physical_type = ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY; + break; + default: + return Status::NotSupported("Unsupported Parquet physical type {}", + tparquet::to_string(field.physical_type)); + } + + const auto& schema = field.parquet_schema; + context->type_length = schema.__isset.type_length ? schema.type_length : -1; + context->decimal_precision = schema.__isset.precision ? schema.precision : -1; + context->decimal_scale = schema.__isset.scale ? schema.scale : -1; + context->timezone = ctz; + if (schema.__isset.logicalType) { + const auto& logical = schema.logicalType; + if (logical.__isset.STRING || logical.__isset.ENUM || logical.__isset.JSON || + logical.__isset.BSON) { + context->logical_type = ParquetLogicalType::STRING; + } else if (logical.__isset.DECIMAL) { + context->logical_type = ParquetLogicalType::DECIMAL; + context->decimal_precision = logical.DECIMAL.precision; + context->decimal_scale = logical.DECIMAL.scale; + } else if (logical.__isset.DATE) { + context->logical_type = ParquetLogicalType::DATE; + } else if (logical.__isset.TIME) { + context->logical_type = ParquetLogicalType::TIME; + context->time_unit = parquet_time_unit(logical.TIME.unit); + } else if (logical.__isset.TIMESTAMP) { + context->logical_type = ParquetLogicalType::TIMESTAMP; + context->time_unit = parquet_time_unit(logical.TIMESTAMP.unit); + context->timestamp_is_adjusted_to_utc = logical.TIMESTAMP.isAdjustedToUTC; + } else if (logical.__isset.INTEGER) { + context->logical_type = ParquetLogicalType::INTEGER; + context->logical_integer_bit_width = logical.INTEGER.bitWidth; + context->logical_integer_is_signed = logical.INTEGER.isSigned; + } else if (logical.__isset.UUID) { + context->logical_type = ParquetLogicalType::UUID; + context->logical_uuid = true; + } else if (logical.__isset.FLOAT16) { + context->logical_type = ParquetLogicalType::FLOAT16; + context->logical_float16 = true; + } + if (context->logical_uuid && + (context->physical_type != ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY || + context->type_length != 16)) { + return Status::Corruption("Parquet UUID field {} must be FIXED_LEN_BYTE_ARRAY(16)", + field.name); + } + if (context->logical_float16 && + (context->physical_type != ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY || + context->type_length != 2)) { + return Status::Corruption("Parquet FLOAT16 field {} must be FIXED_LEN_BYTE_ARRAY(2)", + field.name); + } + return Status::OK(); + } + + if (!schema.__isset.converted_type) { + return Status::OK(); + } + switch (schema.converted_type) { + case tparquet::ConvertedType::UTF8: + case tparquet::ConvertedType::ENUM: + case tparquet::ConvertedType::JSON: + case tparquet::ConvertedType::BSON: + context->logical_type = ParquetLogicalType::STRING; + break; + case tparquet::ConvertedType::DECIMAL: + context->logical_type = ParquetLogicalType::DECIMAL; + break; + case tparquet::ConvertedType::DATE: + context->logical_type = ParquetLogicalType::DATE; + break; + case tparquet::ConvertedType::TIME_MILLIS: + context->logical_type = ParquetLogicalType::TIME; + context->time_unit = ParquetTimeUnit::MILLIS; + break; + case tparquet::ConvertedType::TIME_MICROS: + context->logical_type = ParquetLogicalType::TIME; + context->time_unit = ParquetTimeUnit::MICROS; + break; + case tparquet::ConvertedType::TIMESTAMP_MILLIS: + context->logical_type = ParquetLogicalType::TIMESTAMP; + context->time_unit = ParquetTimeUnit::MILLIS; + // Legacy converted timestamps are defined as UTC-adjusted, unlike an unannotated INT64. + context->timestamp_is_adjusted_to_utc = true; + break; + case tparquet::ConvertedType::TIMESTAMP_MICROS: + context->logical_type = ParquetLogicalType::TIMESTAMP; + context->time_unit = ParquetTimeUnit::MICROS; + context->timestamp_is_adjusted_to_utc = true; + break; + case tparquet::ConvertedType::UINT_8: + case tparquet::ConvertedType::UINT_16: + case tparquet::ConvertedType::UINT_32: + case tparquet::ConvertedType::UINT_64: + case tparquet::ConvertedType::INT_8: + case tparquet::ConvertedType::INT_16: + case tparquet::ConvertedType::INT_32: + case tparquet::ConvertedType::INT_64: + context->logical_type = ParquetLogicalType::INTEGER; + context->logical_integer_is_signed = + schema.converted_type >= tparquet::ConvertedType::INT_8; + context->logical_integer_bit_width = + schema.converted_type == tparquet::ConvertedType::UINT_8 || + schema.converted_type == tparquet::ConvertedType::INT_8 + ? 8 + : schema.converted_type == tparquet::ConvertedType::UINT_16 || + schema.converted_type == tparquet::ConvertedType::INT_16 + ? 16 + : schema.converted_type == tparquet::ConvertedType::UINT_32 || + schema.converted_type == tparquet::ConvertedType::INT_32 + ? 32 + : 64; + break; + default: + break; + } + return Status::OK(); +} + +} // namespace + +#ifdef BE_TEST +Status init_decode_context_for_test(const FieldSchema& field, const cctz::time_zone* ctz, + ParquetDecodeContext* context) { + return init_decode_context(field, ctz, context); +} +#endif + +static void fill_struct_null_map(FieldSchema* field, NullMap& null_map, + const std::vector<level_t>& rep_levels, + const std::vector<level_t>& def_levels) { + size_t num_levels = def_levels.size(); + DCHECK_EQ(num_levels, rep_levels.size()); + size_t origin_size = null_map.size(); + null_map.resize(origin_size + num_levels); + size_t pos = origin_size; + for (size_t i = 0; i < num_levels; ++i) { + // skip the levels affect its ancestor or its descendants + if (def_levels[i] < field->repeated_parent_def_level || + rep_levels[i] > field->repetition_level) { + continue; + } + if (def_levels[i] >= field->definition_level) { + null_map[pos++] = 0; + } else { + null_map[pos++] = 1; + } + } + null_map.resize(pos); +} + +static Status fill_array_offset(FieldSchema* field, ColumnArray::Offsets64& offsets_data, + NullMap* null_map_ptr, const std::vector<level_t>& rep_levels, + const std::vector<level_t>& def_levels) { + size_t num_levels = rep_levels.size(); + if (UNLIKELY(num_levels != def_levels.size())) { + return Status::Corruption("Parquet repetition and definition level counts differ"); + } + size_t origin_size = offsets_data.size(); + offsets_data.resize(origin_size + num_levels); + if (null_map_ptr != nullptr) { + null_map_ptr->resize(origin_size + num_levels); + } + size_t offset_pos = origin_size - 1; + bool parent_opened = false; + for (size_t i = 0; i < num_levels; ++i) { + // skip the levels affect its ancestor or its descendants + if (def_levels[i] < field->repeated_parent_def_level || + rep_levels[i] > field->repetition_level) { + continue; + } + if (rep_levels[i] == field->repetition_level) { + // A continuation can extend only a parent opened by this aligned logical batch. + if (UNLIKELY(!parent_opened)) { + return Status::Corruption( + "Parquet collection starts with an orphan repetition continuation"); + } + offsets_data[offset_pos]++; + continue; + } + parent_opened = true; + offset_pos++; + offsets_data[offset_pos] = offsets_data[offset_pos - 1]; + if (def_levels[i] >= field->definition_level) { + offsets_data[offset_pos]++; + } + if (null_map_ptr != nullptr) { + if (def_levels[i] >= field->definition_level - 1) { + (*null_map_ptr)[offset_pos] = 0; + } else { + (*null_map_ptr)[offset_pos] = 1; + } + } + } + offsets_data.resize(offset_pos + 1); + if (null_map_ptr != nullptr) { + null_map_ptr->resize(offset_pos + 1); + } + return Status::OK(); +} + +Status ColumnReader::create(io::FileReaderSPtr file, FieldSchema* field, + const tparquet::RowGroup& row_group, const RowRanges& row_ranges, + const cctz::time_zone* ctz, io::IOContext* io_ctx, + std::unique_ptr<ColumnReader>& reader, size_t max_buf_size, + const std::unordered_map<int, tparquet::OffsetIndex>& col_offsets, + RuntimeState* state, bool in_collection, + const std::set<uint64_t>& column_ids, + const std::set<uint64_t>& filter_column_ids, + const std::string& page_cache_file_key) { + size_t total_rows = row_group.num_rows; + if (field->data_type->get_primitive_type() == TYPE_ARRAY) { + std::unique_ptr<ColumnReader> element_reader; + RETURN_IF_ERROR(create(file, &field->children[0], row_group, row_ranges, ctz, io_ctx, + element_reader, max_buf_size, col_offsets, state, true, column_ids, + filter_column_ids, page_cache_file_key)); + auto array_reader = ArrayColumnReader::create_unique(row_ranges, total_rows, ctz, io_ctx); + element_reader->set_column_in_nested(); + RETURN_IF_ERROR(array_reader->init(std::move(element_reader), field)); + array_reader->_filter_column_ids = filter_column_ids; + reader.reset(array_reader.release()); + } else if (field->data_type->get_primitive_type() == TYPE_MAP) { + std::unique_ptr<ColumnReader> key_reader; + std::unique_ptr<ColumnReader> value_reader; + + if (column_ids.empty() || + column_ids.find(field->children[0].get_column_id()) != column_ids.end()) { + // Create key reader + RETURN_IF_ERROR(create(file, &field->children[0], row_group, row_ranges, ctz, io_ctx, + key_reader, max_buf_size, col_offsets, state, true, column_ids, + filter_column_ids, page_cache_file_key)); + } else { + auto skip_reader = std::make_unique<SkipReadingReader>(row_ranges, total_rows, ctz, + io_ctx, &field->children[0]); + key_reader = std::move(skip_reader); + } + + if (column_ids.empty() || + column_ids.find(field->children[1].get_column_id()) != column_ids.end()) { + // Create value reader + RETURN_IF_ERROR(create(file, &field->children[1], row_group, row_ranges, ctz, io_ctx, + value_reader, max_buf_size, col_offsets, state, true, column_ids, + filter_column_ids, page_cache_file_key)); + } else { + auto skip_reader = std::make_unique<SkipReadingReader>(row_ranges, total_rows, ctz, + io_ctx, &field->children[1]); + value_reader = std::move(skip_reader); + } + + auto map_reader = MapColumnReader::create_unique(row_ranges, total_rows, ctz, io_ctx); + key_reader->set_column_in_nested(); + value_reader->set_column_in_nested(); + RETURN_IF_ERROR(map_reader->init(std::move(key_reader), std::move(value_reader), field)); + map_reader->_filter_column_ids = filter_column_ids; + reader.reset(map_reader.release()); + } else if (field->data_type->get_primitive_type() == TYPE_STRUCT) { + std::unordered_map<std::string, std::unique_ptr<ColumnReader>> child_readers; + child_readers.reserve(field->children.size()); + int non_skip_reader_idx = -1; + for (int i = 0; i < field->children.size(); ++i) { + auto& child = field->children[i]; + std::unique_ptr<ColumnReader> child_reader; + if (column_ids.empty() || column_ids.find(child.get_column_id()) != column_ids.end()) { + RETURN_IF_ERROR(create(file, &child, row_group, row_ranges, ctz, io_ctx, + child_reader, max_buf_size, col_offsets, state, + in_collection, column_ids, filter_column_ids, + page_cache_file_key)); + child_readers[child.name] = std::move(child_reader); + // Record the first non-SkippingReader + if (non_skip_reader_idx == -1) { + non_skip_reader_idx = i; + } + } else { + auto skip_reader = std::make_unique<SkipReadingReader>(row_ranges, total_rows, ctz, + io_ctx, &child); + skip_reader->_filter_column_ids = filter_column_ids; + child_readers[child.name] = std::move(skip_reader); + } + child_readers[child.name]->set_column_in_nested(); + } + // If all children are SkipReadingReader, force the first child to call create + if (non_skip_reader_idx == -1) { + std::unique_ptr<ColumnReader> child_reader; + RETURN_IF_ERROR(create(file, &field->children[0], row_group, row_ranges, ctz, io_ctx, + child_reader, max_buf_size, col_offsets, state, in_collection, + column_ids, filter_column_ids, page_cache_file_key)); + child_reader->set_column_in_nested(); + child_readers[field->children[0].name] = std::move(child_reader); + } + auto struct_reader = StructColumnReader::create_unique(row_ranges, total_rows, ctz, io_ctx); + RETURN_IF_ERROR(struct_reader->init(std::move(child_readers), field)); + struct_reader->_filter_column_ids = filter_column_ids; + reader.reset(struct_reader.release()); + } else { + auto physical_index = field->physical_column_index; + const auto offset_it = col_offsets.find(physical_index); + const tparquet::OffsetIndex* offset_index = + offset_it != col_offsets.end() ? &offset_it->second : nullptr; + + const tparquet::ColumnChunk& chunk = row_group.columns[physical_index]; + if (in_collection) { + if (offset_index == nullptr) { + auto scalar_reader = ScalarColumnReader<true, false>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } else { + auto scalar_reader = ScalarColumnReader<true, true>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } + } else { + if (offset_index == nullptr) { + auto scalar_reader = ScalarColumnReader<false, false>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } else { + auto scalar_reader = ScalarColumnReader<false, true>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } + } + } + return Status::OK(); +} + +void ColumnReader::_generate_read_ranges(RowRange page_row_range, RowRanges* result_ranges) const { + result_ranges->add(page_row_range); + RowRanges::ranges_intersection(*result_ranges, _row_ranges, result_ranges); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::init( + io::FileReaderSPtr file, FieldSchema* field, size_t max_buf_size, RuntimeState* state, + const std::string& page_cache_file_key) { + _field_schema = field; + auto& chunk_meta = _chunk_meta.meta_data; + int64_t chunk_start = has_dict_page(chunk_meta) ? chunk_meta.dictionary_page_offset Review Comment: [P1] Validate the signed chunk range before these unsigned conversions. The Arrow path being replaced rejected negative starts/lengths, checked `start + length`, and bounded the end by file size. Here a positive start plus a negative `total_compressed_size` whose magnitude exceeds the start becomes an unsigned length with a near-`UINT64_MAX` end, so both stream and page bounds admit following column chunks and can decode their pages as this column. Use one signed, checked, file-bounded range helper for scalar and COUNT readers before constructing either reader; keep the separate old-writer padding after validation, and add negative/overflow/out-of-file metadata tests. ########## be/src/format_v2/parquet/reader/native/column_reader.cpp: ########## @@ -0,0 +1,1529 @@ +// 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 <gen_cpp/parquet_types.h> +#include <limits.h> +#include <sys/types.h> + +#include <algorithm> +#include <utility> + +#include "common/cast_set.h" +#include "common/status.h" +#include "core/column/column.h" +#include "core/column/column_array.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.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_struct.h" +#include "core/data_type/define_primitive_type.h" +#include "format/parquet/schema_desc.h" +#include "format_v2/parquet/reader/native/column_chunk_reader.h" +#include "format_v2/parquet/reader/native/level_decoder.h" +#include "io/fs/tracing_file_reader.h" +#include "runtime/runtime_profile.h" + +namespace doris::format::parquet::native { +namespace { + +ParquetTimeUnit parquet_time_unit(const tparquet::TimeUnit& unit) { + if (unit.__isset.MILLIS) { + return ParquetTimeUnit::MILLIS; + } + if (unit.__isset.MICROS) { + return ParquetTimeUnit::MICROS; + } + if (unit.__isset.NANOS) { + return ParquetTimeUnit::NANOS; + } + return ParquetTimeUnit::UNKNOWN; +} + +bool is_direct_integer_type(PrimitiveType type) { + switch (type) { + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + return true; + default: + return false; + } +} + +bool is_direct_decimal_type(PrimitiveType type) { + switch (type) { + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + return true; + default: + return false; + } +} + +template <typename T> +bool release_vector_if_oversized(std::vector<T>* values, size_t max_retained_bytes) { + DORIS_CHECK(values != nullptr); + if (values->capacity() * sizeof(T) <= max_retained_bytes) { + return false; + } + std::vector<T>().swap(*values); + return true; +} + +size_t retained_set_bytes(const std::unordered_set<size_t>& values) { + return values.bucket_count() * sizeof(void*) + values.size() * sizeof(size_t); +} + +bool is_direct_binary_type(PrimitiveType type) { + return is_string_type(type) || type == TYPE_VARBINARY; +} + +// The target SerDe can fuse physical decode with these logical type changes. Less common schema +// changes retain the generic file-format converter as a compatibility path: the decoder still +// exposes raw spans, but the source SerDe first materializes a reusable source column before the +// generic logical cast. Ordinary scans and numeric widening never allocate that source column. +bool serde_can_materialize_directly(const DataTypePtr& source_type, + const DataTypePtr& target_type) { + const auto source = remove_nullable(source_type)->get_primitive_type(); + const auto target = remove_nullable(target_type)->get_primitive_type(); + return source == target || (is_direct_integer_type(source) && is_direct_integer_type(target)) || + (source == TYPE_FLOAT && target == TYPE_DOUBLE) || + (is_direct_decimal_type(source) && is_direct_decimal_type(target)) || + // Parquet STRING and VARBINARY share BYTE_ARRAY bytes. Materializing through the target + // SerDe preserves those bytes and avoids a converter whose scratch column uses the v1 + // String representation instead of the native ColumnVarbinary representation. + (is_direct_binary_type(source) && is_direct_binary_type(target)); +} + +Status init_decode_context(const FieldSchema& field, const cctz::time_zone* ctz, + ParquetDecodeContext* context) { + DORIS_CHECK(context != nullptr); + switch (field.physical_type) { + case tparquet::Type::BOOLEAN: + context->physical_type = ParquetPhysicalType::BOOLEAN; + break; + case tparquet::Type::INT32: + context->physical_type = ParquetPhysicalType::INT32; + break; + case tparquet::Type::INT64: + context->physical_type = ParquetPhysicalType::INT64; + break; + case tparquet::Type::INT96: + context->physical_type = ParquetPhysicalType::INT96; + break; + case tparquet::Type::FLOAT: + context->physical_type = ParquetPhysicalType::FLOAT; + break; + case tparquet::Type::DOUBLE: + context->physical_type = ParquetPhysicalType::DOUBLE; + break; + case tparquet::Type::BYTE_ARRAY: + context->physical_type = ParquetPhysicalType::BYTE_ARRAY; + break; + case tparquet::Type::FIXED_LEN_BYTE_ARRAY: + context->physical_type = ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY; + break; + default: + return Status::NotSupported("Unsupported Parquet physical type {}", + tparquet::to_string(field.physical_type)); + } + + const auto& schema = field.parquet_schema; + context->type_length = schema.__isset.type_length ? schema.type_length : -1; + context->decimal_precision = schema.__isset.precision ? schema.precision : -1; + context->decimal_scale = schema.__isset.scale ? schema.scale : -1; + context->timezone = ctz; + if (schema.__isset.logicalType) { + const auto& logical = schema.logicalType; + if (logical.__isset.STRING || logical.__isset.ENUM || logical.__isset.JSON || + logical.__isset.BSON) { + context->logical_type = ParquetLogicalType::STRING; + } else if (logical.__isset.DECIMAL) { + context->logical_type = ParquetLogicalType::DECIMAL; + context->decimal_precision = logical.DECIMAL.precision; + context->decimal_scale = logical.DECIMAL.scale; + } else if (logical.__isset.DATE) { + context->logical_type = ParquetLogicalType::DATE; + } else if (logical.__isset.TIME) { + context->logical_type = ParquetLogicalType::TIME; + context->time_unit = parquet_time_unit(logical.TIME.unit); + } else if (logical.__isset.TIMESTAMP) { + context->logical_type = ParquetLogicalType::TIMESTAMP; + context->time_unit = parquet_time_unit(logical.TIMESTAMP.unit); + context->timestamp_is_adjusted_to_utc = logical.TIMESTAMP.isAdjustedToUTC; + } else if (logical.__isset.INTEGER) { + context->logical_type = ParquetLogicalType::INTEGER; + context->logical_integer_bit_width = logical.INTEGER.bitWidth; + context->logical_integer_is_signed = logical.INTEGER.isSigned; + } else if (logical.__isset.UUID) { + context->logical_type = ParquetLogicalType::UUID; + context->logical_uuid = true; + } else if (logical.__isset.FLOAT16) { + context->logical_type = ParquetLogicalType::FLOAT16; + context->logical_float16 = true; + } + if (context->logical_uuid && + (context->physical_type != ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY || + context->type_length != 16)) { + return Status::Corruption("Parquet UUID field {} must be FIXED_LEN_BYTE_ARRAY(16)", + field.name); + } + if (context->logical_float16 && + (context->physical_type != ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY || + context->type_length != 2)) { + return Status::Corruption("Parquet FLOAT16 field {} must be FIXED_LEN_BYTE_ARRAY(2)", + field.name); + } + return Status::OK(); + } + + if (!schema.__isset.converted_type) { + return Status::OK(); + } + switch (schema.converted_type) { + case tparquet::ConvertedType::UTF8: + case tparquet::ConvertedType::ENUM: + case tparquet::ConvertedType::JSON: + case tparquet::ConvertedType::BSON: + context->logical_type = ParquetLogicalType::STRING; + break; + case tparquet::ConvertedType::DECIMAL: + context->logical_type = ParquetLogicalType::DECIMAL; + break; + case tparquet::ConvertedType::DATE: + context->logical_type = ParquetLogicalType::DATE; + break; + case tparquet::ConvertedType::TIME_MILLIS: + context->logical_type = ParquetLogicalType::TIME; + context->time_unit = ParquetTimeUnit::MILLIS; + break; + case tparquet::ConvertedType::TIME_MICROS: + context->logical_type = ParquetLogicalType::TIME; + context->time_unit = ParquetTimeUnit::MICROS; + break; + case tparquet::ConvertedType::TIMESTAMP_MILLIS: + context->logical_type = ParquetLogicalType::TIMESTAMP; + context->time_unit = ParquetTimeUnit::MILLIS; + // Legacy converted timestamps are defined as UTC-adjusted, unlike an unannotated INT64. + context->timestamp_is_adjusted_to_utc = true; + break; + case tparquet::ConvertedType::TIMESTAMP_MICROS: + context->logical_type = ParquetLogicalType::TIMESTAMP; + context->time_unit = ParquetTimeUnit::MICROS; + context->timestamp_is_adjusted_to_utc = true; + break; + case tparquet::ConvertedType::UINT_8: + case tparquet::ConvertedType::UINT_16: + case tparquet::ConvertedType::UINT_32: + case tparquet::ConvertedType::UINT_64: + case tparquet::ConvertedType::INT_8: + case tparquet::ConvertedType::INT_16: + case tparquet::ConvertedType::INT_32: + case tparquet::ConvertedType::INT_64: + context->logical_type = ParquetLogicalType::INTEGER; + context->logical_integer_is_signed = + schema.converted_type >= tparquet::ConvertedType::INT_8; + context->logical_integer_bit_width = + schema.converted_type == tparquet::ConvertedType::UINT_8 || + schema.converted_type == tparquet::ConvertedType::INT_8 + ? 8 + : schema.converted_type == tparquet::ConvertedType::UINT_16 || + schema.converted_type == tparquet::ConvertedType::INT_16 + ? 16 + : schema.converted_type == tparquet::ConvertedType::UINT_32 || + schema.converted_type == tparquet::ConvertedType::INT_32 + ? 32 + : 64; + break; + default: + break; + } + return Status::OK(); +} + +} // namespace + +#ifdef BE_TEST +Status init_decode_context_for_test(const FieldSchema& field, const cctz::time_zone* ctz, + ParquetDecodeContext* context) { + return init_decode_context(field, ctz, context); +} +#endif + +static void fill_struct_null_map(FieldSchema* field, NullMap& null_map, + const std::vector<level_t>& rep_levels, + const std::vector<level_t>& def_levels) { + size_t num_levels = def_levels.size(); + DCHECK_EQ(num_levels, rep_levels.size()); + size_t origin_size = null_map.size(); + null_map.resize(origin_size + num_levels); + size_t pos = origin_size; + for (size_t i = 0; i < num_levels; ++i) { + // skip the levels affect its ancestor or its descendants + if (def_levels[i] < field->repeated_parent_def_level || + rep_levels[i] > field->repetition_level) { + continue; + } + if (def_levels[i] >= field->definition_level) { + null_map[pos++] = 0; + } else { + null_map[pos++] = 1; + } + } + null_map.resize(pos); +} + +static Status fill_array_offset(FieldSchema* field, ColumnArray::Offsets64& offsets_data, + NullMap* null_map_ptr, const std::vector<level_t>& rep_levels, + const std::vector<level_t>& def_levels) { + size_t num_levels = rep_levels.size(); + if (UNLIKELY(num_levels != def_levels.size())) { + return Status::Corruption("Parquet repetition and definition level counts differ"); + } + size_t origin_size = offsets_data.size(); + offsets_data.resize(origin_size + num_levels); + if (null_map_ptr != nullptr) { + null_map_ptr->resize(origin_size + num_levels); + } + size_t offset_pos = origin_size - 1; + bool parent_opened = false; + for (size_t i = 0; i < num_levels; ++i) { + // skip the levels affect its ancestor or its descendants + if (def_levels[i] < field->repeated_parent_def_level || + rep_levels[i] > field->repetition_level) { + continue; + } + if (rep_levels[i] == field->repetition_level) { + // A continuation can extend only a parent opened by this aligned logical batch. + if (UNLIKELY(!parent_opened)) { + return Status::Corruption( + "Parquet collection starts with an orphan repetition continuation"); + } + offsets_data[offset_pos]++; + continue; + } + parent_opened = true; + offset_pos++; + offsets_data[offset_pos] = offsets_data[offset_pos - 1]; + if (def_levels[i] >= field->definition_level) { + offsets_data[offset_pos]++; + } + if (null_map_ptr != nullptr) { + if (def_levels[i] >= field->definition_level - 1) { + (*null_map_ptr)[offset_pos] = 0; + } else { + (*null_map_ptr)[offset_pos] = 1; + } + } + } + offsets_data.resize(offset_pos + 1); + if (null_map_ptr != nullptr) { + null_map_ptr->resize(offset_pos + 1); + } + return Status::OK(); +} + +Status ColumnReader::create(io::FileReaderSPtr file, FieldSchema* field, + const tparquet::RowGroup& row_group, const RowRanges& row_ranges, + const cctz::time_zone* ctz, io::IOContext* io_ctx, + std::unique_ptr<ColumnReader>& reader, size_t max_buf_size, + const std::unordered_map<int, tparquet::OffsetIndex>& col_offsets, + RuntimeState* state, bool in_collection, + const std::set<uint64_t>& column_ids, + const std::set<uint64_t>& filter_column_ids, + const std::string& page_cache_file_key) { + size_t total_rows = row_group.num_rows; + if (field->data_type->get_primitive_type() == TYPE_ARRAY) { + std::unique_ptr<ColumnReader> element_reader; + RETURN_IF_ERROR(create(file, &field->children[0], row_group, row_ranges, ctz, io_ctx, + element_reader, max_buf_size, col_offsets, state, true, column_ids, + filter_column_ids, page_cache_file_key)); + auto array_reader = ArrayColumnReader::create_unique(row_ranges, total_rows, ctz, io_ctx); + element_reader->set_column_in_nested(); + RETURN_IF_ERROR(array_reader->init(std::move(element_reader), field)); + array_reader->_filter_column_ids = filter_column_ids; + reader.reset(array_reader.release()); + } else if (field->data_type->get_primitive_type() == TYPE_MAP) { + std::unique_ptr<ColumnReader> key_reader; + std::unique_ptr<ColumnReader> value_reader; + + if (column_ids.empty() || + column_ids.find(field->children[0].get_column_id()) != column_ids.end()) { + // Create key reader + RETURN_IF_ERROR(create(file, &field->children[0], row_group, row_ranges, ctz, io_ctx, + key_reader, max_buf_size, col_offsets, state, true, column_ids, + filter_column_ids, page_cache_file_key)); + } else { + auto skip_reader = std::make_unique<SkipReadingReader>(row_ranges, total_rows, ctz, + io_ctx, &field->children[0]); + key_reader = std::move(skip_reader); + } + + if (column_ids.empty() || + column_ids.find(field->children[1].get_column_id()) != column_ids.end()) { + // Create value reader + RETURN_IF_ERROR(create(file, &field->children[1], row_group, row_ranges, ctz, io_ctx, + value_reader, max_buf_size, col_offsets, state, true, column_ids, + filter_column_ids, page_cache_file_key)); + } else { + auto skip_reader = std::make_unique<SkipReadingReader>(row_ranges, total_rows, ctz, + io_ctx, &field->children[1]); + value_reader = std::move(skip_reader); + } + + auto map_reader = MapColumnReader::create_unique(row_ranges, total_rows, ctz, io_ctx); + key_reader->set_column_in_nested(); + value_reader->set_column_in_nested(); + RETURN_IF_ERROR(map_reader->init(std::move(key_reader), std::move(value_reader), field)); + map_reader->_filter_column_ids = filter_column_ids; + reader.reset(map_reader.release()); + } else if (field->data_type->get_primitive_type() == TYPE_STRUCT) { + std::unordered_map<std::string, std::unique_ptr<ColumnReader>> child_readers; + child_readers.reserve(field->children.size()); + int non_skip_reader_idx = -1; + for (int i = 0; i < field->children.size(); ++i) { + auto& child = field->children[i]; + std::unique_ptr<ColumnReader> child_reader; + if (column_ids.empty() || column_ids.find(child.get_column_id()) != column_ids.end()) { + RETURN_IF_ERROR(create(file, &child, row_group, row_ranges, ctz, io_ctx, + child_reader, max_buf_size, col_offsets, state, + in_collection, column_ids, filter_column_ids, + page_cache_file_key)); + child_readers[child.name] = std::move(child_reader); + // Record the first non-SkippingReader + if (non_skip_reader_idx == -1) { + non_skip_reader_idx = i; + } + } else { + auto skip_reader = std::make_unique<SkipReadingReader>(row_ranges, total_rows, ctz, + io_ctx, &child); + skip_reader->_filter_column_ids = filter_column_ids; + child_readers[child.name] = std::move(skip_reader); + } + child_readers[child.name]->set_column_in_nested(); + } + // If all children are SkipReadingReader, force the first child to call create + if (non_skip_reader_idx == -1) { + std::unique_ptr<ColumnReader> child_reader; + RETURN_IF_ERROR(create(file, &field->children[0], row_group, row_ranges, ctz, io_ctx, + child_reader, max_buf_size, col_offsets, state, in_collection, + column_ids, filter_column_ids, page_cache_file_key)); + child_reader->set_column_in_nested(); + child_readers[field->children[0].name] = std::move(child_reader); + } + auto struct_reader = StructColumnReader::create_unique(row_ranges, total_rows, ctz, io_ctx); + RETURN_IF_ERROR(struct_reader->init(std::move(child_readers), field)); + struct_reader->_filter_column_ids = filter_column_ids; + reader.reset(struct_reader.release()); + } else { + auto physical_index = field->physical_column_index; + const auto offset_it = col_offsets.find(physical_index); + const tparquet::OffsetIndex* offset_index = + offset_it != col_offsets.end() ? &offset_it->second : nullptr; + + const tparquet::ColumnChunk& chunk = row_group.columns[physical_index]; + if (in_collection) { + if (offset_index == nullptr) { + auto scalar_reader = ScalarColumnReader<true, false>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } else { + auto scalar_reader = ScalarColumnReader<true, true>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } + } else { + if (offset_index == nullptr) { + auto scalar_reader = ScalarColumnReader<false, false>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } else { + auto scalar_reader = ScalarColumnReader<false, true>::create_unique( + row_ranges, total_rows, chunk, offset_index, ctz, io_ctx); + + RETURN_IF_ERROR( + scalar_reader->init(file, field, max_buf_size, state, page_cache_file_key)); + scalar_reader->_filter_column_ids = filter_column_ids; + reader.reset(scalar_reader.release()); + } + } + } + return Status::OK(); +} + +void ColumnReader::_generate_read_ranges(RowRange page_row_range, RowRanges* result_ranges) const { + result_ranges->add(page_row_range); + RowRanges::ranges_intersection(*result_ranges, _row_ranges, result_ranges); +} + +template <bool IN_COLLECTION, bool OFFSET_INDEX> +Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::init( + io::FileReaderSPtr file, FieldSchema* field, size_t max_buf_size, RuntimeState* state, + const std::string& page_cache_file_key) { + _field_schema = field; + auto& chunk_meta = _chunk_meta.meta_data; + int64_t chunk_start = has_dict_page(chunk_meta) ? chunk_meta.dictionary_page_offset + : chunk_meta.data_page_offset; + size_t chunk_len = chunk_meta.total_compressed_size; Review Comment: [P2] Preserve the old-writer column-chunk padding used by the reader this replaces. Arrow's `ComputeColumnChunkRange()` detects parquet-mr versions affected by PARQUET-816, whose `total_compressed_size` omits the dictionary-page header, and adds up to 100 bytes before opening the page reader. This native path uses the underreported length verbatim, while `PageReader::_validate_page_header()` now requires every header and payload to fit it, so dictionary-encoded files from those writers fail on the final page. Carry the writer-version workaround into the native range (bounded by file size) and add a legacy parquet-mr fixture. -- 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]
