github-actions[bot] commented on code in PR #65674: URL: https://github.com/apache/doris/pull/65674#discussion_r3610392222
########## be/src/format_v2/parquet/reader/native/column_chunk_reader.h: ########## @@ -0,0 +1,352 @@ +// 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 <cstddef> +#include <cstdint> +#include <memory> +#include <string> +#include <unordered_map> +#include <vector> + +#include "common/status.h" +#include "core/column/column_string.h" +#include "core/data_type/data_type.h" +#include "core/data_type_serde/parquet_decode_source.h" +#include "format_v2/parquet/native_schema_desc.h" +#include "format_v2/parquet/reader/native/common.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 "util/slice.h" + +namespace doris { +class BlockCompressionCodec; +class DataTypeSerDe; +namespace io { +class BufferedStreamReader; +struct IOContext; +} // namespace io + +} // namespace doris + +namespace doris::format::parquet::native { +using ::doris::ColumnString; + +struct ColumnChunkRange { + size_t offset = 0; + size_t length = 0; +}; + +struct ParquetReaderCompat { + bool parquet_816_padding = false; + bool data_page_v2_always_compressed = false; +}; + +ParquetReaderCompat parquet_reader_compat(const std::string& created_by); +Status compute_column_chunk_range(const tparquet::ColumnMetaData& metadata, size_t file_size, + bool parquet_816_padding, ColumnChunkRange* range); +bool validate_offset_index(const tparquet::OffsetIndex& index, const ColumnChunkRange& chunk_range, + int64_t data_page_offset, int64_t row_count); +bool can_prepare_page_cache_payload(bool session_cache_enabled, bool storage_cache_disabled, + bool cache_available, bool header_available); +Status validate_uncompressed_page_sizes(const tparquet::PageHeader& header, + tparquet::CompressionCodec::type codec, + bool data_page_v2_always_compressed); + +struct ColumnChunkReaderStatistics { + int64_t decompress_time = 0; + int64_t decompress_cnt = 0; + int64_t decode_header_time = 0; + int64_t decode_value_time = 0; + int64_t materialization_time = 0; + int64_t hybrid_selection_batches = 0; + int64_t hybrid_selection_ranges = 0; + int64_t hybrid_selection_null_fallback_batches = 0; + int64_t decode_dict_time = 0; + int64_t decode_level_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_read_counter = 0; + int64_t data_page_read_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_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; +}; + +/** + * Read and decode parquet column data into doris block column. + * <p>Usage:</p> + * // Create chunk reader + * ColumnChunkReader chunk_reader(BufferedStreamReader* reader, + * tparquet::ColumnChunk* column_chunk, + * NativeFieldSchema* fieldSchema); + * // Initialize chunk reader + * chunk_reader.init(); + * while (chunk_reader.has_next_page()) { + * // Seek to next page header. Only read and parse the page header, not page data. + * chunk_reader.next_page(); + * // Load data to decoder. Load the page data into underlying container. + * // Or, we can call the chunk_reader.skip_page() to skip current page. + * chunk_reader.load_page_data(); + * // Decode values into column or slice. + * // Or, we can call chunk_reader.skip_values(num_values) to skip some values. + * chunk_reader.materialize_values(column, serde, context, state, selection); + * } + */ +template <bool IN_COLLECTION, bool OFFSET_INDEX> +class ColumnChunkReader { +public: + ColumnChunkReader(io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk, + NativeFieldSchema* field_schema, const tparquet::OffsetIndex* offset_index, + size_t total_row, io::IOContext* io_ctx, + const ParquetPageReadContext& page_read_ctx, + const ColumnChunkRange* chunk_range = nullptr); + ~ColumnChunkReader() = default; + + // Initialize chunk reader, will generate the decoder and codec. + Status init(); + + // Whether the chunk reader has a more page to read. + bool has_next_page() const { + if constexpr (OFFSET_INDEX) { + return _page_reader->has_next_page(); + } else { + // no offset need parse all page header. + return _chunk_parsed_values < _metadata.num_values; + } + } + + // Skip some values(will not read and parse) in current page if the values are filtered by predicates. + // when skip_data = false, the underlying decoder will not skip data, + // only used when maintaining the consistency of _remaining_num_values. + Status skip_values(size_t num_values, bool skip_data = true); + + // Load page data into the underlying container, + // and initialize the repetition and definition level decoder for current page data. + Status load_page_data(); + Status load_page_data_idempotent() { + if (_state == DATA_LOADED) { + return Status::OK(); + } + return load_page_data(); + } + // The remaining number of values in current page(including null values). Decreased when reading or skipping. + uint32_t remaining_num_values() const { return _remaining_num_values; } + + // Apply one logical selection to the encoded page. Decoder advances physical payload cursors; + // SerDe interprets each selected raw span and materializes the destination Doris column. + Status materialize_values(MutableColumnPtr& doris_column, const DataTypeSerDe& serde, + ParquetDecodeContext& context, ParquetMaterializationState& state, + ColumnSelectVector& select_vector); + + // Get the repetition level decoder of current page. + LevelDecoder& rep_level_decoder() { return _rep_level_decoder; } + // Get the definition level decoder of current page. + LevelDecoder& def_level_decoder() { return _def_level_decoder; } + + level_t max_rep_level() const { return _max_rep_level; } + level_t max_def_level() const { return _max_def_level; } + + bool has_dict() const { return _has_dict; }; + + // Get page decoder + Decoder* get_page_decoder() { return _page_decoder; } + + void release_decoder_scratch(size_t max_retained_bytes) { + for (auto& [encoding, decoder] : _decoders) { + decoder->release_scratch(max_retained_bytes); Review Comment: [P1] Do not reclaim semantic state from the active page decoder This loop includes the decoder referenced by `_page_decoder`, but some `release_scratch()` implementations clear capacity-based buffers that are still required for unread values. In DELTA_BYTE_ARRAY, an early >4 MiB value followed by SSO-sized values leaves `_last_value` oversized; after three 16-batch checks it is swapped empty with `_num_valid_values > 0`, so the next nonzero prefix fails. Decoder reuse makes DELTA_BINARY_PACKED unsafe too: an outlier page can leave >4 MiB capacity in `_delta_bit_widths`, a later page resizes it to a small live table, and reclamation mid-block makes the next miniblock transition index an empty vector. Restrict this path to inactive decoders or batch-only buffers, or defer semantic-state release until page/block exhaustion, and cover both cases with low-threshold multi-batch tests. -- 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]
