This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 878a05d149d [fix](parquet) Handle empty filter pages and offset index
fallback (#68780)
878a05d149d is described below
commit 878a05d149df1baf04e6de4a9853a815fc464d04
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 9 15:45:27 2026 +0800
[fix](parquet) Handle empty filter pages and offset index fallback (#68780)
### What problem does this PR solve?
Parquet V2 scans can fail when an indexed page contributes no predicate
rows: its `NONE` execution kind conflicts with the preceding page.
Parsing a selected page can also discard a stale OffsetIndex after
callers have already calculated its read range, leading to incorrect
projected rows or EOF accounting.
The fix ignores zero-row fragments when merging predicate execution
kinds, reconciles selected-page ranges in
ordinary/fixed-width/dictionary reads, and preserves physical-value
counts and nested row positions across safe sequential fallback.
### Read cost and fallback behavior
- Keep indexed skips lazy after the initial data-page reconciliation.
Unselected pages do not require header reads, page-cache lookups,
payload reads, decompression, or repetition-level scans solely for
validation.
- Do not drain the remaining repetition levels of a partially consumed
indexed V1 data page when advancing. Continue to check selected page
headers, row boundaries, and naturally exhausted level streams.
- Track whether indexed navigation skipped an unread header or an
unfinished nested V1 row span. If a later accessed header invalidates
the index, reject that unsafe sequential fallback instead of reusing
uncertain row positions or an incomplete physical-value prefix. Safe
fallback still succeeds.
- Retain the metadata-only byte-coverage check before publishing
optional page indexes. It adds no file I/O; gaps and uncovered suffixes
still conservatively disable the index pair, including auxiliary gaps or
writer padding.
This preserves the normal index trust boundary: structurally accepted
indexes describe unread pages. The PR does not eagerly verify every
index entry against physical headers/levels and does not claim to detect
every forged row span or coordinated index/header inconsistency.
Comprehensive corrupt-index validation is outside this fix's scope.
Changes remain confined to `format_v2` and its tests. The legacy reader
under `be/src/format` is unchanged.
### Release note
Fix Parquet V2 scans with empty predicate page fragments and
selected-page OffsetIndex fallback while preserving lazy indexed page
skipping. Reject unsafe sequential fallback after an unverified indexed
prefix.
### Check List (For Author)
- [x] Unit tests
- [x] No new configuration or documentation requirement
- Reproduced both eager-read regressions before the implementation
change: a three-page sparse seek made 4 or 5 stream reads rather than 3,
and advancing a header-only/partially consumed V1 page loaded or drained
its levels.
- All **747** related BE ASAN tests passed with
`*Parquet*:FileScannerV2Test.*:TableReaderTest.*`.
- Sparse-read tests cover flat ordinary/fixed-width/dictionary paths,
nested value/levels-only paths, all 16 four-page V1/V2 layouts, and
cache hits/misses. Native 3-page and 64-page seeks parse exactly two
headers and perform exactly three stream reads, independent of the
skipped page count.
- Retained physical-row result comparisons across mixed page versions,
nulls, repeated values, dictionary encoding, batch sizes and all
nonempty subsets of a seven-row fixture. Fallback cases assert exact
success or the explicitly required corruption error.
- Complete Arrow-written file regressions retain coverage for empty
predicate fragments, selected-page fallback, incomplete index pairs and
pruning behavior.
- clang-format 16 on all eight affected C++ files and `git diff --check`
passed.
These are deterministic work-count and correctness checks, not
remote-storage throughput benchmarks.
---
.../parquet/reader/native/column_chunk_reader.cpp | 102 +-
.../parquet/reader/native/column_chunk_reader.h | 3 +
.../parquet/reader/native/column_reader.cpp | 51 +-
.../parquet/reader/native/column_reader.h | 1 +
.../format_v2/parquet/reader/native/page_reader.h | 6 +
.../parquet/reader/native_column_reader.cpp | 18 +-
be/test/format_v2/parquet/native_decoder_test.cpp | 1156 +++++++++++++++++++-
be/test/format_v2/parquet/parquet_scan_test.cpp | 226 ++++
8 files changed, 1491 insertions(+), 72 deletions(-)
diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
index d1677c66676..f524713f107 100644
--- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
@@ -240,7 +240,10 @@ bool validate_offset_index(const tparquet::OffsetIndex&
index, const ColumnChunk
// 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;
+ // A gap can hide an omitted data page even when ColumnIndex has the same
entry count.
+ // Decline such indexes before pruning can lose rows; auxiliary pages and
writer padding
+ // conservatively use sequential traversal. A dictionary prefix ends at
data_page_offset.
+ uint64_t previous_end = static_cast<uint64_t>(data_page_offset);
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 ||
@@ -249,14 +252,14 @@ bool validate_offset_index(const tparquet::OffsetIndex&
index, const ColumnChunk
}
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 ||
+ 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;
+ return previous_end == chunk_end;
}
namespace {
@@ -1067,11 +1070,11 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::load_dictionary_page(bool
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::ensure_first_data_page_parsed() {
- if (_first_data_page_parsed) {
+ if (_first_data_page_parsed && _page_reader->has_active_offset_index()) {
return Status::OK();
}
- // OffsetIndex row bounds are untrusted until page zero has been
reconciled and its declared
- // cardinality checked, so no indexed skip may observe them before this
one-time parse.
+ // OffsetIndex bounds need an initial reconciliation. If the index is
discarded, every later
+ // page needs its header parsed before callers use its sequential row
bounds.
return parse_page_header();
}
@@ -1096,19 +1099,17 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::parse_page_header() {
}
int32_t page_num_values = _page_reader->is_header_v2() ?
header->data_page_header_v2.num_values
:
header->data_page_header.num_values;
- if constexpr (IN_COLLECTION && OFFSET_INDEX) {
- if (!_page_reader->is_header_v2() &&
_page_reader->has_active_offset_index()) {
- // V1 nested pages do not declare their logical row count. An
OffsetIndex span cannot
- // be trusted until repetition levels are decoded, so keep the
sequential cursor path.
- _page_reader->discard_offset_index();
- _offset_index = nullptr;
- }
- }
+ // A later V1 page can still use a valid OffsetIndex in a mixed-version
chunk. Its first
+ // repetition level is checked when loaded, and its row count when the
levels are exhausted.
const bool active_offset_index = _page_reader->has_active_offset_index();
+ if (_skipped_unverified_indexed_page && !active_offset_index) {
+ // Lazy skips leave part of the physical prefix unknown. Once the
index is discarded,
+ // neither its absolute row coordinate nor the parsed-value EOF count
can be reused.
+ return Status::Corruption("Parquet OffsetIndex fallback after skipping
an unverified page");
+ }
if (page_num_values < 0 || page_num_values > _metadata.num_values ||
- (!active_offset_index &&
- static_cast<uint64_t>(page_num_values) >
- static_cast<uint64_t>(_metadata.num_values) -
_chunk_parsed_values)) {
+ static_cast<uint64_t>(page_num_values) >
+ static_cast<uint64_t>(_metadata.num_values) -
_chunk_parsed_values) {
// Page counts are untrusted and feed both level decoders and scratch
sizing. Bound each
// page by the column metadata before converting to unsigned counters.
return Status::Corruption("Parquet data page value count {} exceeds
column total {}",
@@ -1142,10 +1143,9 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::parse_page_header() {
_remaining_def_nums = page_num_values;
_remaining_num_values = page_num_values;
- // no offset will parse all header.
- if (!active_offset_index) {
- _chunk_parsed_values += _remaining_num_values;
- }
+ // Count each parsed data header exactly once, including the indexed
prefix. A safe fallback
+ // uses this physical value count for EOF; nested logical rows are not
interchangeable with it.
+ _chunk_parsed_values += _remaining_num_values;
_first_data_page_parsed = true;
_state = HEADER_PARSED;
return Status::OK();
@@ -1155,6 +1155,17 @@ template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() {
if constexpr (OFFSET_INDEX) {
RETURN_IF_ERROR(ensure_first_data_page_parsed());
+ if (_page_reader->has_active_offset_index()) {
+ // Preserve header/payload-free indexed skips. An unread header or
an unfinished V1
+ // row span prevents a later sequential fallback, but does not
require eager I/O.
+ if (_state == INITIALIZED) {
+ _skipped_unverified_indexed_page = true;
+ } else if constexpr (IN_COLLECTION) {
+ if (!_page_reader->is_header_v2() && _remaining_rep_nums != 0)
{
+ _skipped_unverified_indexed_page = true;
+ }
+ }
+ }
} else {
// Load dictionary state before advancing can jump past the physical
dictionary page.
RETURN_IF_ERROR(_ensure_dictionary_page_loaded());
@@ -1171,7 +1182,18 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::next_page() {
_decompress_release_pending = false;
_decompress_release_threshold = std::numeric_limits<size_t>::max();
}
+ const bool indexed = _page_reader->has_active_offset_index();
+ if (!indexed && (_page_reader->is_header_v2() || !IN_COLLECTION)) {
+ _current_row = _page_reader->end_row();
+ }
RETURN_IF_ERROR(_page_reader->next_page());
+ if (indexed) {
+ _current_row = _page_reader->start_row();
+ } else if constexpr (IN_COLLECTION) {
+ // V1 row counts come from repetition levels, so PageReader cannot
advance this coordinate
+ // from its header. Synchronize before a following V2 header derives
its sequential bounds.
+ _page_reader->set_sequential_row_start(_current_row);
+ }
_state = INITIALIZED;
return Status::OK();
}
@@ -1396,6 +1418,18 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::load_page_data() {
_remaining_rep_nums));
}
}
+ if constexpr (IN_COLLECTION && OFFSET_INDEX) {
+ if (_max_rep_level > 0 && _remaining_rep_nums > 0 &&
+ _page_reader->has_active_offset_index()) {
+ // Indexed pages must start at a row boundary, including V1 pages
after indexed skips.
+ // Reject continuations before a partial selection can expose
values from the wrong row.
+ if (_rep_level_decoder.get_next() != 0) {
+ return Status::Corruption(
+ "Indexed Parquet nested page must start at a row
boundary");
+ }
+ _rep_level_decoder.rewind_one();
+ }
+ }
if (_max_def_level > 0) {
SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
if (header->__isset.data_page_header_v2) {
@@ -2262,29 +2296,29 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::seek_to_nested_row(size_t
if constexpr (IN_COLLECTION && OFFSET_INDEX) {
RETURN_IF_ERROR(ensure_first_data_page_parsed());
}
- if constexpr (OFFSET_INDEX) {
- if (_page_reader->has_active_offset_index()) {
- while (true) {
+ while (true) {
+ if constexpr (OFFSET_INDEX) {
+ if (_page_reader->has_active_offset_index()) {
if (_page_reader->start_row() <= left_row && left_row <
_page_reader->end_row()) {
- break;
+ RETURN_IF_ERROR(parse_page_header());
+ if (_page_reader->has_active_offset_index()) {
+ RETURN_IF_ERROR(load_page_data());
+ RETURN_IF_ERROR(_skip_nested_rows_in_page(left_row -
_current_row));
+ _current_row = left_row;
+ return Status::OK();
+ }
+ // Parsing can discard the index and change this page's
bounds. Re-enter the
+ // sequential decision below instead of seeking within the
stale indexed span.
} else if (has_next_page()) {
RETURN_IF_ERROR(next_page());
- _current_row = _page_reader->start_row();
+ continue;
} 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;
- return Status::OK();
}
- }
- 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()) {
diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
b/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
index a50b7e4bc08..266bcca9b6d 100644
--- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
+++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
@@ -434,6 +434,8 @@ private:
LevelDecoder _rep_level_decoder;
LevelDecoder _def_level_decoder;
+ // Parsed headers contribute exactly once. This is a complete physical
prefix only while
+ // no indexed page has been skipped unverified; otherwise sequential
fallback is rejected.
size_t _chunk_parsed_values = 0;
// this page remaining rep/def nums
// if max_rep_level = 0 / max_def_level = 0, this value retail hava value.
@@ -454,6 +456,7 @@ private:
Slice _v2_def_levels;
bool _dict_checked = false;
bool _first_data_page_parsed = false;
+ bool _skipped_unverified_indexed_page = false;
bool _has_dict = false;
bool _nested_row_started = false;
Decoder* _page_decoder = nullptr;
diff --git a/be/src/format_v2/parquet/reader/native/column_reader.cpp
b/be/src/format_v2/parquet/reader/native/column_reader.cpp
index 314a2869041..1744966a061 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp
@@ -729,6 +729,27 @@ void ColumnReader::_generate_read_ranges(RowRange
page_row_range, RowRanges* res
RowRanges::ranges_intersection(*result_ranges, _row_ranges, result_ranges);
}
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::_get_page_read_ranges(
+ int64_t* right_row, RowRanges* read_ranges) {
+ RETURN_IF_ERROR(_chunk_reader->ensure_first_data_page_parsed());
+ *right_row = _chunk_reader->page_end_row();
+ _generate_read_ranges(RowRange {_current_row_index, *right_row},
read_ranges);
+ if (read_ranges->count() == 0) {
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(_chunk_reader->parse_page_header());
+ const auto parsed_end = _chunk_reader->page_end_row();
+ if (*right_row != parsed_end) {
+ // Parsing a selected later page can discard its OffsetIndex. Both the
selection and the
+ // page-advance decision must use the reconciled bound, or unread
values shift into the next page.
+ *right_row = parsed_end;
+ read_ranges->clear();
+ _generate_read_ranges(RowRange {_current_row_index, *right_row},
read_ranges);
+ }
+ return Status::OK();
+}
+
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::init(
io::FileReaderSPtr file, NativeFieldSchema* field, size_t
max_buf_size, RuntimeState* state,
@@ -1331,18 +1352,11 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_fixed_width_filter(
}
int64_t right_row = 0;
- if constexpr (OFFSET_INDEX == false) {
- RETURN_IF_ERROR(_chunk_reader->parse_page_header());
- } else {
- RETURN_IF_ERROR(_chunk_reader->ensure_first_data_page_parsed());
- }
- right_row = _chunk_reader->page_end_row();
RowRanges read_ranges;
- _generate_read_ranges(RowRange {_current_row_index, right_row},
&read_ranges);
+ RETURN_IF_ERROR(_get_page_read_ranges(&right_row, &read_ranges));
if (read_ranges.count() == 0) {
_current_row_index = right_row;
} else {
- RETURN_IF_ERROR(_chunk_reader->parse_page_header());
RETURN_IF_ERROR(_chunk_reader->load_page_data_idempotent());
if (!_chunk_reader->can_filter_fixed_width_values(conjuncts,
column_id, _serde.get(),
&_decode_context)) {
@@ -1468,18 +1482,11 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_dictionary_filter(
}
int64_t right_row = 0;
- if constexpr (OFFSET_INDEX == false) {
- RETURN_IF_ERROR(_chunk_reader->parse_page_header());
- right_row = _chunk_reader->page_end_row();
- } else {
- right_row = _chunk_reader->page_end_row();
- }
RowRanges read_ranges;
- _generate_read_ranges(RowRange {_current_row_index, right_row},
&read_ranges);
+ RETURN_IF_ERROR(_get_page_read_ranges(&right_row, &read_ranges));
if (read_ranges.count() == 0) {
_current_row_index = right_row;
} else {
- RETURN_IF_ERROR(_chunk_reader->parse_page_header());
RETURN_IF_ERROR(_chunk_reader->load_page_data_idempotent());
if (_chunk_reader->current_encoding() !=
tparquet::Encoding::RLE_DICTIONARY) {
return Status::OK();
@@ -1764,17 +1771,10 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_data(
}
int64_t right_row = 0;
- if constexpr (OFFSET_INDEX == false) {
- RETURN_IF_ERROR(_chunk_reader->parse_page_header());
- } else {
- RETURN_IF_ERROR(_chunk_reader->ensure_first_data_page_parsed());
- }
- right_row = _chunk_reader->page_end_row();
+ RowRanges read_ranges;
+ RETURN_IF_ERROR(_get_page_read_ranges(&right_row, &read_ranges));
do {
- // generate the row ranges that should be read
- RowRanges read_ranges;
- _generate_read_ranges(RowRange {_current_row_index, right_row},
&read_ranges);
if (read_ranges.count() == 0) {
// skip the whole page
_current_row_index = right_row;
@@ -1800,7 +1800,6 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_data(
}
}
// load page data to decode or skip values
- RETURN_IF_ERROR(_chunk_reader->parse_page_header());
RETURN_IF_ERROR(_chunk_reader->load_page_data_idempotent());
size_t has_read = 0;
for (size_t idx = 0; idx < read_ranges.range_size(); idx++) {
diff --git a/be/src/format_v2/parquet/reader/native/column_reader.h
b/be/src/format_v2/parquet/reader/native/column_reader.h
index b43822cd165..0b1d578d611 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.h
+++ b/be/src/format_v2/parquet/reader/native/column_reader.h
@@ -444,6 +444,7 @@ private:
size_t retained_batch_scratch_bytes() const;
size_t active_batch_scratch_bytes() const;
+ Status _get_page_read_ranges(int64_t* right_row, RowRanges* read_ranges);
Status _skip_values(size_t num_values);
Status _read_values(size_t num_values, ColumnPtr& doris_column, const
DataTypePtr& type,
FilterMap& filter_map, bool is_dict_filter);
diff --git a/be/src/format_v2/parquet/reader/native/page_reader.h
b/be/src/format_v2/parquet/reader/native/page_reader.h
index da2c066891b..bf009f472f1 100644
--- a/be/src/format_v2/parquet/reader/native/page_reader.h
+++ b/be/src/format_v2/parquet/reader/native/page_reader.h
@@ -229,6 +229,12 @@ public:
size_t end_row() const { return _end_row; }
+ void set_sequential_row_start(size_t row) {
+ DCHECK(!has_active_offset_index());
+ _start_row = row;
+ _end_row = row;
+ }
+
bool has_active_offset_index() const {
if constexpr (OFFSET_INDEX) {
return _offset_index != nullptr;
diff --git a/be/src/format_v2/parquet/reader/native_column_reader.cpp
b/be/src/format_v2/parquet/reader/native_column_reader.cpp
index 666fdbba327..179a09ad345 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp
@@ -542,6 +542,16 @@ Status NativeColumnReader::read_with_fixed_width_filter(
row_filter->clear();
return Status::OK();
}
+ // A skipped page can advance the native cursor without evaluating a
predicate. Its NONE
+ // execution kind must not replace or conflict with the preceding data
page's kind.
+ if (loop_rows == 0) {
+ if (!eof && ++consecutive_empty_calls > _row_group_rows + 1) {
+ return Status::Corruption(
+ "Native parquet fixed-width predicate made no progress
for column {}",
+ _name);
+ }
+ continue;
+ }
if (*execution_kind == DirectPredicateExecutionKind::NONE) {
*execution_kind = loop_kind;
} else if (loop_kind !=
DirectPredicateExecutionKind::DEFINITION_LEVEL) {
@@ -550,14 +560,6 @@ Status NativeColumnReader::read_with_fixed_width_filter(
*execution_kind = loop_kind;
}
row_filter->insert(row_filter->end(), loop_filter.begin(),
loop_filter.end());
- if (loop_rows == 0 && !eof) {
- if (++consecutive_empty_calls > _row_group_rows + 1) {
- return Status::Corruption(
- "Native parquet fixed-width predicate made no progress
for column {}",
- _name);
- }
- continue;
- }
consecutive_empty_calls = 0;
*rows_read += static_cast<int64_t>(loop_rows);
}
diff --git a/be/test/format_v2/parquet/native_decoder_test.cpp
b/be/test/format_v2/parquet/native_decoder_test.cpp
index af4a17d1b8a..7d8ca6a7176 100644
--- a/be/test/format_v2/parquet/native_decoder_test.cpp
+++ b/be/test/format_v2/parquet/native_decoder_test.cpp
@@ -43,6 +43,7 @@
#include "exprs/vectorized_fn_call.h"
#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
+#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/reader/native/byte_array_dict_decoder.h"
#include "format_v2/parquet/reader/native/column_reader.h"
#include "format_v2/parquet/reader/native/decoder.h"
@@ -51,6 +52,8 @@
#include "format_v2/parquet/reader/native/level_decoder.h"
#include "format_v2/parquet/reader/native/level_reader.h"
#include "format_v2/parquet/reader/native/page_reader.h"
+#include "format_v2/parquet/reader/native_column_reader.h"
+#include "format_v2/parquet/reader/variant_column_reader.h"
#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
#include "util/block_compression.h"
@@ -1316,7 +1319,7 @@ TEST(ParquetV2NativeDecoderTest,
NullableSparsePlainDateTimeSelectionBatchesPhys
ColumnChunkReaderStatistics statistics;
const ParquetDecodeContext context {.physical_type =
ParquetPhysicalType::INT64,
.logical_type =
ParquetLogicalType::TIMESTAMP,
- .time_unit = ParquetTimeUnit::MICROS};
+ .time_unit =
::doris::ParquetTimeUnit::MICROS};
ASSERT_TRUE(materialize_selected_plain_int64(physical_micros,
LOGICAL_VALUES, null_runs, filter,
type, &column, &null_map,
&statistics, context)
.ok());
@@ -1342,10 +1345,10 @@ TEST(ParquetV2NativeDecoderTest,
NegativeNanosFloorAcrossPlainAndDictionaryTimes
"1970-01-01 00:00:00.000000", "1970-01-01 00:00:00.000001"};
const ParquetDecodeContext local_context {.physical_type =
ParquetPhysicalType::INT64,
.logical_type =
ParquetLogicalType::TIMESTAMP,
- .time_unit =
ParquetTimeUnit::NANOS};
+ .time_unit =
::doris::ParquetTimeUnit::NANOS};
const ParquetDecodeContext utc_context {.physical_type =
ParquetPhysicalType::INT64,
.logical_type =
ParquetLogicalType::TIMESTAMP,
- .time_unit =
ParquetTimeUnit::NANOS,
+ .time_unit =
::doris::ParquetTimeUnit::NANOS,
.timestamp_is_adjusted_to_utc =
true};
for (const bool dictionary : {false, true}) {
@@ -3454,6 +3457,81 @@ TEST(ParquetV2NativeDecoderTest,
LazyFixedWidthFilterUsesReconciledFirstPageRang
EXPECT_EQ(row_filter, (IColumn::Filter {1, 1}));
}
+// Exercise the adapter and native reader together: an empty fragment must not
change the
+// predicate domain, and an index fallback must still refresh every subsequent
page's row bounds.
+void check_fixed_width_filter_across_pages(bool shifted_index) {
+ const std::vector<std::vector<uint8_t>> pages
{serialize_plain_int32_page({10, 11}),
+
serialize_plain_int32_page({20, 21}),
+
serialize_plain_int32_page({30, 31})};
+ std::vector<uint8_t> bytes;
+ tparquet::OffsetIndex offset_index;
+ for (size_t page = 0; page < pages.size(); ++page) {
+ tparquet::PageLocation location;
+ location.__set_offset(bytes.size() + (shifted_index && page > 0 ? 1 :
0));
+ location.__set_compressed_page_size(pages[page].size() +
+ (shifted_index && page == 0 ? 1 :
0));
+ location.__set_first_row_index(page * 2);
+ offset_index.page_locations.push_back(location);
+ bytes.insert(bytes.end(), pages[page].begin(), pages[page].end());
+ }
+ bytes.push_back(0);
+ tparquet::ColumnChunk chunk;
+ chunk.meta_data.__set_type(tparquet::Type::INT32);
+ chunk.meta_data.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ chunk.meta_data.__set_num_values(6);
+ chunk.meta_data.__set_total_compressed_size(bytes.size());
+ chunk.meta_data.__set_data_page_offset(0);
+ chunk.meta_data.__set_encodings({tparquet::Encoding::PLAIN});
+ NativeFieldSchema field;
+ field.physical_type = tparquet::Type::INT32;
+ field.data_type = std::make_shared<DataTypeInt32>();
+ field.parquet_schema.__set_type(tparquet::Type::INT32);
+
field.parquet_schema.__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ auto ranges = ::doris::RowRanges::create_single(0, shifted_index ? 6 : 2);
+ if (!shifted_index) {
+ ranges.add({4, 6});
+ }
+ auto file = std::make_shared<NativeDecoderMemoryFileReader>(bytes);
+ auto native = std::make_unique<ScalarColumnReader<false, true>>(ranges, 6,
chunk, &offset_index,
+ nullptr,
nullptr);
+ ASSERT_TRUE(native->init(file, &field, bytes.size(), nullptr, "",
ParquetReaderCompat {}, true)
+ .ok());
+ ParquetColumnSchema schema;
+ schema.name = "value";
+ schema.type = field.data_type;
+ NativeColumnReader reader(schema, schema.type, schema.type, nullptr, {});
+ reader._native_reader = std::move(native);
+ reader._row_group_rows = 6;
+ auto projected = ColumnInt32::create();
+ IColumn::Filter row_filter;
+ int64_t rows = 0;
+ bool used_filter = false;
+ DirectPredicateExecutionKind kind = DirectPredicateExecutionKind::NONE;
+ const auto status = reader.read_with_fixed_width_filter(
+ ranges.count(), nullptr, false,
+ {create_int32_raw_comparison(0, "gt", TExprOpcode::GT, 10)}, 0,
projected.get(),
+ &row_filter, &rows, &used_filter, &kind);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_TRUE(used_filter);
+ EXPECT_EQ(rows, ranges.count());
+ EXPECT_EQ(kind, DirectPredicateExecutionKind::CONVERTED_FIXED);
+ if (shifted_index) {
+ EXPECT_EQ(row_filter, (IColumn::Filter {0, 1, 1, 1, 1, 1}));
+ EXPECT_EQ(projected->get_data(), (ColumnInt32::Container {11, 20, 21,
30, 31}));
+ } else {
+ EXPECT_EQ(row_filter, (IColumn::Filter {0, 1, 1, 1}));
+ EXPECT_EQ(projected->get_data(), (ColumnInt32::Container {11, 30,
31}));
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, FixedWidthFilterSkipsEmptyPageFragments) {
+ check_fixed_width_filter_across_pages(false);
+}
+
+TEST(ParquetV2NativeDecoderTest,
FixedWidthFilterContinuesAfterOffsetIndexFallback) {
+ check_fixed_width_filter_across_pages(true);
+}
+
TEST(ParquetV2NativeDecoderTest,
LazyFlatIndexedFallbackUsesReconciledFirstPageRange) {
const auto first_page = serialize_plain_int32_page({10, 11});
const auto second_page = serialize_plain_int32_page({20});
@@ -3484,7 +3562,7 @@ TEST(ParquetV2NativeDecoderTest,
LazyFlatIndexedFallbackUsesReconciledFirstPageR
field.parquet_schema.__set_type(tparquet::Type::INT32);
field.parquet_schema.__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
auto file = std::make_shared<NativeDecoderMemoryFileReader>(bytes);
- const auto row_ranges = ::doris::RowRanges::create_single(0, 2);
+ const auto row_ranges = ::doris::RowRanges::create_single(0, 3);
ScalarColumnReader<false, true> reader(row_ranges, 3, chunk,
&offset_index, nullptr, nullptr);
ASSERT_TRUE(reader.init(file, &field, bytes.size(), nullptr, "",
ParquetReaderCompat {}, true)
.ok());
@@ -3500,6 +3578,12 @@ TEST(ParquetV2NativeDecoderTest,
LazyFlatIndexedFallbackUsesReconciledFirstPageR
ASSERT_EQ(rows, 2);
EXPECT_EQ(assert_cast<const ColumnInt32&>(*values).get_data(),
(ColumnInt32::Container {10, 11}));
+ values = ColumnInt32::create();
+ ASSERT_TRUE(
+ reader.read_column_data(values, field.data_type, nullptr, filter,
1, &rows, &eof, false)
+ .ok());
+ ASSERT_EQ(rows, 1);
+ EXPECT_EQ(assert_cast<const ColumnInt32&>(*values).get_data(),
(ColumnInt32::Container {20}));
}
TEST(ParquetV2NativeDecoderTest, LazyNestedV2SeekValidatesFirstPageRowRange) {
@@ -3559,6 +3643,1068 @@ TEST(ParquetV2NativeDecoderTest,
LazyNestedV2SeekValidatesFirstPageRowRange) {
EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
}
+TEST(ParquetV2NativeDecoderTest,
SelectedLaterPageReconcilesRangesAfterIndexFallback) {
+ for (const bool shrinking : {false, true}) {
+ for (const int mode : {0, 1, 2}) {
+ for (const size_t batch_size : {1, 2, 3}) {
+ for (const bool cache_hit : {false, true}) {
+ SCOPED_TRACE(testing::Message()
+ << "mode=" << mode << ", batch=" << batch_size
+ << ", cache=" << cache_hit << ", shrinking="
<< shrinking);
+ const std::vector<std::vector<int32_t>> values {
+ {10, 11}, {20, 21, 22}, {30, 31}};
+ std::vector<uint8_t> bytes(1, 0);
+ auto dictionary = ColumnInt32::create();
+ dictionary->get_data() = {10, 11, 20, 21, 22, 30, 31};
+ if (mode == 2) {
+ std::vector<uint8_t> payload(dictionary->size() *
sizeof(int32_t));
+ memcpy(payload.data(), dictionary->get_data().data(),
payload.size());
+ tparquet::PageHeader header;
+ header.__set_type(tparquet::PageType::DICTIONARY_PAGE);
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ tparquet::DictionaryPageHeader data;
+ data.__set_num_values(dictionary->size());
+ data.__set_encoding(tparquet::Encoding::PLAIN);
+ header.__set_dictionary_page_header(data);
+ auto page = serialize_page(header, payload);
+ bytes.insert(bytes.end(), page.begin(), page.end());
+ }
+ const size_t data_offset = bytes.size();
+ tparquet::OffsetIndex index;
+ std::vector<std::vector<uint8_t>> pages;
+ uint32_t next_id = 0;
+ for (size_t i = 0; i < values.size(); ++i) {
+ if (mode == 2) {
+ faststring ids;
+ RleEncoder<uint32_t> encoder(&ids, 3);
+ for (size_t j = 0; j < values[i].size(); ++j) {
+ encoder.Put(next_id++);
+ }
+ for (size_t j = values[i].size(); j < 8; ++j) {
+ encoder.Put(0);
+ }
+ encoder.Flush();
+ std::vector<uint8_t> payload(ids.size() + 1, 3);
+ memcpy(payload.data() + 1, ids.data(), ids.size());
+ tparquet::PageHeader header;
+ header.__set_type(tparquet::PageType::DATA_PAGE);
+ header.__set_compressed_page_size(payload.size());
+
header.__set_uncompressed_page_size(payload.size());
+ tparquet::DataPageHeader data;
+ data.__set_num_values(values[i].size());
+
data.__set_encoding(tparquet::Encoding::RLE_DICTIONARY);
+
data.__set_definition_level_encoding(tparquet::Encoding::RLE);
+
data.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+ header.__set_data_page_header(data);
+ pages.push_back(serialize_page(header, payload));
+ } else {
+
pages.push_back(serialize_plain_int32_page(values[i]));
+ }
+ tparquet::PageLocation location;
+ location.__set_offset(bytes.size());
+
location.__set_compressed_page_size(pages.back().size() - (i == 1 ? 1 : 0));
+ // Reconciliation can expand a selected range or
remove it entirely.
+ // Neither case may advance using the stale indexed
page boundary.
+ location.__set_first_row_index(shrinking && i == 2 ? 6
: i * 2);
+ index.page_locations.push_back(location);
+ bytes.insert(bytes.end(), pages.back().begin(),
pages.back().end());
+ }
+ const std::string cache_key =
+ fmt::format("selected-late-{}-{}-{}", mode,
batch_size, shrinking);
+ if (cache_hit) {
+ auto* page = new DataPage(pages[1].size(), true,
segment_v2::DATA_PAGE);
+ memcpy(page->data(), pages[1].data(), pages[1].size());
+ page->reset_size(pages[1].size());
+ PageCacheHandle handle;
+ StoragePageCache::instance()->insert(
+ StoragePageCache::CacheKey(cache_key,
bytes.size(),
+
index.page_locations[1].offset),
+ page, &handle, segment_v2::DATA_PAGE);
+ }
+ tparquet::ColumnChunk chunk;
+ chunk.meta_data.__set_type(tparquet::Type::INT32);
+
chunk.meta_data.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ chunk.meta_data.__set_num_values(7);
+ chunk.meta_data.__set_total_compressed_size(bytes.size() -
1);
+ chunk.meta_data.__set_data_page_offset(data_offset);
+
chunk.meta_data.__set_encodings({tparquet::Encoding::PLAIN});
+ if (mode == 2) {
+ chunk.meta_data.__set_dictionary_page_offset(1);
+ chunk.meta_data.__set_encodings(
+ {tparquet::Encoding::PLAIN,
tparquet::Encoding::RLE_DICTIONARY});
+ }
+ NativeFieldSchema field;
+ field.physical_type = tparquet::Type::INT32;
+ field.data_type = std::make_shared<DataTypeInt32>();
+ field.parquet_schema.__set_type(tparquet::Type::INT32);
+ field.parquet_schema.__set_repetition_type(
+ tparquet::FieldRepetitionType::REQUIRED);
+ const auto ranges =
+ ::doris::RowRanges::create_single(shrinking ? 5 :
2, shrinking ? 6 : 5);
+ const size_t expected_rows = shrinking ? 1 : 3;
+ auto file =
std::make_shared<NativeDecoderMemoryFileReader>(bytes);
+ ScalarColumnReader<false, true> reader(ranges, 7, chunk,
&index, nullptr,
+ nullptr);
+ ASSERT_TRUE(reader.init(file, &field, bytes.size(),
nullptr,
+ cache_hit ? cache_key : "",
ParquetReaderCompat {},
+ true)
+ .ok());
+ FilterMap filter;
+ ASSERT_TRUE(filter.init(nullptr, expected_rows,
false).ok());
+ ColumnPtr ordinary = ColumnInt32::create();
+ auto projected = ColumnInt32::create();
+ size_t consumed = 0;
+ bool eof = false;
+ for (size_t call = 0; consumed < expected_rows && !eof;
++call) {
+ ASSERT_LT(call, 8);
+ size_t rows = 0;
+ Status status;
+ IColumn::Filter row_filter;
+ bool used = false;
+ const size_t count = std::min(batch_size,
expected_rows - consumed);
+ if (mode == 0) {
+ status = reader.read_column_data(ordinary,
field.data_type, nullptr,
+ filter, count,
&rows, &eof, false);
+ } else if (mode == 1) {
+ DirectPredicateExecutionKind kind =
DirectPredicateExecutionKind::NONE;
+ status = reader.read_fixed_width_filter(
+ {create_int32_raw_comparison(0, "gt",
TExprOpcode::GT, 0)}, 0,
+ filter, count, projected.get(),
&row_filter, &rows, &eof, &used,
+ &kind);
+ } else {
+ size_t survivors = 0;
+ bool direct = false;
+ status = reader.read_dictionary_filter(
+ IColumn::Filter(7, 1), filter, count,
dictionary.get(),
+ projected.get(), nullptr, &row_filter,
&survivors, &rows, &eof,
+ &direct, &used);
+ EXPECT_EQ(survivors, rows);
+ }
+ ASSERT_TRUE(status.ok()) << status;
+ if (mode != 0) {
+ EXPECT_TRUE(used);
+ EXPECT_EQ(row_filter, IColumn::Filter(rows, 1));
+ }
+ consumed += rows;
+ }
+ EXPECT_EQ(consumed, expected_rows);
+ const auto& actual =
+ mode == 0 ? assert_cast<const
ColumnInt32&>(*ordinary) : *projected;
+ EXPECT_EQ(actual.get_data(), (shrinking ?
ColumnInt32::Container {30}
+ :
ColumnInt32::Container {20, 21, 22}));
+
EXPECT_EQ(reader.column_statistics().page_cache_hit_counter > 0, cache_hit);
+ }
+ }
+ }
+ }
+}
+
+// Serialize independent page versions and row shapes so navigation tests do
not inherit a writer's
+// fixed page layout. Expected values are derived from the physical row
sequence, never the index.
+struct NestedNavigationFile {
+ std::vector<uint8_t> bytes {0};
+ tparquet::ColumnChunk chunk;
+ tparquet::OffsetIndex index;
+ NativeFieldSchema field;
+ size_t rows = 0;
+ std::vector<bool> page_versions;
+ std::vector<std::vector<int32_t>> row_values;
+ std::vector<std::vector<uint8_t>> pages;
+
+ NestedNavigationFile(const std::vector<bool>& v2, const
std::vector<size_t>& page_rows,
+ bool repeated = false, bool dictionary = false, bool
nullable = false,
+ bool nested = true)
+ : page_versions(v2) {
+ field.physical_type = tparquet::Type::INT32;
+ field.data_type = std::make_shared<DataTypeInt32>();
+ field.repetition_level = nested ? 1 : 0;
+ field.definition_level = nullable ? 1 : 0;
+ if (nullable) {
+ field.data_type = make_nullable(field.data_type);
+ }
+ field.parquet_schema.__set_type(tparquet::Type::INT32);
+ field.parquet_schema.__set_repetition_type(
+ nullable ? tparquet::FieldRepetitionType::OPTIONAL
+ : tparquet::FieldRepetitionType::REQUIRED);
+ const size_t row_count = std::accumulate(page_rows.begin(),
page_rows.end(), size_t {0});
+ const size_t value_count = row_count * (repeated ? 2 : 1);
+ if (dictionary) {
+ std::vector<int32_t> dictionary_values(value_count);
+ std::iota(dictionary_values.begin(), dictionary_values.end(), 100);
+ std::vector<uint8_t> payload(value_count * sizeof(int32_t));
+ memcpy(payload.data(), dictionary_values.data(), payload.size());
+ tparquet::PageHeader header;
+ header.__set_type(tparquet::PageType::DICTIONARY_PAGE);
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ tparquet::DictionaryPageHeader data;
+ data.__set_num_values(value_count);
+ data.__set_encoding(tparquet::Encoding::PLAIN);
+ header.__set_dictionary_page_header(data);
+ const auto page = serialize_page(header, payload);
+ bytes.insert(bytes.end(), page.begin(), page.end());
+ chunk.meta_data.__set_dictionary_page_offset(1);
+ }
+ chunk.meta_data.__set_data_page_offset(bytes.size());
+ size_t value_index = 0;
+ for (size_t i = 0; i < v2.size(); ++i) {
+ std::vector<int32_t> values;
+ std::vector<uint32_t> levels;
+ std::vector<uint32_t> definitions;
+ for (size_t row = 0; row < page_rows[i]; ++row) {
+ row_values.emplace_back();
+ for (size_t element = 0; element < (repeated ? 2 : 1);
++element) {
+ levels.push_back(element == 0 ? 0 : 1);
+ const bool is_null = nullable && value_index % 3 == 1;
+ const int32_t value = 100 + value_index++;
+ definitions.push_back(is_null ? 0 : 1);
+ if (!is_null) {
+ values.push_back(value);
+ }
+ row_values.back().push_back(is_null ?
std::numeric_limits<int32_t>::min()
+ : value);
+ }
+ }
+ faststring encoded_levels;
+ std::vector<uint8_t> payload;
+ if (nested) {
+ RleEncoder<uint32_t> level_encoder(&encoded_levels, 1);
+ for (const auto level : levels) {
+ level_encoder.Put(level);
+ }
+ for (size_t padding = levels.size(); padding % 8 != 0;
++padding) {
+ level_encoder.Put(0);
+ }
+ level_encoder.Flush();
+ if (!v2[i]) {
+ const uint32_t size = encoded_levels.size();
+ const auto* raw = reinterpret_cast<const uint8_t*>(&size);
+ payload.insert(payload.end(), raw, raw + sizeof(size));
+ }
+ payload.insert(payload.end(), encoded_levels.data(),
+ encoded_levels.data() + encoded_levels.size());
+ }
+ faststring encoded_definitions;
+ if (nullable) {
+ RleEncoder<uint32_t> def_encoder(&encoded_definitions, 1);
+ for (const auto level : definitions) {
+ def_encoder.Put(level);
+ }
+ for (size_t padding = definitions.size(); padding % 8 != 0;
++padding) {
+ def_encoder.Put(0);
+ }
+ def_encoder.Flush();
+ if (!v2[i]) {
+ const uint32_t size = encoded_definitions.size();
+ const auto* raw = reinterpret_cast<const uint8_t*>(&size);
+ payload.insert(payload.end(), raw, raw + sizeof(size));
+ }
+ payload.insert(payload.end(), encoded_definitions.data(),
+ encoded_definitions.data() +
encoded_definitions.size());
+ }
+ if (dictionary) {
+ faststring ids;
+ RleEncoder<uint32_t> encoder(&ids, 8);
+ for (const auto value : values) {
+ encoder.Put(value - 100);
+ }
+ for (size_t padding = values.size(); padding % 8 != 0;
++padding) {
+ encoder.Put(0);
+ }
+ encoder.Flush();
+ payload.push_back(8);
+ payload.insert(payload.end(), ids.data(), ids.data() +
ids.size());
+ } else {
+ const auto* raw = reinterpret_cast<const
uint8_t*>(values.data());
+ payload.insert(payload.end(), raw, raw + values.size() *
sizeof(int32_t));
+ }
+ const auto encoding =
+ dictionary ? tparquet::Encoding::RLE_DICTIONARY :
tparquet::Encoding::PLAIN;
+ tparquet::PageHeader header;
+ header.__set_type(v2[i] ? tparquet::PageType::DATA_PAGE_V2
+ : tparquet::PageType::DATA_PAGE);
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ if (v2[i]) {
+ tparquet::DataPageHeaderV2 data;
+ data.__set_num_values(levels.size());
+ data.__set_num_rows(page_rows[i]);
+ data.__set_num_nulls(levels.size() - values.size());
+ data.__set_encoding(encoding);
+
data.__set_repetition_levels_byte_length(encoded_levels.size());
+
data.__set_definition_levels_byte_length(encoded_definitions.size());
+ data.__set_is_compressed(false);
+ header.__set_data_page_header_v2(data);
+ } else {
+ tparquet::DataPageHeader data;
+ data.__set_num_values(levels.size());
+ data.__set_encoding(encoding);
+ data.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+ data.__set_definition_level_encoding(tparquet::Encoding::RLE);
+ header.__set_data_page_header(data);
+ }
+ pages.push_back(serialize_page(header, payload));
+ tparquet::PageLocation location;
+ location.__set_offset(bytes.size());
+ location.__set_compressed_page_size(pages.back().size());
+ location.__set_first_row_index(rows);
+ index.page_locations.push_back(location);
+ rows += page_rows[i];
+ bytes.insert(bytes.end(), pages.back().begin(),
pages.back().end());
+ }
+ chunk.meta_data.__set_type(tparquet::Type::INT32);
+ chunk.meta_data.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ chunk.meta_data.__set_num_values(value_count);
+ chunk.meta_data.__set_total_compressed_size(bytes.size() - 1);
+ chunk.meta_data.__set_encodings(
+ dictionary
+ ? std::vector<tparquet::Encoding::type>
{tparquet::Encoding::PLAIN,
+
tparquet::Encoding::RLE_DICTIONARY}
+ : std::vector<tparquet::Encoding::type>
{tparquet::Encoding::PLAIN});
+ }
+
+ Status read(const RowRanges& ranges, size_t batch_size, bool indexed, bool
cache,
+ std::vector<int32_t>* actual, bool levels_only = false, int
flat_mode = 0,
+ int64_t* parsed_headers = nullptr) {
+ static size_t cache_id = 0;
+ const std::string cache_key = cache ?
fmt::format("nested-navigation-{}", ++cache_id) : "";
+ if (cache) {
+ for (size_t i = 0; i < pages.size(); ++i) {
+ auto* page = new DataPage(pages[i].size(), true,
segment_v2::DATA_PAGE);
+ memcpy(page->data(), pages[i].data(), pages[i].size());
+ page->reset_size(pages[i].size());
+ PageCacheHandle handle;
+ StoragePageCache::instance()->insert(
+ StoragePageCache::CacheKey(cache_key, bytes.size(),
+
index.page_locations[i].offset),
+ page, &handle, segment_v2::DATA_PAGE);
+ }
+ }
+ auto file = std::make_shared<NativeDecoderMemoryFileReader>(bytes);
+ auto read_with = [&]<bool NESTED, bool INDEXED>() -> Status {
+ ScalarColumnReader<NESTED, INDEXED> reader(
+ ranges, rows, chunk, indexed ? &index : nullptr, nullptr,
nullptr);
+ RETURN_IF_ERROR(reader.init(file, &field, bytes.size(), nullptr,
cache_key,
+ ParquetReaderCompat {}, true));
+ if constexpr (NESTED) {
+ reader.set_column_in_nested();
+ }
+ FilterMap filter;
+ RETURN_IF_ERROR(filter.init(nullptr, rows, false));
+ bool eof = false;
+ size_t calls = 0;
+ while (!eof) {
+ if (++calls > rows + pages.size() + 1) {
+ return Status::InternalError("Nested navigation did not
reach EOF");
+ }
+ ColumnPtr output = field.data_type->create_column();
+ size_t read_rows = 0;
+ if (levels_only) {
+ RETURN_IF_ERROR(
+ reader.read_column_levels(filter, batch_size,
&read_rows, &eof));
+ actual->insert(actual->end(), read_rows, 0);
+ } else if (flat_mode != 0) {
+ auto projected = ColumnInt32::create();
+ IColumn::Filter row_filter;
+ bool used = false;
+ if (flat_mode == 1) {
+ DirectPredicateExecutionKind kind =
DirectPredicateExecutionKind::NONE;
+ RETURN_IF_ERROR(reader.read_fixed_width_filter(
+ {create_int32_raw_comparison(0, "gt",
TExprOpcode::GT, 0)}, 0,
+ filter, batch_size, projected.get(),
&row_filter, &read_rows, &eof,
+ &used, &kind));
+ } else {
+ auto dictionary = ColumnInt32::create();
+ for (int32_t i = 0; i < chunk.meta_data.num_values;
++i) {
+ dictionary->insert_value(100 + i);
+ }
+ size_t survivors = 0;
+ bool direct = false;
+ RETURN_IF_ERROR(reader.read_dictionary_filter(
+ IColumn::Filter(chunk.meta_data.num_values,
1), filter, batch_size,
+ dictionary.get(), projected.get(), nullptr,
&row_filter, &survivors,
+ &read_rows, &eof, &direct, &used));
+ EXPECT_EQ(survivors, read_rows);
+ }
+ EXPECT_TRUE(used);
+ EXPECT_EQ(row_filter, IColumn::Filter(read_rows, 1));
+ actual->insert(actual->end(),
projected->get_data().begin(),
+ projected->get_data().end());
+ } else {
+ RETURN_IF_ERROR(reader.read_column_data(output,
field.data_type, nullptr,
+ filter,
batch_size, &read_rows, &eof,
+ false));
+ if (field.data_type->is_nullable()) {
+ const auto& nullable_output = assert_cast<const
ColumnNullable&>(*output);
+ const auto& values =
+ assert_cast<const
ColumnInt32&>(nullable_output.get_nested_column())
+ .get_data();
+ for (size_t i = 0; i < values.size(); ++i) {
+ actual->push_back(nullable_output.is_null_at(i)
+ ?
std::numeric_limits<int32_t>::min()
+ : values[i]);
+ }
+ } else {
+ const auto& values = assert_cast<const
ColumnInt32&>(*output).get_data();
+ actual->insert(actual->end(), values.begin(),
values.end());
+ }
+ }
+ }
+ if (parsed_headers != nullptr) {
+ *parsed_headers =
reader.column_statistics().parse_page_header_num;
+ }
+ return Status::OK();
+ };
+ if (field.repetition_level != 0) {
+ return indexed ? read_with.template operator()<true, true>()
+ : read_with.template operator()<true, false>();
+ }
+ return indexed ? read_with.template operator()<false, true>()
+ : read_with.template operator()<false, false>();
+ }
+
+ bool rejects_fallback(const RowRanges& ranges, int stale_page) const {
+ if (stale_page <= 0 || !page_versions.front()) {
+ return false;
+ }
+ auto selected = [&](size_t row) {
+ for (size_t i = 0; i < ranges.range_size(); ++i) {
+ if (ranges.get_range_from(i) <= row && row <
ranges.get_range_to(i)) {
+ return true;
+ }
+ }
+ return false;
+ };
+ auto page_selected = [&](int page) {
+ const size_t start = index.page_locations[page].first_row_index;
+ const size_t end = page + 1 == page_versions.size()
+ ? rows
+ : index.page_locations[page +
1].first_row_index;
+ for (size_t row = start; row < end; ++row) {
+ if (selected(row)) {
+ return true;
+ }
+ }
+ return false;
+ };
+ if (!page_selected(stale_page)) {
+ return false;
+ }
+ for (int page = 0; page < stale_page; ++page) {
+ if (page > 0 && !page_selected(page)) {
+ return true;
+ }
+ const size_t end = index.page_locations[page + 1].first_row_index;
+ if (!page_versions[page] && !selected(end - 1)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ std::vector<int32_t> expected(const RowRanges& ranges) const {
+ std::vector<int32_t> result;
+ for (size_t i = 0; i < ranges.range_size(); ++i) {
+ for (size_t row = ranges.get_range_from(i); row <
ranges.get_range_to(i); ++row) {
+ result.insert(result.end(), row_values[row].begin(),
row_values[row].end());
+ }
+ }
+ return result;
+ }
+};
+
+TEST(ParquetV2NativeDecoderTest,
IncompleteOffsetIndexCannotRelabelMixedPageRows) {
+ for (const bool dictionary : {false, true}) {
+ NestedNavigationFile file({true, true, false}, {1, 1, 2}, true,
dictionary);
+ file.index.page_locations.erase(file.index.page_locations.begin() + 1);
+ file.index.page_locations.back().first_row_index = 1;
+ const ColumnChunkRange range {.offset = 1, .length = file.bytes.size()
- 1};
+ const bool usable = validate_offset_index(file.index, range,
+
file.chunk.meta_data.data_page_offset, file.rows);
+ EXPECT_FALSE(usable);
+ const auto selection = RowRanges::create_single(1, 2);
+ std::vector<int32_t> actual;
+ ASSERT_TRUE(file.read(selection, 1, usable, false, &actual).ok());
+ EXPECT_EQ(actual, file.expected(selection));
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest,
IndexGapsWithAuxiliaryPagesOrPaddingRemainReadable) {
+ for (const bool auxiliary : {false, true}) {
+ for (const bool dictionary : {false, true}) {
+ NestedNavigationFile file({true, false, true}, {1, 1, 1}, true,
dictionary);
+ std::vector<uint8_t> extra(16, 0);
+ if (auxiliary) {
+ tparquet::PageHeader header;
+ header.__set_type(tparquet::PageType::INDEX_PAGE);
+ header.__set_compressed_page_size(extra.size());
+ header.__set_uncompressed_page_size(extra.size());
+ header.__set_index_page_header(tparquet::IndexPageHeader {});
+ extra = serialize_page(header, extra);
+ const size_t offset = file.index.page_locations[1].offset;
+ file.bytes.insert(file.bytes.begin() + offset, extra.begin(),
extra.end());
+ for (size_t i = 1; i < file.index.page_locations.size(); ++i) {
+ file.index.page_locations[i].offset += extra.size();
+ }
+ } else {
+ file.bytes.insert(file.bytes.end(), extra.begin(),
extra.end());
+ }
+ file.chunk.meta_data.total_compressed_size += extra.size();
+ const bool usable = validate_offset_index(
+ file.index, {.offset = 1, .length = file.bytes.size() - 1},
+ file.chunk.meta_data.data_page_offset, file.rows);
+ EXPECT_FALSE(usable);
+ const auto ranges = RowRanges::create_single(0, 3);
+ std::vector<int32_t> actual;
+ ASSERT_TRUE(file.read(ranges, 1, usable, false, &actual).ok());
+ EXPECT_EQ(actual, file.expected(ranges));
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, OffsetIndexRequiresCompleteDataByteCoverage) {
+ const ColumnChunkRange range {.offset = 100, .length = 100};
+ tparquet::OffsetIndex index;
+ for (int i = 0; i < 3; ++i) {
+ tparquet::PageLocation location;
+ location.__set_offset(110 + i * 30);
+ location.__set_compressed_page_size(30);
+ location.__set_first_row_index(i);
+ index.page_locations.push_back(location);
+ }
+ // The prefix belongs to a dictionary; the complete data-page suffix is
still indexed.
+ ASSERT_TRUE(validate_offset_index(index, range, 110, 3));
+ for (size_t omitted = 0; omitted < 3; ++omitted) {
+ auto incomplete = index;
+ incomplete.page_locations.erase(incomplete.page_locations.begin() +
omitted);
+ EXPECT_FALSE(validate_offset_index(incomplete, range, 110, 3)) <<
omitted;
+ }
+ auto gap = index;
+ --gap.page_locations[0].compressed_page_size;
+ EXPECT_FALSE(validate_offset_index(gap, range, 110, 3));
+ // Unclassified trailing bytes could be padding, an auxiliary page, or an
omitted data page.
+ // Decline the optional index instead of interpreting those bytes as
harmless padding.
+ EXPECT_FALSE(validate_offset_index(index, {.offset = 100, .length = 101},
110, 3));
+}
+
+TEST(ParquetV2NativeDecoderTest, SparseIndexedReadsKeepUnselectedPagesLazy) {
+ for (int versions = 0; versions < 16; ++versions) {
+ for (const bool nested : {false, true}) {
+ for (const bool cache : {false, true}) {
+ for (int mode = 0; mode < 3; ++mode) {
+ SCOPED_TRACE(testing::Message()
+ << "versions=" << versions << ", nested=" <<
nested
+ << ", cache=" << cache << ", mode=" << mode);
+ NestedNavigationFile file({bool(versions & 1),
bool(versions & 2),
+ bool(versions & 4),
bool(versions & 8)},
+ {1, 3, 1, 2}, nested, mode == 2,
nested, nested);
+ const auto ranges = RowRanges::create_single(5, 6);
+ std::vector<int32_t> actual;
+ const bool levels_only = nested && mode == 1;
+ int64_t headers = 0;
+ const auto status = file.read(ranges, 1, true, cache,
&actual, levels_only,
+ nested ? 0 : mode, &headers);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(actual,
+ levels_only ? std::vector<int32_t>(1, 0) :
file.expected(ranges));
+ // A first nested V1 page retains the existing sequential
policy. Otherwise
+ // only the first and selected data headers (plus a
dictionary) are needed.
+ EXPECT_EQ(headers, (nested && !(versions & 1) ? 4 : 2) +
(mode == 2));
+ }
+ }
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, PartialNestedPageRejectsUnsafeFallback) {
+ for (const bool cache : {false, true}) {
+ for (const bool levels_only : {false, true}) {
+ NestedNavigationFile file({true, false, false}, {1, 3, 2}, true);
+ --file.index.page_locations.back().compressed_page_size;
+ auto ranges = RowRanges::create_single(1, 2);
+ ranges.add({4, 5});
+ std::vector<int32_t> actual;
+ const auto status = file.read(ranges, 1, true, cache, &actual,
levels_only);
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ EXPECT_EQ(actual, levels_only ? std::vector<int32_t>(1, 0) :
file.row_values[1]);
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, FinalNestedFallbackPreservesColumnValueCount)
{
+ for (const bool repeated : {false, true}) {
+ for (const bool cache : {false, true}) {
+ NestedNavigationFile file({true, false}, {1, 1}, repeated);
+ --file.index.page_locations.back().compressed_page_size;
+ const auto ranges = RowRanges::create_single(0, 2);
+ std::vector<int32_t> actual;
+ const auto status = file.read(ranges, 1, true, cache, &actual);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(actual, file.expected(ranges));
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest,
NestedNavigationPreservesRowsAcrossPageVersionTransitions) {
+ for (int versions = 0; versions < 8; ++versions) {
+ for (const bool indexed : {false, true}) {
+ for (const bool dictionary : {false, true}) {
+ for (const bool nullable : {false, true}) {
+ for (const bool cache : {false, true}) {
+ for (const size_t batch : {1, 2, 7}) {
+ for (int selection = 0; selection < 3;
++selection) {
+ SCOPED_TRACE(testing::Message()
+ << "versions=" << versions << ",
indexed=" << indexed
+ << ", dictionary=" << dictionary
+ << ", nullable=" << nullable <<
", cache=" << cache
+ << ", batch=" << batch << ",
selection=" << selection);
+ NestedNavigationFile file({bool(versions & 1),
bool(versions & 2),
+ bool(versions & 4)},
+ {2, 3, 2}, true,
dictionary, nullable);
+ auto ranges = RowRanges::create_single(0, 7);
+ if (selection == 1) {
+ ranges = RowRanges::create_single(1, 2);
+ ranges.add({3, 4});
+ ranges.add({6, 7});
+ } else if (selection == 2) {
+ ranges = RowRanges::create_single(6, 7);
+ }
+ for (const bool levels_only : {false, true}) {
+ std::vector<int32_t> actual;
+ const auto status = file.read(ranges,
batch, indexed, cache,
+ &actual,
levels_only);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(actual,
+ levels_only ?
std::vector<int32_t>(ranges.count(), 0)
+ :
file.expected(ranges));
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, NestedNavigationFallbackMatrix) {
+ for (int versions = 0; versions < 8; ++versions) {
+ for (int stale_page = 0; stale_page < 3; ++stale_page) {
+ for (const bool cache : {false, true}) {
+ for (const bool dictionary : {false, true}) {
+ for (const size_t batch : {1, 2, 7}) {
+ for (int selection = 0; selection < 3; ++selection) {
+ SCOPED_TRACE(testing::Message()
+ << "versions=" << versions << ",
stale=" << stale_page
+ << ", cache=" << cache << ",
dictionary=" << dictionary
+ << ", batch=" << batch << ",
selection=" << selection);
+ NestedNavigationFile file(
+ {bool(versions & 1), bool(versions & 2),
bool(versions & 4)},
+ {2, 3, 2}, true, dictionary, true);
+
--file.index.page_locations[stale_page].compressed_page_size;
+ auto ranges = RowRanges::create_single(0, 7);
+ if (selection == 1) {
+ ranges = RowRanges::create_single(1, 2);
+ ranges.add({3, 4});
+ ranges.add({6, 7});
+ } else if (selection == 2) {
+ ranges = RowRanges::create_single(6, 7);
+ }
+ for (const bool levels_only : {false, true}) {
+ std::vector<int32_t> actual;
+ const auto status =
+ file.read(ranges, batch, true, cache,
&actual, levels_only);
+ if (file.rejects_fallback(ranges, stale_page))
{
+
EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ continue;
+ }
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(actual, levels_only
+ ?
std::vector<int32_t>(ranges.count(), 0)
+ :
file.expected(ranges));
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, NestedFallbackReevaluatesSelectedPageRange) {
+ NestedNavigationFile file({true, true, false}, {2, 1, 3}, true);
+ file.index.page_locations[2].first_row_index = 4;
+ --file.index.page_locations[1].compressed_page_size;
+ const auto ranges = RowRanges::create_single(3, 4);
+ std::vector<int32_t> actual;
+ const auto status = file.read(ranges, 1, true, false, &actual);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(actual, file.expected(ranges));
+}
+
+TEST(ParquetV2NativeDecoderTest, NestedNavigationAllRowSelections) {
+ for (int versions = 0; versions < 8; ++versions) {
+ for (unsigned mask = 1; mask < 128; ++mask) {
+ RowRanges ranges;
+ for (int64_t row = 0; row < 7; ++row) {
+ if (mask & (1U << row)) {
+ ranges.add({row, row + 1});
+ }
+ }
+ // Every nonempty row subset is checked against physical values.
Vary encoding, nulls
+ // and cache independently of the navigation policy, including
partial and whole skips.
+ for (int index_mode = -2; index_mode < 3; ++index_mode) {
+ NestedNavigationFile file(
+ {bool(versions & 1), bool(versions & 2), bool(versions
& 4)}, {2, 3, 2},
+ true, mask & 1, mask & 2);
+ if (index_mode >= 0) {
+
--file.index.page_locations[index_mode].compressed_page_size;
+ }
+ for (const size_t batch : {1, 2, 7}) {
+ for (const bool levels_only : {false, true}) {
+ SCOPED_TRACE(testing::Message()
+ << "versions=" << versions << ", mask="
<< mask
+ << ", index_mode=" << index_mode << ",
batch=" << batch
+ << ", levels_only=" << levels_only);
+ std::vector<int32_t> actual;
+ const auto status = file.read(ranges, batch,
index_mode != -2, mask & 4,
+ &actual, levels_only);
+ if (file.rejects_fallback(ranges, index_mode)) {
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) <<
status;
+ continue;
+ }
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(actual, levels_only ?
std::vector<int32_t>(ranges.count(), 0)
+ : file.expected(ranges));
+ }
+ }
+ }
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest,
IndexedNestedFallbackRequiresCompleteV1RowSpan) {
+ for (const bool middle_v2 : {false, true}) {
+ for (const bool partial : {false, true}) {
+ NestedNavigationFile file({true, middle_v2, false}, {1, 3, 2},
true);
+ --file.index.page_locations.back().compressed_page_size;
+ MemoryBufferedReader stream(file.bytes);
+ ColumnChunkReader<true, true> reader(&stream, &file.chunk,
&file.field, &file.index,
+ file.rows, nullptr,
ParquetPageReadContext());
+ ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ ASSERT_TRUE(reader.next_page().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ ASSERT_TRUE(reader.load_page_data().ok());
+ std::vector<level_t> levels;
+ size_t rows = 0;
+ bool cross_page = false;
+ ASSERT_TRUE(
+ reader.load_page_nested_rows(levels, partial ? 1 : 3,
&rows, &cross_page).ok());
+ ASSERT_TRUE(reader.next_page().ok());
+ const auto status = reader.parse_page_header();
+ if (partial && !middle_v2) {
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ } else {
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(reader.page_start_row(), 4);
+ EXPECT_EQ(reader._chunk_parsed_values,
file.chunk.meta_data.num_values);
+ }
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest,
IndexedNestedNavigationSkipsHeadersAndPayloads) {
+ for (const bool middle_v2 : {false, true}) {
+ for (const size_t page_count : {3, 64}) {
+ std::vector<bool> versions(page_count, middle_v2);
+ versions.front() = true;
+ versions.back() = false;
+ NestedNavigationFile file(versions,
std::vector<size_t>(page_count, 3), true);
+ MemoryBufferedReader stream(file.bytes);
+ ColumnChunkReader<true, true> reader(&stream, &file.chunk,
&file.field, &file.index,
+ file.rows, nullptr,
ParquetPageReadContext());
+ ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.seek_to_nested_row(file.rows - 1).ok());
+ const auto& statistics = reader.chunk_statistics();
+ EXPECT_EQ(statistics.parse_page_header_num, 2);
+ EXPECT_EQ(statistics.skip_page_header_num, page_count - 2);
+ // Only the first and selected headers and the selected payload
need reading, regardless
+ // of the skipped page count. Skipping V1 pages must not load
their repetition levels.
+ EXPECT_EQ(stream.read_count(), 3);
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest,
IndexedNestedPartialPageAdvanceDoesNotDrainLevels) {
+ for (const bool partial : {false, true}) {
+ NestedNavigationFile file({true, false, true}, {1, 3, 2}, true);
+ MemoryBufferedReader stream(file.bytes);
+ ColumnChunkReader<true, true> reader(&stream, &file.chunk,
&file.field, &file.index,
+ file.rows, nullptr,
ParquetPageReadContext());
+ ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ ASSERT_TRUE(reader.next_page().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ if (partial) {
+ ASSERT_TRUE(reader.load_page_data().ok());
+ std::vector<level_t> levels;
+ size_t rows = 0;
+ bool cross_page = false;
+ ASSERT_TRUE(reader.load_page_nested_rows(levels, 1, &rows,
&cross_page).ok());
+ ASSERT_EQ(rows, 1);
+ }
+ const auto reads = stream.read_count();
+ const auto remaining = reader._remaining_rep_nums;
+ ASSERT_GT(remaining, 0);
+ ASSERT_TRUE(reader.next_page().ok());
+ EXPECT_EQ(stream.read_count(), reads);
+ EXPECT_EQ(reader._remaining_rep_nums, remaining);
+ EXPECT_EQ(reader.page_start_row(), 4);
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest, IndexedNestedSeekReadsMixedPageVersions) {
+ for (const bool bad_span : {false, true}) {
+ for (const bool with_index : {false, true}) {
+ for (const bool continuation : {false, true}) {
+ if ((!with_index && (continuation || bad_span)) ||
(continuation && bad_span)) {
+ continue;
+ }
+ SCOPED_TRACE(testing::Message() << "index=" << with_index <<
", continuation="
+ << continuation << ",
bad_span=" << bad_span);
+ std::vector<uint8_t> bytes;
+ tparquet::OffsetIndex index;
+ for (int32_t i = 0; i < 3; ++i) {
+ const bool v1 = i == 2;
+ std::vector<uint8_t> payload;
+ if (v1) {
+ payload = {2, 0, 0, 0};
+ }
+ payload.push_back(2);
+ payload.push_back(v1 && continuation ? 1 : 0);
+ const int32_t value = (i + 1) * 10;
+ const auto* raw = reinterpret_cast<const uint8_t*>(&value);
+ payload.insert(payload.end(), raw, raw + sizeof(value));
+ tparquet::PageHeader header;
+ header.__set_type(v1 ? tparquet::PageType::DATA_PAGE
+ : tparquet::PageType::DATA_PAGE_V2);
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ if (v1) {
+ tparquet::DataPageHeader data;
+ data.__set_num_values(1);
+ data.__set_encoding(tparquet::Encoding::PLAIN);
+
data.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+
data.__set_definition_level_encoding(tparquet::Encoding::RLE);
+ header.__set_data_page_header(data);
+ } else {
+ tparquet::DataPageHeaderV2 data;
+ data.__set_num_values(1);
+ data.__set_num_rows(1);
+ data.__set_num_nulls(0);
+ data.__set_encoding(tparquet::Encoding::PLAIN);
+ data.__set_repetition_levels_byte_length(2);
+ data.__set_definition_levels_byte_length(0);
+ data.__set_is_compressed(false);
+ header.__set_data_page_header_v2(data);
+ }
+ const auto page = serialize_page(header, payload);
+ tparquet::PageLocation location;
+ location.__set_offset(bytes.size());
+ location.__set_compressed_page_size(page.size());
+ location.__set_first_row_index(i);
+ index.page_locations.push_back(location);
+ bytes.insert(bytes.end(), page.begin(), page.end());
+ }
+ tparquet::ColumnChunk chunk;
+ chunk.meta_data.__set_type(tparquet::Type::INT32);
+
chunk.meta_data.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ chunk.meta_data.__set_num_values(3);
+ chunk.meta_data.__set_total_compressed_size(bytes.size());
+ chunk.meta_data.__set_data_page_offset(0);
+ NativeFieldSchema field;
+ field.physical_type = tparquet::Type::INT32;
+ field.repetition_level = 1;
+ MemoryBufferedReader stream(bytes);
+ ColumnChunkReader<true, true> reader(
+ &stream, &chunk, &field, with_index ? &index :
nullptr, bad_span ? 4 : 3,
+ nullptr, ParquetPageReadContext(false, ""));
+ ASSERT_TRUE(reader.init().ok());
+ auto status = reader.seek_to_nested_row(2);
+ if (continuation) {
+ // An indexed page must start a row; accepting a
continuation would relabel it.
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ continue;
+ }
+ ASSERT_TRUE(status.ok()) << status;
+ std::vector<level_t> reps;
+ size_t rows = 0;
+ bool cross_page = false;
+ status = reader.load_page_nested_rows(reps, 1, &rows,
&cross_page);
+ if (bad_span) {
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ continue;
+ }
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(rows, 1);
+ EXPECT_EQ(reps, (std::vector<level_t> {0}));
+ EXPECT_FALSE(cross_page);
+ EXPECT_FALSE(reader.has_next_page());
+ std::vector<level_t> defs;
+ ASSERT_TRUE(reader.fill_def(defs).ok());
+ DataTypeInt32 type;
+ auto output = type.create_column();
+ ParquetDecodeContext context;
+ context.physical_type = ParquetPhysicalType::INT32;
+ ParquetMaterializationState state;
+ FilterMap filter;
+ ASSERT_TRUE(filter.init(nullptr, 1, false).ok());
+ ColumnSelectVector selection;
+ const std::vector<uint16_t> null_runs {1};
+ ASSERT_TRUE(selection.init(null_runs, 1, nullptr, &filter,
0).ok());
+ ASSERT_TRUE(reader.materialize_values(output,
*type.get_serde(), context, state,
+ selection)
+ .ok());
+ EXPECT_EQ(assert_cast<const ColumnInt32&>(*output).get_data(),
+ (ColumnInt32::Container {30}));
+ }
+ }
+ }
+}
+
+TEST(ParquetV2NativeDecoderTest,
LateOffsetIndexFallbackRequiresVerifiedRowCoordinates) {
+ auto check = []<bool NESTED>(bool skip_middle, bool stale_size, bool
cache_hit) {
+ SCOPED_TRACE(testing::Message() << "nested=" << NESTED << ", skip=" <<
skip_middle
+ << ", stale=" << stale_size << ",
cache=" << cache_hit);
+ const bool reject = skip_middle && stale_size;
+ const std::vector<std::vector<int32_t>> values {{10}, {20}, {30, 31}};
+ std::vector<uint8_t> bytes;
+ std::vector<std::vector<uint8_t>> pages;
+ tparquet::OffsetIndex index;
+ for (size_t page = 0; page < values.size(); ++page) {
+ std::vector<uint8_t> payload;
+ if constexpr (NESTED) {
+ payload = {cast_set<uint8_t>(values[page].size() * 2), 0};
+ }
+ const size_t level_bytes = payload.size();
+ const auto* raw = reinterpret_cast<const
uint8_t*>(values[page].data());
+ payload.insert(payload.end(), raw, raw + values[page].size() *
sizeof(int32_t));
+ tparquet::PageHeader header;
+ header.__set_type(tparquet::PageType::DATA_PAGE_V2);
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ header.__isset.data_page_header_v2 = true;
+ auto& data = header.data_page_header_v2;
+ data.__set_num_values(values[page].size());
+ data.__set_num_rows(values[page].size());
+ data.__set_num_nulls(0);
+ data.__set_encoding(tparquet::Encoding::PLAIN);
+ data.__set_repetition_levels_byte_length(level_bytes);
+ data.__set_definition_levels_byte_length(0);
+ data.__set_is_compressed(false);
+ pages.push_back(serialize_page(header, payload));
+ tparquet::PageLocation location;
+ location.__set_offset(bytes.size());
+ location.__set_compressed_page_size(pages.back().size() -
+ (stale_size && page == 2 ? 1 :
0));
+ // A skipped middle page hides the extra indexed row; fallback
must not label physical
+ // row 2 as row 3 merely because the last page's own header is
internally consistent.
+ location.__set_first_row_index(page == 2 && reject ? 3 : page);
+ index.page_locations.push_back(location);
+ bytes.insert(bytes.end(), pages.back().begin(),
pages.back().end());
+ }
+ // Production loaders discard the short-size index early. Inject it
directly here to
+ // retain coverage of the reader's defensive header reconciliation as
well.
+ EXPECT_EQ(validate_offset_index(index, {.offset = 0, .length =
bytes.size()}, 0, 4),
+ !stale_size);
+ const std::string cache_key =
+ fmt::format("late-index-{}-{}-{}", NESTED, skip_middle,
stale_size);
+ if (cache_hit) {
+ auto* cached = new DataPage(pages.back().size(), true,
segment_v2::DATA_PAGE);
+ memcpy(cached->data(), pages.back().data(), pages.back().size());
+ cached->reset_size(pages.back().size());
+ PageCacheHandle handle;
+ StoragePageCache::instance()->insert(
+ StoragePageCache::CacheKey(cache_key, bytes.size(),
+
index.page_locations.back().offset),
+ cached, &handle, segment_v2::DATA_PAGE);
+ }
+ tparquet::ColumnChunk chunk;
+ chunk.meta_data.__set_type(tparquet::Type::INT32);
+ chunk.meta_data.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ chunk.meta_data.__set_num_values(4);
+ chunk.meta_data.__set_total_compressed_size(bytes.size());
+ chunk.meta_data.__set_data_page_offset(0);
+ chunk.meta_data.__set_encodings({tparquet::Encoding::PLAIN});
+ NativeFieldSchema field;
+ field.physical_type = tparquet::Type::INT32;
+ field.data_type = std::make_shared<DataTypeInt32>();
+ field.repetition_level = NESTED ? 1 : 0;
+ field.parquet_schema.__set_type(tparquet::Type::INT32);
+
field.parquet_schema.__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ const ParquetPageReadContext context(cache_hit, cache_key);
+ Status status;
+ if constexpr (NESTED) {
+ MemoryBufferedReader stream(bytes);
+ ColumnChunkReader<true, true> reader(&stream, &chunk, &field,
&index, 4, nullptr,
+ context);
+ ASSERT_TRUE(reader.init().ok());
+ if (!skip_middle) {
+ ASSERT_TRUE(reader.seek_to_nested_row(1).ok());
+ }
+ status = reader.seek_to_nested_row(3);
+ EXPECT_EQ(reader.chunk_statistics().page_cache_hit_counter > 0,
cache_hit);
+ } else {
+ auto ranges = ::doris::RowRanges::create_single(skip_middle ? 3 :
0, 4);
+ auto file = std::make_shared<NativeDecoderMemoryFileReader>(bytes);
+ ScalarColumnReader<false, true> reader(ranges, 4, chunk, &index,
nullptr, nullptr);
+ ASSERT_TRUE(reader.init(file, &field, bytes.size(), nullptr,
cache_hit ? cache_key : "",
+ ParquetReaderCompat {}, true)
+ .ok());
+ FilterMap filter;
+ ASSERT_TRUE(filter.init(nullptr, ranges.count(), false).ok());
+ ColumnPtr output = ColumnInt32::create();
+ size_t consumed = 0;
+ size_t calls = 0;
+ bool eof = false;
+ while (consumed < ranges.count() && !eof) {
+ ASSERT_LE(++calls, pages.size());
+ size_t rows = 0;
+ status = reader.read_column_data(output, field.data_type,
nullptr, filter,
+ ranges.count() - consumed,
&rows, &eof, false);
+ if (!status.ok()) {
+ break;
+ }
+ consumed += rows;
+ }
+ EXPECT_EQ(reader.column_statistics().page_cache_hit_counter > 0,
cache_hit);
+ if (!reject) {
+ EXPECT_EQ(assert_cast<const ColumnInt32&>(*output).get_data(),
+ (skip_middle ? ColumnInt32::Container {31}
+ : ColumnInt32::Container {10, 20, 30,
31}));
+ } else {
+ EXPECT_EQ(output->size(), 0);
+ }
+ }
+ if (reject) {
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ } else {
+ EXPECT_TRUE(status.ok()) << status;
+ }
+ };
+ for (const bool skip_middle : {false, true}) {
+ for (const bool stale_size : {false, true}) {
+ for (const bool cache_hit : {false, true}) {
+ check.template operator()<false>(skip_middle, stale_size,
cache_hit);
+ check.template operator()<true>(skip_middle, stale_size,
cache_hit);
+ }
+ }
+ }
+}
+
TEST(ParquetV2NativeDecoderTest,
FlatPagesRejectLogicalAndPhysicalCardinalityMismatch) {
auto init_chunk = [](tparquet::PageHeader header, bool with_offset_index) {
std::vector<uint8_t>
payload(static_cast<size_t>(header.compressed_page_size), 0);
@@ -4492,6 +5638,8 @@ TEST(ParquetV2NativeDecoderTest,
OffsetIndexValidationRejectsBackwardAndOverlapp
EXPECT_FALSE(validate_offset_index(index, range, 110, 20));
second.__set_first_row_index(10);
+ second.__set_offset(130);
+ second.__set_compressed_page_size(70);
index.page_locations = {first, second};
EXPECT_TRUE(validate_offset_index(index, range, 110, 20));
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index c045683e424..d8934a4f690 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -1859,6 +1859,114 @@ void write_misdeclared_two_page_parquet_file(const
std::string& file_path) {
DORIS_CHECK(output.good());
}
+void shorten_first_offset_index_page(const std::string& file_path) {
+ std::ifstream input(file_path, std::ios::binary | std::ios::ate);
+ ASSERT_TRUE(input.good());
+ const auto input_size = static_cast<std::streamoff>(input.tellg());
+ ASSERT_GE(input_size, 8);
+ std::vector<uint8_t> bytes(cast_set<size_t>(input_size));
+ input.seekg(0);
+ input.read(reinterpret_cast<char*>(bytes.data()), bytes.size());
+ ASSERT_TRUE(input.good());
+ input.close();
+ const uint32_t footer_size = decode_fixed32_le(bytes.data() + bytes.size()
- 8);
+ const size_t footer_offset = bytes.size() - 8 - footer_size;
+ uint32_t thrift_size = footer_size;
+ tparquet::FileMetaData metadata;
+ ASSERT_TRUE(deserialize_thrift_msg(bytes.data() + footer_offset,
&thrift_size, true, &metadata)
+ .ok());
+ auto& column = metadata.row_groups[0].columns[0];
+ ASSERT_TRUE(column.__isset.offset_index_offset);
+ tparquet::OffsetIndex index;
+ thrift_size = column.offset_index_length;
+ ASSERT_TRUE(deserialize_thrift_msg(bytes.data() +
column.offset_index_offset, &thrift_size,
+ true, &index)
+ .ok());
+ ASSERT_GE(index.page_locations.size(), 3);
+ // Keep the Arrow-written data intact; only the optional index's first
page size is stale.
+ --index.page_locations[0].compressed_page_size;
+ std::vector<uint8_t> serialized_index;
+ ThriftSerializer serializer(true, 1024);
+ ASSERT_TRUE(serializer.serialize(&index, &serialized_index).ok());
+ bytes.resize(footer_offset);
+ column.__set_offset_index_offset(bytes.size());
+ column.__set_offset_index_length(serialized_index.size());
+ bytes.insert(bytes.end(), serialized_index.begin(),
serialized_index.end());
+ std::vector<uint8_t> footer;
+ ASSERT_TRUE(serializer.serialize(&metadata, &footer).ok());
+ bytes.insert(bytes.end(), footer.begin(), footer.end());
+ std::array<uint8_t, 4> length {};
+ encode_fixed32_le(length.data(), cast_set<uint32_t>(footer.size()));
+ bytes.insert(bytes.end(), length.begin(), length.end());
+ bytes.insert(bytes.end(), {'P', 'A', 'R', '1'});
+ std::ofstream output(file_path, std::ios::binary | std::ios::trunc);
+ output.write(reinterpret_cast<const char*>(bytes.data()), bytes.size());
+ ASSERT_TRUE(output.good());
+}
+
+void omit_page_from_indexes(const std::string& file_path, int omitted,
int32_t* matching_value) {
+ std::ifstream input(file_path, std::ios::binary | std::ios::ate);
+ ASSERT_TRUE(input.good());
+ std::vector<uint8_t>
bytes(cast_set<size_t>(static_cast<std::streamoff>(input.tellg())));
+ input.seekg(0);
+ input.read(reinterpret_cast<char*>(bytes.data()), bytes.size());
+ ASSERT_TRUE(input.good());
+ input.close();
+ ASSERT_GE(bytes.size(), 8);
+ const uint32_t footer_size = decode_fixed32_le(bytes.data() + bytes.size()
- 8);
+ const size_t footer_offset = bytes.size() - 8 - footer_size;
+ uint32_t length = footer_size;
+ tparquet::FileMetaData metadata;
+ ASSERT_TRUE(
+ deserialize_thrift_msg(bytes.data() + footer_offset, &length,
true, &metadata).ok());
+ auto& column = metadata.row_groups[0].columns[0];
+ tparquet::OffsetIndex offsets;
+ length = column.offset_index_length;
+ ASSERT_TRUE(deserialize_thrift_msg(bytes.data() +
column.offset_index_offset, &length, true,
+ &offsets)
+ .ok());
+ tparquet::ColumnIndex statistics;
+ length = column.column_index_length;
+ ASSERT_TRUE(deserialize_thrift_msg(bytes.data() +
column.column_index_offset, &length, true,
+ &statistics)
+ .ok());
+ ASSERT_GE(offsets.page_locations.size(), 3);
+ const size_t page = omitted < 0 ? offsets.page_locations.size() - 1 :
omitted;
+ ASSERT_LT(page, offsets.page_locations.size());
+ ASSERT_EQ(statistics.min_values[page].size(), sizeof(int32_t));
+ *matching_value = cast_set<int32_t>(decode_fixed32_le(
+ reinterpret_cast<const
uint8_t*>(statistics.min_values[page].data())));
+ // Keep the physical page containing the sole predicate match intact, but
omit it from both
+ // indexes. Equal index cardinalities must not make an incomplete page
list safe for pruning.
+ offsets.page_locations.erase(offsets.page_locations.begin() + page);
+ statistics.null_pages.erase(statistics.null_pages.begin() + page);
+ statistics.min_values.erase(statistics.min_values.begin() + page);
+ statistics.max_values.erase(statistics.max_values.begin() + page);
+ if (statistics.__isset.null_counts) {
+ statistics.null_counts.erase(statistics.null_counts.begin() + page);
+ }
+ bytes.resize(footer_offset);
+ ThriftSerializer serializer(true, 1024);
+ std::vector<uint8_t> serialized;
+ ASSERT_TRUE(serializer.serialize(&offsets, &serialized).ok());
+ column.__set_offset_index_offset(bytes.size());
+ column.__set_offset_index_length(serialized.size());
+ bytes.insert(bytes.end(), serialized.begin(), serialized.end());
+ ASSERT_TRUE(serializer.serialize(&statistics, &serialized).ok());
+ column.__set_column_index_offset(bytes.size());
+ column.__set_column_index_length(serialized.size());
+ bytes.insert(bytes.end(), serialized.begin(), serialized.end());
+ ASSERT_TRUE(serializer.serialize(&metadata, &serialized).ok());
+ bytes.insert(bytes.end(), serialized.begin(), serialized.end());
+ std::array<uint8_t, 4> footer_length {};
+ encode_fixed32_le(footer_length.data(),
cast_set<uint32_t>(serialized.size()));
+ bytes.insert(bytes.end(), footer_length.begin(), footer_length.end());
+ bytes.insert(bytes.end(), {'P', 'A', 'R', '1'});
+ std::ofstream output(file_path, std::ios::binary | std::ios::trunc);
+ output.write(reinterpret_cast<const char*>(bytes.data()), bytes.size());
+ ASSERT_TRUE(output.good());
+}
+
void write_long_prefix_parquet_file(const std::string& file_path, size_t rows)
{
std::vector<int32_t> ids(rows);
std::vector<int32_t> first(rows);
@@ -4623,6 +4731,124 @@ TEST_F(ParquetScanTest,
PredicateOnlyUint32UsesConvertedDecoderDirectPath) {
EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
}
+TEST_F(ParquetScanTest, IncompletePageIndexesRetainTheOnlyPredicateMatch) {
+ const bool original = config::enable_parquet_page_index;
+ DEFER({ config::enable_parquet_page_index = original; });
+ for (const bool dictionary : {false, true}) {
+ for (const int omitted : {0, 1, -1}) {
+ std::vector<int32_t> ids(128);
+ std::iota(ids.begin(), ids.end(), 0);
+ auto table = arrow::Table::Make(
+ arrow::schema({arrow::field("value", arrow::int32(),
false)}),
+ {build_int32_array(ids)});
+ write_table(_file_path, table, ids.size(), dictionary, true);
+ int32_t expected = -1;
+ omit_page_from_indexes(_file_path, omitted, &expected);
+ for (const bool enabled : {false, true}) {
+ SCOPED_TRACE(testing::Message() << "dictionary=" << dictionary
<< ", omitted="
+ << omitted << ", index=" <<
enabled);
+ config::enable_parquet_page_index = enabled;
+ RuntimeProfile profile("incomplete_index");
+ auto reader = create_reader(0, -1, &profile);
+ reader->set_batch_size(1);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder builder(request.get());
+
ASSERT_TRUE(builder.add_predicate_column(format::LocalColumnId(0)).ok());
+ auto conjunct =
+ create_int32_function_conjunct(0, "eq",
TExprOpcode::EQ, expected, false);
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ DEFER({ conjunct->close(); });
+ request->conjuncts.push_back(conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<int32_t> actual;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ for (size_t row = 0; row < rows; ++row) {
+
actual.push_back(int32_data_column(*block.get_by_position(0).column)
+ .get_element(row));
+ }
+ }
+ EXPECT_EQ(actual, std::vector<int32_t> {expected});
+ EXPECT_EQ(counter_value(profile, "FilteredRowsByPage"), 0);
+ EXPECT_EQ(counter_value(profile, "PageIndexReadCalls"), 0);
+ }
+ }
+ }
+}
+
+TEST_F(ParquetScanTest, OffsetIndexFallbackPreservesFilteredRowsAcrossPages) {
+ std::vector<int32_t> values(24);
+ std::iota(values.begin(), values.end(), 0);
+ std::vector<int32_t> payloads(24);
+ std::iota(payloads.begin(), payloads.end(), 100);
+ auto arrow_schema = arrow::schema({arrow::field("value", arrow::int32(),
false),
+ arrow::field("payload", arrow::int32(),
false)});
+ auto table = arrow::Table::Make(arrow_schema,
+ {build_int32_array(values),
build_int32_array(payloads)});
+ const bool original_page_index = config::enable_parquet_page_index;
+ DEFER({ config::enable_parquet_page_index = original_page_index; });
+ for (const bool dictionary : {false, true}) {
+ write_table(_file_path, table, 24, dictionary, true);
+ shorten_first_offset_index_page(_file_path);
+ for (const bool enable_index : {false, true}) {
+ for (const int batch_size : {5, 64}) {
+ SCOPED_TRACE(testing::Message() << "dictionary=" << dictionary
<< ", index="
+ << enable_index << ", batch="
<< batch_size);
+ config::enable_parquet_page_index = enable_index;
+ RuntimeProfile profile("index_fallback");
+ auto reader = create_reader(0, -1, &profile);
+ reader->set_batch_size(batch_size);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder builder(request.get());
+
ASSERT_TRUE(builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ auto conjunct = create_int32_function_conjunct(0, "gt",
TExprOpcode::GT, 0, false);
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ DEFER({ conjunct->close(); });
+ ASSERT_TRUE(conjunct->root()->can_evaluate_zonemap_filter());
+ request->conjuncts.push_back(conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<int32_t> actual_values;
+ std::vector<int32_t> actual_payloads;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ const auto status = reader->get_block(&block, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ for (size_t row = 0; row < rows; ++row) {
+
actual_values.push_back(int32_data_column(*block.get_by_position(0).column)
+ .get_element(row));
+ actual_payloads.push_back(
+
int32_data_column(*block.get_by_position(1).column)
+ .get_element(row));
+ }
+ }
+ EXPECT_EQ(actual_values, (std::vector<int32_t>(values.begin()
+ 1, values.end())));
+ EXPECT_EQ(actual_payloads,
+ (std::vector<int32_t>(payloads.begin() + 1,
payloads.end())));
+ EXPECT_GT(counter_value(profile, dictionary ?
"DictionaryPredicateDirectBatches"
+ :
"FixedWidthPredicateDirectBatches"),
+ 0);
+ EXPECT_EQ(counter_value(profile, "PageIndexReadCalls") > 0,
enable_index);
+ }
+ }
+ }
+}
+
TEST_F(ParquetScanTest,
FixedWidthPredicateReportsUnsupportedEncodingAfterProgress) {
write_misdeclared_two_page_parquet_file(_file_path);
RuntimeProfile profile("profile");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
