github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3601733085
##########
be/src/core/data_type_serde/data_type_decimal_serde.cpp:
##########
@@ -179,6 +188,140 @@ Status read_decimal_decoded_values(IColumn& column, const
DecodedColumnView& vie
return Status::OK();
}
+template <PrimitiveType T>
+Status scale_parquet_decimal(typename
PrimitiveTypeTraits<T>::CppType::NativeType value,
+ int32_t source_scale, int32_t target_scale,
+ typename
PrimitiveTypeTraits<T>::CppType::NativeType* result) {
+ using NativeType = typename PrimitiveTypeTraits<T>::CppType::NativeType;
+ DORIS_CHECK(result != nullptr);
+ if (source_scale == target_scale) {
+ *result = value;
+ return Status::OK();
+ }
+ if (source_scale > target_scale) {
+ *result = value / decimal_scale_multiplier<NativeType>(source_scale -
target_scale);
Review Comment:
[P1] Preserve scale-down conversion failures. Every decimal-to-decimal
schema change is now direct-materialized, but this branch divides and returns
OK even when discarded digits are non-zero. For example, raw `1234` from
DECIMAL(4,2) becomes raw `123` in DECIMAL(3,1), whereas
`DecimalToDecimalConverter` rejects the same input via `src_value % multiplier
!= 0`; nullable non-strict scans also lose their conversion-failure null.
Please reject a non-zero remainder before dividing and add plain/dictionary
tests for exact and lossy scale-down.
##########
be/src/format_v2/parquet/reader/native_column_reader.cpp:
##########
@@ -0,0 +1,657 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/parquet/reader/native_column_reader.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <limits>
+#include <ranges>
+#include <string>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/config.h"
+#include "core/assert_cast.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/parquet/vparquet_file_metadata.h"
+#include "format_v2/column_data.h"
+#include "format_v2/parquet/parquet_column_schema.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::parquet {
+namespace {
+
+constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20;
+
+DataTypePtr projected_type(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex* projection) {
+ if (!format::is_partial_projection(projection)) {
+ return schema.type;
+ }
+ switch (schema.kind) {
+ case ParquetColumnSchemaKind::PRIMITIVE:
+ return schema.type;
+ case ParquetColumnSchemaKind::STRUCT: {
+ DataTypes child_types;
+ Strings child_names;
+ child_types.reserve(projection->children.size());
+ child_names.reserve(projection->children.size());
+ for (const auto& child_projection : projection->children) {
+ const auto child_it = std::ranges::find_if(schema.children,
[&](const auto& child) {
+ return child->local_id == child_projection.local_id();
+ });
+ DORIS_CHECK(child_it != schema.children.end());
+ child_types.push_back(make_nullable(projected_type(**child_it,
&child_projection)));
+ child_names.push_back((*child_it)->name);
+ }
+ DataTypePtr type = std::make_shared<DataTypeStruct>(child_types,
child_names);
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ case ParquetColumnSchemaKind::LIST: {
+ DORIS_CHECK(schema.children.size() == 1);
+ const auto* child_projection =
+ format::find_child_projection(projection,
schema.children[0]->local_id);
+ DORIS_CHECK(child_projection != nullptr);
+ DataTypePtr type = std::make_shared<DataTypeArray>(
+ projected_type(*schema.children[0], child_projection));
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ case ParquetColumnSchemaKind::MAP: {
+ DORIS_CHECK(schema.children.size() == 2);
+ const auto* value_projection =
+ format::find_child_projection(projection,
schema.children[1]->local_id);
+ DORIS_CHECK(value_projection != nullptr);
+ DataTypePtr type = std::make_shared<DataTypeMap>(
+ make_nullable(schema.children[0]->type),
+ make_nullable(projected_type(*schema.children[1],
value_projection)));
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ }
+ DORIS_CHECK(false);
+ return nullptr;
+}
+
+const FieldSchema* find_child_field(const FieldSchema& parent, const
ParquetColumnSchema& child) {
+ auto field_it = std::ranges::find_if(parent.children, [&](const
FieldSchema& field) {
+ return (child.parquet_field_id >= 0 && field.field_id ==
child.parquet_field_id) ||
+ field.name == child.name;
+ });
+ return field_it == parent.children.end() ? nullptr : &*field_it;
+}
+
+void collect_projected_ids(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex* projection,
+ const FieldSchema& native_field,
std::set<uint64_t>* ids) {
+ DORIS_CHECK(ids != nullptr);
+ if (!format::is_partial_projection(projection)) {
+ return;
+ }
+ for (const auto& child_projection : projection->children) {
+ const auto schema_it = std::ranges::find_if(schema.children, [&](const
auto& child) {
+ return child->local_id == child_projection.local_id();
+ });
+ DORIS_CHECK(schema_it != schema.children.end());
+ const FieldSchema* child_field = find_child_field(native_field,
**schema_it);
+ DORIS_CHECK(child_field != nullptr);
+ ids->insert(child_field->get_column_id());
+ collect_projected_ids(**schema_it, &child_projection, *child_field,
ids);
+ }
+ if (schema.kind == ParquetColumnSchemaKind::MAP) {
+ DORIS_CHECK(!native_field.children.empty());
+ // MAP entry existence and offsets are owned by the key stream even
for value-only
+ // projections. Keep the key reader live so the native complex reader
can validate
+ // key/value entry alignment while constructing offsets.
+ ids->insert(native_field.children[0].get_column_id());
+ }
+}
+
+Status append_non_null_dictionary_values(MutableColumnPtr& target,
MutableColumnPtr values) {
+ DORIS_CHECK(target);
+ DORIS_CHECK(values);
+ const size_t value_count = values->size();
+ if (auto* nullable = check_and_get_column<ColumnNullable>(*target);
nullable != nullptr) {
+ nullable->get_nested_column().insert_range_from(*values, 0,
value_count);
+ auto& null_map = nullable->get_null_map_data();
+ null_map.resize_fill(null_map.size() + value_count, 0);
+ return Status::OK();
+ }
+ target->insert_range_from(*values, 0, value_count);
+ return Status::OK();
+}
+
+} // namespace
+
+NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema,
+ DataTypePtr projected_type,
+ ParquetColumnReaderProfile profile)
+ : ParquetColumnReader(schema, std::move(projected_type), profile),
+ _nested(schema.kind != ParquetColumnSchemaKind::PRIMITIVE) {}
+
+NativeColumnReader::~NativeColumnReader() {
+ (void)sync_native_profile();
+}
+
+Status NativeColumnReader::create(
+ const ParquetColumnSchema& column_schema, const
format::LocalColumnIndex* projection,
+ io::FileReaderSPtr file, const FileMetaData* metadata, int
row_group_id,
+ const std::vector<RowRange>& selected_ranges,
+ const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
+ const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState*
runtime_state,
+ bool enable_page_cache, const std::string& page_cache_file_key,
+ bool enable_dictionary_filter, ParquetColumnReaderProfile profile,
+ std::unique_ptr<ParquetColumnReader>* reader) {
+ if (reader == nullptr) {
+ return Status::InvalidArgument("Native parquet reader result is null");
+ }
+ if (file == nullptr || metadata == nullptr) {
+ return Status::InvalidArgument("Native parquet file context is not
initialized");
+ }
+ if (row_group_id < 0 ||
+ row_group_id >=
static_cast<int>(metadata->to_thrift().row_groups.size())) {
+ return Status::InvalidArgument("Invalid native parquet row group {}",
row_group_id);
+ }
+ const auto& native_schema = metadata->schema();
+ if (column_schema.local_id < 0 || column_schema.local_id >=
native_schema.size()) {
+ return Status::InvalidArgument("Invalid native parquet top-level
column id {} for {}",
+ column_schema.local_id,
column_schema.name);
+ }
+ auto* field =
const_cast<FieldSchema*>(native_schema.get_column(column_schema.local_id));
+ DORIS_CHECK(field != nullptr);
+ if (field->name != column_schema.name &&
+ !(field->field_id >= 0 && field->field_id ==
column_schema.parquet_field_id)) {
+ return Status::Corruption(
+ "Native/metadata parquet schema mismatch at column {}:
native={}, arrow={}",
+ column_schema.local_id, field->name, column_schema.name);
+ }
+
+ auto type = projected_type(column_schema, projection);
+ std::shared_ptr<TableSchemaChangeHelper::Node> schema_node;
+
RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_name(type,
*field,
+
schema_node));
+ std::set<uint64_t> projected_ids;
+ collect_projected_ids(column_schema, projection, *field, &projected_ids);
+
+ auto native_reader = std::unique_ptr<NativeColumnReader>(
+ new NativeColumnReader(column_schema, std::move(type), profile));
+ RETURN_IF_ERROR(native_reader->init(
+ std::move(file), metadata, row_group_id, field,
std::move(schema_node),
+ std::move(projected_ids), selected_ranges, offset_indexes,
timezone, io_ctx,
+ runtime_state, enable_page_cache, page_cache_file_key,
enable_dictionary_filter));
+ *reader = std::move(native_reader);
+ return Status::OK();
+}
+
+Status NativeColumnReader::init(
+ io::FileReaderSPtr file, const FileMetaData* metadata, int
row_group_id, FieldSchema* field,
+ std::shared_ptr<TableSchemaChangeHelper::Node> schema_node,
+ std::set<uint64_t> projected_column_ids, const std::vector<RowRange>&
selected_ranges,
+ const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
+ const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState*
runtime_state,
+ bool enable_page_cache, const std::string& page_cache_file_key,
+ bool enable_dictionary_filter) {
+ DORIS_CHECK(file != nullptr);
+ DORIS_CHECK(metadata != nullptr);
+ DORIS_CHECK(field != nullptr);
+ DORIS_CHECK(schema_node != nullptr);
+ const auto& row_group = metadata->to_thrift().row_groups[row_group_id];
+ DORIS_CHECK(row_group.num_rows > 0);
+ _row_group_rows = row_group.num_rows;
+ _selected_ranges = selected_ranges;
+ DORIS_CHECK(!_selected_ranges.empty());
+ for (const auto& range : _selected_ranges) {
+ DORIS_CHECK(range.start >= 0);
+ DORIS_CHECK(range.length > 0);
+ DORIS_CHECK(range.start + range.length <= _row_group_rows);
+ _row_ranges.add(::doris::RowRange(range.start, range.start +
range.length));
+ }
+ // Offset indexes are immutable row-group metadata owned by
ParquetScanScheduler. Sharing them
+ // avoids retaining the full N-column page-location map once per projected
reader.
+ _offset_indexes = &offset_indexes;
+ _schema_node = std::move(schema_node);
+ _projected_column_ids = std::move(projected_column_ids);
+ _dictionary_filter_enabled = enable_dictionary_filter;
+
+ const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb <<
20;
+ const size_t max_column_buffer = config::parquet_column_max_buffer_mb <<
20;
+ const size_t max_buffer_size = std::min(max_group_buffer,
max_column_buffer);
+ RuntimeState* native_runtime_state = runtime_state;
+ const bool runtime_page_cache_enabled =
+ runtime_state == nullptr ||
+ runtime_state->query_options().enable_parquet_file_page_cache;
+ if (runtime_page_cache_enabled != enable_page_cache) {
+ TQueryOptions query_options =
+ runtime_state == nullptr ? TQueryOptions() :
runtime_state->query_options();
+ query_options.__set_enable_parquet_file_page_cache(enable_page_cache);
+ _page_cache_runtime_state = RuntimeState::create_unique(query_options,
TQueryGlobals());
+ native_runtime_state = _page_cache_runtime_state.get();
+ }
+ const auto& thrift_metadata = metadata->to_thrift();
+ const auto compat = native::parquet_reader_compat(
+ thrift_metadata.__isset.created_by ? thrift_metadata.created_by :
"");
+ RETURN_IF_ERROR(native::ColumnReader::create(
+ std::move(file), field, row_group, _row_ranges, timezone, io_ctx,
_native_reader,
+ max_buffer_size, *_offset_indexes, native_runtime_state, false,
_projected_column_ids,
+ _filter_column_ids, page_cache_file_key, compat,
+ runtime_state != nullptr && runtime_state->enable_strict_mode()));
+ DORIS_CHECK(_native_reader != nullptr);
+ _skip_column = _type->create_column();
+ return Status::OK();
+}
+
+Status NativeColumnReader::read_with_filter(int64_t rows, const uint8_t*
filter_data,
+ bool filter_all, MutableColumnPtr&
column,
+ const DataTypePtr& output_type,
bool dictionary_ids,
+ int64_t* rows_read) {
+ DORIS_CHECK(rows >= 0);
+ DORIS_CHECK(column);
+ DORIS_CHECK(output_type != nullptr);
+ DORIS_CHECK(rows_read != nullptr);
+ *rows_read = 0;
+ if (rows == 0) {
+ return Status::OK();
+ }
+
+ ::doris::FilterMap filter;
+ RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows),
filter_all));
+ _native_reader->reset_filter_map_index();
+ ColumnPtr native_column(std::move(column));
+ bool eof = false;
+ int64_t native_calls = 0;
+ int64_t consecutive_empty_calls = 0;
+ while (*rows_read < rows && !eof) {
+ ++native_calls;
+ size_t loop_rows = 0;
+ RETURN_IF_ERROR(_native_reader->read_column_data(
+ native_column, output_type, _schema_node, filter,
+ static_cast<size_t>(rows - *rows_read), &loop_rows, &eof,
dictionary_ids));
+ if (loop_rows == 0 && !eof) {
+ // A selected RowRanges plan may reject the current data page
completely. V1 advances
+ // the page cursor and deliberately returns zero rows so the
caller can request the
+ // next page. Bound consecutive empty transitions by the Row Group
row count to retain
+ // a deterministic corruption exit if a decoder ever stops
advancing.
+ if (++consecutive_empty_calls > _row_group_rows + 1) {
+ column = IColumn::mutate(std::move(native_column));
+ return Status::Corruption("Native parquet reader made no
progress for column {}",
+ _name);
+ }
+ continue;
+ }
+ consecutive_empty_calls = 0;
+ *rows_read += static_cast<int64_t>(loop_rows);
+ }
+ column = IColumn::mutate(std::move(native_column));
+ if (_profile.native_read_calls != nullptr) {
+ COUNTER_UPDATE(_profile.native_read_calls, native_calls);
+ }
+ if (_nested && _profile.nested_batches != nullptr) {
+ COUNTER_UPDATE(_profile.nested_batches, 1);
+ }
+ _native_reader->release_batch_scratch(MAX_RETAINED_BATCH_SCRATCH_BYTES);
+ if (*rows_read != rows) {
+ return Status::Corruption("Native parquet reader returned {} rows,
expected {} for {}",
+ *rows_read, rows, _name);
+ }
+ return Status::OK();
+}
+
+Status NativeColumnReader::validate_selected_span(int64_t rows) {
+ DORIS_CHECK(rows >= 0);
+ while (_selected_range_idx < _selected_ranges.size()) {
+ const auto& range = _selected_ranges[_selected_range_idx];
+ const int64_t range_end = range.start + range.length;
+ if (_logical_row_position < range_end) {
+ break;
+ }
+ ++_selected_range_idx;
+ }
+ if (_selected_range_idx >= _selected_ranges.size()) {
+ return Status::Corruption("Native parquet read past selected ranges
for column {}", _name);
+ }
+ const auto& range = _selected_ranges[_selected_range_idx];
+ if (_logical_row_position < range.start ||
+ rows > range.start + range.length - _logical_row_position) {
+ return Status::Corruption(
+ "Native parquet read [{}, {}) crosses selected range [{}, {})
for column {}",
+ _logical_row_position, _logical_row_position + rows,
range.start,
+ range.start + range.length, _name);
+ }
+ return Status::OK();
+}
+
+void NativeColumnReader::advance_selected_span(int64_t rows) {
+ _logical_row_position += rows;
+ while (_selected_range_idx < _selected_ranges.size() &&
+ _logical_row_position >=
_selected_ranges[_selected_range_idx].start +
+
_selected_ranges[_selected_range_idx].length) {
+ ++_selected_range_idx;
+ }
+}
+
+Status NativeColumnReader::read(int64_t rows, MutableColumnPtr& column,
int64_t* rows_read) {
+ RETURN_IF_ERROR(validate_selected_span(rows));
+ RETURN_IF_ERROR(read_with_filter(rows, nullptr, false, column, _type,
false, rows_read));
+ advance_selected_span(*rows_read);
+ update_reader_read_rows(*rows_read);
+ return Status::OK();
+}
+
+Status NativeColumnReader::skip(int64_t rows) {
+ if (rows <= 0) {
+ return Status::OK();
+ }
+ DORIS_CHECK(_logical_row_position <= _row_group_rows - rows);
+ int64_t remaining = rows;
+ int64_t native_skipped_rows = 0;
+ while (remaining > 0) {
+ while (_selected_range_idx < _selected_ranges.size() &&
+ _logical_row_position >=
_selected_ranges[_selected_range_idx].start +
+
_selected_ranges[_selected_range_idx].length) {
+ ++_selected_range_idx;
+ }
+ if (_selected_range_idx >= _selected_ranges.size()) {
+ _logical_row_position += remaining;
+ break;
+ }
+ const auto& range = _selected_ranges[_selected_range_idx];
+ if (_logical_row_position < range.start) {
+ const int64_t gap = std::min(remaining, range.start -
_logical_row_position);
+ _logical_row_position += gap;
+ remaining -= gap;
+ continue;
+ }
+ const int64_t selected_rows =
+ std::min(remaining, range.start + range.length -
_logical_row_position);
+ _skip_column->clear();
+ // resize() preserves survivor bytes from the previous select(). An
all-filtered nested read
+ // consumes the raw bitmap, so every slot must be explicitly reset
before a lazy skip.
+ _filter_scratch.assign(static_cast<size_t>(selected_rows), 0);
Review Comment:
[P1] Bound the bitmap for accumulated lazy skips. The scheduler accumulates
every fully filtered batch in `_pending_non_predicate_skip_rows`, then sends
the complete prefix to every lazy reader at the first surviving batch. This
`assign` therefore allocates one byte per pending row per lazy column and
retains that capacity for the row-group lifetime; the 4 MiB native decoder
scratch cap does not cover this adapter-owned vector. A valid long filtered
prefix with a wide lazy projection can consume gigabytes before producing the
survivor. Please chunk the pending skip into bounded spans, or make the
all-filtered path avoid a dense map, and cover a long-prefix multi-column case.
##########
be/src/format_v2/parquet/reader/native/level_decoder.cpp:
##########
@@ -0,0 +1,239 @@
+// 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/level_decoder.h"
+
+#include <gen_cpp/parquet_types.h>
+
+#include <algorithm>
+
+#include "common/cast_set.h"
+#include "format/parquet/parquet_common.h"
+#include "util/bit_stream_utils.inline.h"
+#include "util/bit_util.h"
+#include "util/coding.h"
+
+namespace doris::format::parquet::native {
+
+static constexpr size_t V1_LEVEL_SIZE = 4;
+
+Status LevelDecoder::init(Slice* slice, tparquet::Encoding::type encoding,
level_t max_level,
+ uint32_t num_levels) {
+ _encoding = encoding;
+ _bit_width = cast_set<level_t>(BitUtil::log2(max_level + 1));
+ _max_level = max_level;
+ _num_levels = num_levels;
+ _has_buffered_level = false;
+ _can_rewind = false;
+ switch (encoding) {
+ case tparquet::Encoding::RLE: {
+ if (slice->size < V1_LEVEL_SIZE) {
+ return Status::Corruption("Wrong parquet level format");
+ }
+
+ uint8_t* data = (uint8_t*)slice->data;
+ uint32_t num_bytes = decode_fixed32_le(data);
+ if (num_bytes > slice->size - V1_LEVEL_SIZE) {
+ return Status::Corruption("Wrong parquet level format");
+ }
+ _rle_decoder = RleBatchDecoder<uint16_t>(data + V1_LEVEL_SIZE,
num_bytes, _bit_width);
+
+ slice->data += V1_LEVEL_SIZE + num_bytes;
+ slice->size -= V1_LEVEL_SIZE + num_bytes;
+ break;
+ }
+ case tparquet::Encoding::BIT_PACKED: {
+ uint32_t num_bits = num_levels * _bit_width;
+ uint32_t num_bytes = BitUtil::RoundUpNumBytes(num_bits);
+ if (num_bytes > slice->size) {
+ return Status::Corruption("Wrong parquet level format");
+ }
+ _bit_packed_decoder = BitReader((uint8_t*)slice->data, num_bytes);
+
+ slice->data += num_bytes;
+ slice->size -= num_bytes;
+ break;
+ }
+ default:
+ return Status::IOError("Unsupported encoding for parquet level");
+ }
+ return Status::OK();
+}
+
+Status LevelDecoder::init_v2(const Slice& levels, level_t max_level, uint32_t
num_levels) {
+ _encoding = tparquet::Encoding::RLE;
+ _bit_width = cast_set<level_t>(BitUtil::log2(max_level + 1));
+ _max_level = max_level;
+ _num_levels = num_levels;
+ _has_buffered_level = false;
+ _can_rewind = false;
+ size_t byte_length = levels.size;
+ _rle_decoder = RleBatchDecoder<uint16_t>((uint8_t*)levels.data,
cast_set<int>(byte_length),
+ _bit_width);
+ return Status::OK();
+}
+
+size_t LevelDecoder::get_levels(level_t* levels, size_t n) {
+ _can_rewind = false;
+ // toto template.
+ if (_encoding == tparquet::Encoding::RLE) {
+ n = std::min((size_t)_num_levels, n);
+ size_t num_decoded = 0;
+ if (_has_buffered_level && n > 0) {
+ levels[num_decoded++] = _buffered_level;
+ _has_buffered_level = false;
+ }
+ if (num_decoded < n) {
+ const size_t remaining = n - num_decoded;
+ _rle_scratch.resize(remaining);
+ const size_t batch_decoded =
+ _rle_decoder.GetBatch(_rle_scratch.data(),
cast_set<uint32_t>(remaining));
+ for (size_t i = 0; i < batch_decoded; ++i) {
+ levels[num_decoded + i] = cast_set<level_t>(_rle_scratch[i]);
Review Comment:
[P1] Reject levels above the schema maximum. The decoder stores
`_max_level`, but this batch path and the run/single-value paths return any
bit-width-representable value without checking it. For example, `max_level ==
2` uses two bits, so malformed RLE or BIT_PACKED data can encode 3; nested
readers then accept it through `def_level >= definition_level` and consume
payload as a present element instead of reporting corruption. Please validate
decoded values in every cursor path and add RLE/BIT_PACKED nested cases for an
out-of-range level.
##########
be/src/core/data_type_serde/data_type_decimal_serde.cpp:
##########
@@ -179,6 +188,140 @@ Status read_decimal_decoded_values(IColumn& column, const
DecodedColumnView& vie
return Status::OK();
}
+template <PrimitiveType T>
+Status scale_parquet_decimal(typename
PrimitiveTypeTraits<T>::CppType::NativeType value,
+ int32_t source_scale, int32_t target_scale,
+ typename
PrimitiveTypeTraits<T>::CppType::NativeType* result) {
+ using NativeType = typename PrimitiveTypeTraits<T>::CppType::NativeType;
+ DORIS_CHECK(result != nullptr);
+ if (source_scale == target_scale) {
+ *result = value;
+ return Status::OK();
+ }
+ if (source_scale > target_scale) {
+ *result = value / decimal_scale_multiplier<NativeType>(source_scale -
target_scale);
+ return Status::OK();
+ }
+ const auto multiplier = decimal_scale_multiplier<NativeType>(target_scale
- source_scale);
+ if (common::mul_overflow(value, multiplier, *result)) {
+ return Status::DataQualityError("Parquet decimal overflows while
scaling from {} to {}",
+ source_scale, target_scale);
+ }
+ return Status::OK();
+}
+
+template <PrimitiveType T>
+class DecimalParquetConsumer final : public ParquetFixedValueConsumer,
+ public ParquetBinaryValueConsumer {
+public:
+ using FieldType = typename PrimitiveTypeTraits<T>::CppType;
+ using NativeType = typename FieldType::NativeType;
+
+ DecimalParquetConsumer(IColumn& column, const ParquetDecodeContext&
context,
+ UInt32 target_precision, int32_t target_scale,
+ ParquetMaterializationState* state = nullptr)
+ : _data(assert_cast<ColumnDecimal<T>&>(column).get_data()),
+ _context(context),
+ _target_precision(target_precision),
+ _target_scale(target_scale),
+ _state(state) {}
+
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ if (_context.physical_type == ParquetPhysicalType::INT32) {
+ DORIS_CHECK_EQ(value_width, sizeof(int32_t));
+ return append_integers<int32_t>(values, num_values);
+ }
+ if (_context.physical_type == ParquetPhysicalType::INT64) {
+ DORIS_CHECK_EQ(value_width, sizeof(int64_t));
+ return append_integers<int64_t>(values, num_values);
+ }
+ if (_context.physical_type !=
ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY) {
+ return Status::NotSupported("Unsupported Parquet physical type {}
for decimal SerDe",
+
static_cast<int>(_context.physical_type));
+ }
+ DORIS_CHECK_EQ(value_width, static_cast<size_t>(_context.type_length));
+ const size_t old_size = _data.size();
+ _data.resize(old_size + num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ auto status =
+ append_binary_value(values + row * value_width,
value_width, old_size + row);
+ if (!status.ok()) {
+ if (_state != nullptr &&
_state->mark_conversion_failure(old_size + row)) {
+ _data[old_size + row] = FieldType();
+ continue;
+ }
+ _data.resize(old_size);
+ return status;
+ }
+ }
+ return Status::OK();
+ }
+
+ Status consume(const StringRef* values, size_t num_values) override {
+ const size_t old_size = _data.size();
+ _data.resize(old_size + num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ auto status = append_binary_value(reinterpret_cast<const
uint8_t*>(values[row].data),
+ values[row].size, old_size +
row);
+ if (!status.ok()) {
+ if (_state != nullptr &&
_state->mark_conversion_failure(old_size + row)) {
+ _data[old_size + row] = FieldType();
+ continue;
+ }
+ _data.resize(old_size);
+ return status;
+ }
+ }
+ return Status::OK();
+ }
+
+private:
+ template <typename SourceType>
+ Status append_integers(const uint8_t* values, size_t num_values) {
+ const size_t old_size = _data.size();
+ _data.resize(old_size + num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ const auto source_value = unaligned_load<SourceType>(values + row
* sizeof(SourceType));
+ auto status = append_native_value(NativeType(source_value),
old_size + row);
Review Comment:
[P1] Keep decimal values wide until conversion succeeds. Because every
decimal source/target pair is direct-materialized, this cast narrows the
physical INT64 value before rescaling or checking target precision. For
example, raw `4,294,967,297` from DECIMAL(10,0) becomes `1` in DECIMAL32(9,0)
and is accepted instead of failing/nulling; the binary sibling also rejects
wider sign-extended encodings that could fit after exact scale-down. Please
decode/rescale in a source- or max-width intermediate, validate there, and
narrow only on success, with plain/dictionary integer and binary narrowing
tests.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1122 @@
+// 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 <parquet/metadata.h>
+#include <string.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#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 {
+
+ParquetReaderCompat parquet_reader_compat(const std::string& created_by) {
+ if (created_by.empty()) {
+ return {};
+ }
+ const ::parquet::ApplicationVersion version(created_by);
+ return {.parquet_816_padding =
+
version.VersionLt(::parquet::ApplicationVersion::PARQUET_816_FIXED_VERSION()),
+ .data_page_v2_always_compressed = version.VersionLt(
+
::parquet::ApplicationVersion::PARQUET_CPP_10353_FIXED_VERSION())};
+}
+
+Status compute_column_chunk_range(const tparquet::ColumnMetaData& metadata,
size_t file_size,
+ bool parquet_816_padding, ColumnChunkRange*
range) {
+ DORIS_CHECK(range != nullptr);
+ int64_t start = metadata.data_page_offset;
+ if (metadata.__isset.dictionary_page_offset &&
metadata.dictionary_page_offset >= 0 &&
+ metadata.dictionary_page_offset < start) {
+ start = metadata.dictionary_page_offset;
+ }
+ const int64_t length = metadata.total_compressed_size;
+ if (UNLIKELY(start < 0 || length < 0)) {
+ return Status::Corruption("Parquet column chunk has a negative offset
or length");
+ }
+ const uint64_t unsigned_start = static_cast<uint64_t>(start);
+ const uint64_t unsigned_length = static_cast<uint64_t>(length);
+ if (UNLIKELY(unsigned_start > file_size || unsigned_length > file_size -
unsigned_start)) {
+ // Thrift range fields are signed and untrusted; validate before
converting them to the
+ // unsigned stream-reader coordinates so overflow cannot wrap back
into the file.
+ return Status::Corruption("Parquet column chunk [{}, {}) exceeds file
size {}", start,
+ unsigned_start + unsigned_length, file_size);
+ }
+ size_t bounded_length = static_cast<size_t>(unsigned_length);
+ if (parquet_816_padding) {
+ // parquet-mr before PARQUET-816 under-reported the chunk by up to 100
bytes. Padding stays
+ // file-bounded and is only enabled for the affected writer versions.
+ bounded_length += std::min<size_t>(100, file_size - unsigned_start -
unsigned_length);
+ }
+ range->offset = static_cast<size_t>(unsigned_start);
+ range->length = bounded_length;
+ return Status::OK();
+}
+
+bool validate_offset_index(const tparquet::OffsetIndex& index, const
ColumnChunkRange& chunk_range,
+ int64_t data_page_offset, int64_t row_count) {
+ if (index.page_locations.empty() || data_page_offset < 0 || row_count < 0
||
+ index.page_locations.front().first_row_index != 0 ||
+ index.page_locations.front().offset != data_page_offset ||
+ chunk_range.length > std::numeric_limits<size_t>::max() -
chunk_range.offset) {
+ return false;
+ }
+ // Row indexes alone cannot detect a uniformly shifted OffsetIndex. Anchor
its first location
+ // to the owning metadata so page-to-row mapping cannot silently move by
one physical page.
+ const uint64_t chunk_begin = chunk_range.offset;
+ const uint64_t chunk_end = chunk_begin + chunk_range.length;
+ uint64_t previous_end = chunk_begin;
+ int64_t previous_row = -1;
+ for (const auto& location : index.page_locations) {
+ if (location.first_row_index <= previous_row ||
location.first_row_index >= row_count ||
+ location.offset < 0 || location.compressed_page_size <= 0) {
+ return false;
+ }
+ const uint64_t begin = static_cast<uint64_t>(location.offset);
+ const uint64_t size =
static_cast<uint64_t>(location.compressed_page_size);
+ if (begin < chunk_begin || begin < previous_end || begin > chunk_end ||
+ size > chunk_end - begin) {
+ return false;
+ }
+ previous_row = location.first_row_index;
+ previous_end = begin + size;
+ }
+ return true;
+}
+
+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,
+ const ColumnChunkRange* chunk_range)
+ : _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) {
+ if (chunk_range != nullptr) {
+ _chunk_range = *chunk_range;
+ _has_validated_chunk_range = true;
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() {
+ size_t start_offset = _has_validated_chunk_range
+ ? _chunk_range.offset
+ : (has_dict_page(_metadata) ?
_metadata.dictionary_page_offset
+ :
_metadata.data_page_offset);
+ size_t chunk_size =
+ _has_validated_chunk_range ? _chunk_range.length :
_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() {
+ while (true) {
+ RETURN_IF_ERROR(_page_reader->parse_page_header());
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ if (header->type == tparquet::PageType::DATA_PAGE ||
+ header->type == tparquet::PageType::DATA_PAGE_V2) {
+ _state = INITIALIZED;
+ return parse_page_header();
+ }
+ if (header->type != tparquet::PageType::DICTIONARY_PAGE) {
+ RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+ _state = INITIALIZED;
+ continue;
+ }
+ // 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;
+ // A dictionary is the only non-data page with decoder state. Any
following index or
+ // extension pages are skipped by the same pre-data loop.
+ }
+}
+
+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();
+ }
+ const tparquet::PageHeader* header = nullptr;
+ while (true) {
+ RETURN_IF_ERROR(_page_reader->parse_page_header());
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ if (header->type == tparquet::PageType::DATA_PAGE ||
+ header->type == tparquet::PageType::DATA_PAGE_V2) {
+ break;
+ }
+ if (header->type == tparquet::PageType::DICTIONARY_PAGE) {
+ return Status::Corruption("Parquet dictionary page appears after
data pages");
+ }
+ RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+ }
+ 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
||
+
_page_read_ctx.data_page_v2_always_compressed);
+ bool page_has_compression = header->__isset.data_page_header ||
is_v2_compressed;
+
+ if (page_has_compression) {
+ // Decompress payload for immediate decoding
+ _reserve_decompress_buf(uncompressed_size);
+ _page_data = Slice(_decompress_buf.get(), uncompressed_size);
+ SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+ _chunk_statistics.decompress_cnt++;
+
RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data,
&_page_data));
+ if (UNLIKELY(_page_data.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption("Parquet page decompressed to {}
bytes, expected {}",
+ _page_data.size,
uncompressed_size);
+ }
+
+ // Decide whether to cache decompressed payload or compressed
payload based on threshold
+ bool cache_payload_decompressed = should_cache_decompressed(
+ header, _metadata,
_page_read_ctx.data_page_v2_always_compressed);
+
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
+ StoragePageCache::instance() != nullptr &&
+ !_page_reader->header_bytes().empty()) {
+ if (cache_payload_decompressed) {
+ _insert_page_into_cache(level_bytes, _page_data);
+
_chunk_statistics.page_cache_decompressed_write_counter += 1;
+ } else {
+ if (config::enable_parquet_cache_compressed_pages) {
+ // cache the compressed payload as-is (header |
levels | compressed_payload)
+ _insert_page_into_cache(
+ level_bytes, Slice(compressed_data.data,
compressed_data.size));
+
_chunk_statistics.page_cache_compressed_write_counter += 1;
+ }
+ }
+ }
+ } else {
+ // no compression on this page, use the data directly
+ _page_data = Slice(compressed_data.data, compressed_data.size);
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
+ StoragePageCache::instance() != nullptr) {
+ _insert_page_into_cache(level_bytes, _page_data);
+ _chunk_statistics.page_cache_decompressed_write_counter +=
1;
+ }
+ }
+ } else {
+ // For uncompressed page, we may still need to extract v2 levels
+ std::vector<uint8_t> level_bytes;
+ Slice uncompressed_data;
+ RETURN_IF_ERROR(_page_reader->get_page_data(uncompressed_data));
+ 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;
+ size_t level_sz = rl + dl;
+ if (level_sz > 0) {
+ level_bytes.resize(level_sz);
+ memcpy(level_bytes.data(), uncompressed_data.data,
level_sz);
+ }
+ RETURN_IF_ERROR(_get_uncompressed_levels(header_v2,
uncompressed_data));
+ }
+ // copy page data out
+ _page_data = Slice(uncompressed_data.data, uncompressed_data.size);
+ // Optionally cache uncompressed data for uncompressed pages
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
StoragePageCache::instance() != nullptr) {
+ _insert_page_into_cache(level_bytes, _page_data);
+ _chunk_statistics.page_cache_decompressed_write_counter += 1;
+ }
+ }
+ }
+
+ // Initialize repetition level and definition level. Skip when level = 0,
which means required field.
+ if (_max_rep_level > 0) {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+ if (header->__isset.data_page_header_v2) {
+ RETURN_IF_ERROR(_rep_level_decoder.init_v2(_v2_rep_levels,
_max_rep_level,
+ _remaining_rep_nums));
+ } else {
+ RETURN_IF_ERROR(_rep_level_decoder.init(
+ &_page_data,
header->data_page_header.repetition_level_encoding, _max_rep_level,
+ _remaining_rep_nums));
+ }
+ }
+ if (_max_def_level > 0) {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+ if (header->__isset.data_page_header_v2) {
+ RETURN_IF_ERROR(_def_level_decoder.init_v2(_v2_def_levels,
_max_def_level,
+ _remaining_def_nums));
+ } else {
+ RETURN_IF_ERROR(_def_level_decoder.init(
+ &_page_data,
header->data_page_header.definition_level_encoding, _max_def_level,
+ _remaining_def_nums));
+ }
+ }
+ auto encoding = header->__isset.data_page_header_v2 ?
header->data_page_header_v2.encoding
+ :
header->data_page_header.encoding;
+ // change the deprecated encoding to RLE_DICTIONARY
+ if (encoding == tparquet::Encoding::PLAIN_DICTIONARY) {
+ encoding = tparquet::Encoding::RLE_DICTIONARY;
+ }
+ _current_encoding = encoding;
+
+ // Reuse page decoder
+ if (_decoders.find(static_cast<int>(encoding)) != _decoders.end()) {
+ _page_decoder = _decoders[static_cast<int>(encoding)].get();
+ } else {
+ std::unique_ptr<Decoder> page_decoder;
+ RETURN_IF_ERROR(Decoder::get_decoder(_metadata.type, encoding,
page_decoder));
+ // Set type length
+ page_decoder->set_type_length(_get_type_length());
+ _decoders[static_cast<int>(encoding)] = std::move(page_decoder);
+ _page_decoder = _decoders[static_cast<int>(encoding)].get();
+ }
+ // Encoding headers cannot legitimately advertise more physical values
than the data page's
+ // logical value count; establish the bound before decoders inspect
external counts.
+ _page_decoder->set_expected_values(_remaining_num_values);
+ RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));
+
+ _state = DATA_LOADED;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_decode_dict_page() {
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ DCHECK_EQ(tparquet::PageType::DICTIONARY_PAGE, header->type);
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_dict_time);
+
+ // Using the PLAIN_DICTIONARY enum value is deprecated in the Parquet 2.0
specification.
+ // Prefer using RLE_DICTIONARY in a data page and PLAIN in a dictionary
page for Parquet 2.0+ files.
+ // refer: https://github.com/apache/parquet-format/blob/master/Encodings.md
+ tparquet::Encoding::type dict_encoding =
header->dictionary_page_header.encoding;
+ if (dict_encoding != tparquet::Encoding::PLAIN_DICTIONARY &&
+ dict_encoding != tparquet::Encoding::PLAIN) {
+ return Status::InternalError("Unsupported dictionary encoding {}",
+ tparquet::to_string(dict_encoding));
+ }
+
+ // Prepare dictionary data
+ int32_t uncompressed_size = header->uncompressed_page_size;
+ if (_block_compress_codec == nullptr &&
+ UNLIKELY(header->compressed_page_size != uncompressed_size)) {
+ // UNCOMPRESSED pages use the compressed size as their physical copy
length.
+ return Status::Corruption(
+ "Uncompressed Parquet dictionary sizes differ: compressed={},
uncompressed={}",
+ header->compressed_page_size, uncompressed_size);
+ }
+ auto dict_data = make_unique_buffer<uint8_t>(uncompressed_size);
+ bool dict_loaded = false;
+
+ // Try to load dictionary page from cache
+ 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();
+ // Dictionary page layout in cache: header | payload (compressed
or uncompressed)
+ Slice payload_slice(cached.data + header_size, cached.size -
header_size);
+
+ bool cache_payload_is_decompressed =
_page_reader->is_cache_payload_decompressed();
+
+ if (cache_payload_is_decompressed) {
+ // Use cached decompressed dictionary data
+ if (UNLIKELY(payload_slice.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption(
+ "Cached Parquet dictionary payload has size {},
expected {}",
+ payload_slice.size, uncompressed_size);
+ }
+ memcpy(dict_data.get(), payload_slice.data,
payload_slice.size);
+ dict_loaded = true;
+ } else {
+ CHECK(_block_compress_codec);
+ // Decompress cached compressed dictionary data
+ Slice dict_slice(dict_data.get(), uncompressed_size);
+
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &dict_slice));
+ if (UNLIKELY(dict_slice.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption(
+ "Parquet dictionary decompressed to {} bytes,
expected {}",
+ dict_slice.size, uncompressed_size);
+ }
+ dict_loaded = true;
+ }
+
+ // When dictionary page is loaded from cache, we need to skip the
page data
+ // to update the offset correctly (similar to calling
get_page_data())
+ if (dict_loaded) {
+ _page_reader->skip_page_data();
+ }
+ }
+ }
+
+ if (!dict_loaded) {
+ // Load and decompress dictionary page from file
+ if (_block_compress_codec != nullptr) {
+ auto dict_num = header->dictionary_page_header.num_values;
+ if (dict_num == 0 && uncompressed_size != 0) {
+ return Status::IOError(
+ "Dictionary page's num_values is {} but
uncompressed_size is {}", dict_num,
+ uncompressed_size);
+ }
+ Slice compressed_data;
+ Slice dict_slice(dict_data.get(), uncompressed_size);
+ if (dict_num != 0) {
+ RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+
RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data,
&dict_slice));
+ if (UNLIKELY(dict_slice.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption(
+ "Parquet dictionary decompressed to {} bytes,
expected {}",
+ dict_slice.size, uncompressed_size);
+ }
+ }
+
+ // Decide whether to cache decompressed or compressed dictionary
based on threshold
+ // If uncompressed_page_size == 0, should_cache_decompressed will
return true
+ bool cache_payload_decompressed = should_cache_decompressed(
+ header, _metadata,
_page_read_ctx.data_page_v2_always_compressed);
+
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
StoragePageCache::instance() != nullptr &&
+ !_page_reader->header_bytes().empty()) {
+ std::vector<uint8_t> empty_levels; // Dictionary pages don't
have levels
+ if (cache_payload_decompressed) {
+ // Cache the decompressed dictionary page
+ // If dict_num == 0, `dict_slice` will be empty
+ _insert_page_into_cache(empty_levels, dict_slice);
+ _chunk_statistics.page_cache_decompressed_write_counter +=
1;
+ } else {
+ if (config::enable_parquet_cache_compressed_pages) {
+ DCHECK(!compressed_data.empty());
+ // Cache the compressed dictionary page
+ _insert_page_into_cache(empty_levels,
+ Slice(compressed_data.data,
compressed_data.size));
+ _chunk_statistics.page_cache_compressed_write_counter
+= 1;
+ }
+ }
+ }
+ // `get_page_data` not called, we should skip the page data
+ // Because `_insert_page_into_cache` will use _page_reader, we
should exec `skip_page_data` after `_insert_page_into_cache`
+ if (dict_num == 0) {
+ _page_reader->skip_page_data();
+ }
+ } else {
+ Slice dict_slice;
+ RETURN_IF_ERROR(_page_reader->get_page_data(dict_slice));
+ // The data is stored by BufferedStreamReader, we should copy it
out
+ memcpy(dict_data.get(), dict_slice.data, dict_slice.size);
+
+ // Cache the uncompressed dictionary page
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
StoragePageCache::instance() != nullptr &&
+ !_page_reader->header_bytes().empty()) {
+ std::vector<uint8_t> empty_levels;
+ Slice payload(dict_data.get(), uncompressed_size);
+ _insert_page_into_cache(empty_levels, payload);
+ _chunk_statistics.page_cache_decompressed_write_counter += 1;
+ }
+ }
+ }
+
+ // Cache page decoder
+ std::unique_ptr<Decoder> page_decoder;
+ RETURN_IF_ERROR(
+ Decoder::get_decoder(_metadata.type,
tparquet::Encoding::RLE_DICTIONARY, page_decoder));
+ // Set type length
+ page_decoder->set_type_length(_get_type_length());
+ // Set the dictionary data
+ RETURN_IF_ERROR(page_decoder->set_dict(dict_data, uncompressed_size,
+
header->dictionary_page_header.num_values));
+ _decoders[static_cast<int>(tparquet::Encoding::RLE_DICTIONARY)] =
std::move(page_decoder);
+
+ _has_dict = true;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_reserve_decompress_buf(size_t size) {
+ if (size > _decompress_buf_size) {
+ _decompress_buf_size = BitUtil::next_power_of_two(size);
+ _decompress_buf = make_unique_buffer<uint8_t>(_decompress_buf_size);
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_insert_page_into_cache(
+ const std::vector<uint8_t>& level_bytes, const Slice& payload) {
+ StoragePageCache::CacheKey key =
+
_page_reader->make_page_cache_key(_page_reader->header_start_offset());
+ const std::vector<uint8_t>& header_bytes = _page_reader->header_bytes();
+ size_t total = header_bytes.size() + level_bytes.size() + payload.size;
+ auto page = std::make_unique<DataPage>(total, true, segment_v2::DATA_PAGE);
+ size_t pos = 0;
+ memcpy(page->data() + pos, header_bytes.data(), header_bytes.size());
+ pos += header_bytes.size();
+ if (!level_bytes.empty()) {
+ memcpy(page->data() + pos, level_bytes.data(), level_bytes.size());
+ pos += level_bytes.size();
+ }
+ if (payload.size > 0) {
+ memcpy(page->data() + pos, payload.data, payload.size);
+ pos += payload.size;
+ }
+ page->reset_size(total);
+ PageCacheHandle handle;
+ StoragePageCache::instance()->insert(key, page.get(), &handle,
segment_v2::DATA_PAGE);
+ page.release();
+ _chunk_statistics.page_cache_write_counter += 1;
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_values(size_t
num_values,
+ bool
skip_data) {
+ if (UNLIKELY(_remaining_num_values < num_values)) {
+ return Status::IOError("Skip too many values in current page. {} vs.
{}",
+ _remaining_num_values, num_values);
+ }
+ if (skip_data) {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+ RETURN_IF_ERROR(_page_decoder->skip_values(num_values));
+ }
+ // Commit logical page progress only after the physical decoder accepted
the whole request.
+ _remaining_num_values -= num_values;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::materialize_values(
+ MutableColumnPtr& doris_column, const DataTypeSerDe& serde,
ParquetDecodeContext& context,
+ ParquetMaterializationState& state, ColumnSelectVector& select_vector)
{
+ if (select_vector.num_values() == 0) {
+ return Status::OK();
+ }
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+ if (UNLIKELY((doris_column->is_column_dictionary() ||
context.dictionary_index_only) &&
+ !_has_dict)) {
+ return Status::IOError("Not dictionary coded");
+ }
+ if (UNLIKELY(_remaining_num_values < select_vector.num_values())) {
+ return Status::IOError("Decode too many values in current page");
+ }
+ RETURN_IF_ERROR(translate_value_encoding(_current_encoding,
&context.encoding));
+ Status status;
+ if (select_vector.has_filter()) {
+ if (select_vector.num_nulls() == 0) {
+ ++_chunk_statistics.hybrid_selection_batches;
+ status = decode_selected_non_null_values(*doris_column, serde,
*_page_decoder, context,
+ state, select_vector,
+
&_chunk_statistics.materialization_time);
+ _chunk_statistics.hybrid_selection_ranges +=
state.selection.ranges.size();
+ } else {
+ ++_chunk_statistics.hybrid_selection_null_fallback_batches;
+ status = decode_selected_values<true>(*doris_column, serde,
*_page_decoder, context,
+ state, select_vector,
+
&_chunk_statistics.materialization_time);
+ }
+ } else {
+ status = decode_selected_values<false>(*doris_column, serde,
*_page_decoder, context, state,
+ select_vector,
+
&_chunk_statistics.materialization_time);
+ }
+ RETURN_IF_ERROR(status);
+ _remaining_num_values -= select_vector.num_values();
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::seek_to_nested_row(size_t left_row) {
+ if constexpr (OFFSET_INDEX) {
+ while (true) {
+ if (_page_reader->start_row() <= left_row && left_row <
_page_reader->end_row()) {
+ break;
+ } else if (has_next_page()) {
+ RETURN_IF_ERROR(next_page());
+ _current_row = _page_reader->start_row();
+ } else [[unlikely]] {
+ return Status::InternalError("no match seek row {}, current
row {}", left_row,
+ _current_row);
+ }
+ };
+
+ RETURN_IF_ERROR(parse_page_header());
+ RETURN_IF_ERROR(load_page_data());
+ RETURN_IF_ERROR(_skip_nested_rows_in_page(left_row - _current_row));
+ _current_row = left_row;
+ } else {
+ while (true) {
+ RETURN_IF_ERROR(parse_page_header());
+ if (_page_reader->is_header_v2() || !IN_COLLECTION) {
+ if (_page_reader->start_row() <= left_row && left_row <
_page_reader->end_row()) {
+ RETURN_IF_ERROR(load_page_data());
+ // this page contain this row.
+ RETURN_IF_ERROR(_skip_nested_rows_in_page(left_row -
_current_row));
+ _current_row = left_row;
+ break;
+ }
+
+ _current_row = _page_reader->end_row();
+ if (has_next_page()) [[likely]] {
+ RETURN_IF_ERROR(next_page());
+ } else {
+ return Status::InternalError("no match seek row {},
current row {}", left_row,
+ _current_row);
+ }
+ } else {
+ RETURN_IF_ERROR(load_page_data());
+ std::vector<level_t> rep_levels;
+ std::vector<level_t> def_levels;
+ bool cross_page = false;
+
+ size_t result_rows = 0;
+ RETURN_IF_ERROR(load_page_nested_rows(rep_levels, left_row -
_current_row,
+ &result_rows,
&cross_page));
+ RETURN_IF_ERROR(fill_def(def_levels));
+ RETURN_IF_ERROR(skip_nested_values(def_levels));
+ bool need_load_next_page = true;
+ while (cross_page) {
+ need_load_next_page = false;
+ rep_levels.clear();
+ def_levels.clear();
+ RETURN_IF_ERROR(load_cross_page_nested_row(rep_levels,
&cross_page));
+ RETURN_IF_ERROR(fill_def(def_levels));
+ RETURN_IF_ERROR(skip_nested_values(def_levels));
+ }
+ if (left_row == _current_row) {
+ break;
+ }
+ if (need_load_next_page) {
+ if (has_next_page()) [[likely]] {
+ RETURN_IF_ERROR(next_page());
+ } else {
+ return Status::InternalError("no match seek row {},
current row {}",
+ left_row, _current_row);
+ }
+ }
+ }
+ };
+ }
+
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_skip_nested_rows_in_page(size_t num_rows) {
+ if (num_rows == 0) {
+ return Status::OK();
+ }
+
+ std::vector<level_t> rep_levels;
+ std::vector<level_t> def_levels;
+
+ bool cross_page = false;
+ size_t result_rows = 0;
+ RETURN_IF_ERROR(load_page_nested_rows(rep_levels, num_rows, &result_rows,
&cross_page));
+ RETURN_IF_ERROR(fill_def(def_levels));
+ RETURN_IF_ERROR(skip_nested_values(def_levels));
+ DCHECK(cross_page == false);
+ if (num_rows != result_rows) [[unlikely]] {
+ return Status::InternalError("no match skip rows, expect {} vs. real
{}", num_rows,
+ result_rows);
+ }
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_nested_rows(
+ std::vector<level_t>& rep_levels, size_t max_rows, size_t*
result_rows, bool* cross_page) {
+ if (_state != DATA_LOADED) [[unlikely]] {
+ return Status::IOError("Should load page data first to load nested
rows");
+ }
+ *cross_page = false;
+ *result_rows = 0;
+ rep_levels.reserve(rep_levels.size() + _remaining_rep_nums);
Review Comment:
[P1] Bound nested level allocation by the requested rows. The page checks
only reject negative `num_values`; that untrusted count is copied to
`_remaining_rep_nums`, and this reserve allocates the entire remainder before
the loop can stop at `max_rows` or detect a short level stream. A tiny nested
page advertising `INT_MAX` values can therefore request roughly 4 GiB for this
int16 vector, with `fill_def` able to repeat the allocation. Please consume
levels in bounded spans tied to the requested batch/payload and add
tiny-payload huge-count V1/V2 corruption 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]
