Gabriel39 commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3600748646
##########
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:
Fixed in e0b6d792106: validate first_row_index, strictly increasing rows and
physical ranges, positive page sizes, and file-bounded column-chunk ownership
before publishing OffsetIndex.
##########
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:
Fixed in e0b6d792106: skip INDEX_PAGE and unknown auxiliary pages before
decoder initialization, while keeping OffsetIndex data-page ordinals unchanged.
##########
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:
Fixed in e0b6d792106: derive the pre-Arrow-3 Data Page V2 compressed-flag
compatibility from created_by and apply it consistently to cold reads and
page-cache admission/hits.
##########
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:
Fixed in e0b6d792106: nullable non-strict
numeric/date/time/timestamp/decimal conversion failures now append a default
and mark the exact output row NULL; strict and non-nullable paths still return
errors, including dictionary pages.
##########
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:
Fixed in e0b6d792106: scalar and levels-only COUNT readers now share signed,
overflow-safe, file-bounded Column Chunk range validation before constructing
streams.
##########
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:
Fixed in e0b6d792106: apply PARQUET-816 padding only to affected parquet-mr
versions, cap it at 100 bytes, and keep the padded range inside the file.
--
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]
