This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 5b3ac63f8b4 [improvement](parquet) Make V2 column initialization lazy
(#66073)
5b3ac63f8b4 is described below
commit 5b3ac63f8b49018265f7a80db9e2b86db67c4cbb
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Jul 27 10:16:28 2026 +0800
[improvement](parquet) Make V2 column initialization lazy (#66073)
## Proposed changes
Copy of #66073, close #66073
- make File Scanner V2 Parquet column chunk initialization perform no
page I/O
- lazily detect and decode dictionary pages on the first operation that
needs page state
- preserve dictionary initialization before sequential advances and
OffsetIndex seeks
- add regression coverage for zero-I/O initialization, dictionary
probing, and indexed page skipping
## Test
- `./run-be-ut.sh --run --filter=ParquetV2NativeDecoderTest.* -j 24`
(106 tests passed)
- `build-support/check-format.sh`
- changed-line `clang-tidy`
---------
Co-authored-by: Gabriel <[email protected]>
---
.../parquet/reader/native/column_chunk_reader.cpp | 60 ++-
.../parquet/reader/native/column_chunk_reader.h | 7 +-
.../parquet/reader/native/column_reader.cpp | 25 +-
.../iceberg_position_delete_sys_table_reader.cpp | 10 +-
be/src/runtime/runtime_profile.cpp | 17 +
be/src/runtime/runtime_profile.h | 5 +
be/test/format_v2/parquet/native_decoder_test.cpp | 506 ++++++++++++++++++++-
be/test/runtime/runtime_profile_test.cpp | 31 ++
8 files changed, 624 insertions(+), 37 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 0d59b7e6bc2..b7bea8e7053 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
@@ -844,7 +844,6 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::init() {
// 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();
}
@@ -908,30 +907,56 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::read_levels(
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_parse_first_page_header() {
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_ensure_dictionary_page_loaded() {
+ if (_dict_checked) {
+ return Status::OK();
+ }
+
+ DORIS_CHECK(_state == INITIALIZED);
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 constexpr (IN_COLLECTION && OFFSET_INDEX) {
+ if (header->type == tparquet::PageType::DATA_PAGE &&
+ _page_reader->has_active_offset_index()) {
+ // V1 nested pages expose row boundaries only in
repetition levels, so an
+ // indexed seek must not skip the first page before those
levels are decoded.
+ _page_reader->discard_offset_index();
+ _offset_index = nullptr;
+ }
+ }
+ _dict_checked = true;
+ return Status::OK();
}
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.
+ // A nested V1 chunk must inspect its first data-page type before
indexed seeking can skip
+ // that page; dictionary discovery alone is not enough to make the
OffsetIndex trustworthy.
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::load_dictionary_page(bool* has_dict) {
+ RETURN_IF_ERROR(_ensure_dictionary_page_loaded());
+ *has_dict = _has_dict;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::ensure_first_data_page_parsed() {
+ if (_first_data_page_parsed) {
+ 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.
+ return parse_page_header();
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
@@ -939,6 +964,7 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::parse_page_header() {
if (_state == HEADER_PARSED || _state == DATA_LOADED) {
return Status::OK();
}
+ RETURN_IF_ERROR(_ensure_dictionary_page_loaded());
const tparquet::PageHeader* header = nullptr;
while (true) {
RETURN_IF_ERROR(_page_reader->parse_page_header());
@@ -1004,12 +1030,19 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::parse_page_header() {
if (!active_offset_index) {
_chunk_parsed_values += _remaining_num_values;
}
+ _first_data_page_parsed = true;
_state = HEADER_PARSED;
return Status::OK();
}
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());
+ } else {
+ // Load dictionary state before advancing can jump past the physical
dictionary page.
+ RETURN_IF_ERROR(_ensure_dictionary_page_loaded());
+ }
// Level parsing advances _page_data past the allocation base, so retain
explicit ownership
// state instead of inferring whether current decoders still reference
decompressed storage.
_page_uses_decompress_buf = false;
@@ -1022,8 +1055,8 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::next_page() {
_decompress_release_pending = false;
_decompress_release_threshold = std::numeric_limits<size_t>::max();
}
- _state = INITIALIZED;
RETURN_IF_ERROR(_page_reader->next_page());
+ _state = INITIALIZED;
return Status::OK();
}
@@ -1909,6 +1942,9 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::filter_dictionary_indices
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::seek_to_nested_row(size_t left_row) {
+ 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) {
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 4a6dbecd4c5..5b6a7ad8be8 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
@@ -220,7 +220,7 @@ public:
level_t max_rep_level() const { return _max_rep_level; }
level_t max_def_level() const { return _max_def_level; }
- bool has_dict() const { return _has_dict; };
+ Status load_dictionary_page(bool* has_dict);
// Get page decoder
Decoder* get_page_decoder() { return _page_decoder; }
@@ -306,6 +306,7 @@ public:
size_t page_end_row() const { return _page_reader->end_row(); }
+ Status ensure_first_data_page_parsed();
Status parse_page_header();
Status next_page();
@@ -346,8 +347,7 @@ public:
private:
enum ColumnChunkReaderState { NOT_INIT, INITIALIZED, HEADER_PARSED,
DATA_LOADED, PAGE_SKIPPED };
- // for check dict page.
- Status _parse_first_page_header();
+ Status _ensure_dictionary_page_loaded();
Status _decode_dict_page();
void _reserve_decompress_buf(size_t size);
@@ -410,6 +410,7 @@ private:
Slice _v2_rep_levels;
Slice _v2_def_levels;
bool _dict_checked = false;
+ bool _first_data_page_parsed = 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 3f06d92dc9a..bffad92baf2 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp
@@ -1168,10 +1168,10 @@ 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());
- right_row = _chunk_reader->page_end_row();
} else {
- right_row = _chunk_reader->page_end_row();
+ 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);
if (read_ranges.count() == 0) {
@@ -1521,6 +1521,14 @@ ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::materialize_dictionary_values(
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Result<MutableColumnPtr> ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::dictionary_values(
const DataTypePtr& target_type) {
+ bool has_dict = false;
+ auto status = _chunk_reader->load_dictionary_page(&has_dict);
+ if (!status.ok()) {
+ return ResultError(std::move(status));
+ }
+ if (!has_dict) {
+ return ResultError(Status::NotSupported("Parquet column has no
reusable dictionary"));
+ }
Decoder* dictionary_decoder = _chunk_reader->dictionary_decoder();
if (dictionary_decoder == nullptr || dictionary_decoder->dictionary_size()
== 0) {
return ResultError(Status::NotSupported("Parquet column has no
reusable dictionary"));
@@ -1536,15 +1544,6 @@ Result<MutableColumnPtr>
ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::dictio
return materialize_dictionary_values(ids.get(), target_type);
}
-template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_try_load_dict_page(bool* loaded,
-
bool* has_dict) {
- // _chunk_reader init will load first page header to check whether has
dict page
- *loaded = true;
- *has_dict = _chunk_reader->has_dict();
- return Status::OK();
-}
-
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::read_column_data(
ColumnPtr& doris_column, const DataTypePtr& type,
@@ -1587,10 +1586,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());
- right_row = _chunk_reader->page_end_row();
} else {
- right_row = _chunk_reader->page_end_row();
+ RETURN_IF_ERROR(_chunk_reader->ensure_first_data_page_parsed());
}
+ right_row = _chunk_reader->page_end_row();
do {
// generate the row ranges that should be read
diff --git
a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
index 76d39c72fb1..b2b37c1bf09 100644
--- a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
+++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
@@ -316,12 +316,10 @@ Status
IcebergPositionDeleteSysTableV2Reader::_init_position_delete_reader() {
static constexpr const char* kPositionReaderProfile =
"IcebergPositionDeleteFileReader";
if (_position_reader_profile == nullptr) {
- _position_reader_profile =
_scanner_profile->get_child(kPositionReaderProfile);
- if (_position_reader_profile == nullptr) {
- // The outer system-table reader calls the inner reader
synchronously. Giving both the
- // same profile would nest identical counter pointers and
double-count every timer.
- _position_reader_profile =
_scanner_profile->create_child(kPositionReaderProfile);
- }
+ // The nested reader needs a distinct profile to avoid double-counting
its timers. Split
+ // readers share the scanner profile and initialize concurrently, so
lookup and creation
+ // must also be atomic to preserve the unique child-name invariant.
+ _position_reader_profile =
_scanner_profile->get_or_create_child(kPositionReaderProfile);
}
_position_reader = std::make_unique<PositionDeleteFileTableReader>();
RETURN_IF_ERROR(_position_reader->init({
diff --git a/be/src/runtime/runtime_profile.cpp
b/be/src/runtime/runtime_profile.cpp
index f48f2935369..36bc79b8215 100644
--- a/be/src/runtime/runtime_profile.cpp
+++ b/be/src/runtime/runtime_profile.cpp
@@ -393,6 +393,23 @@ RuntimeProfile* RuntimeProfile::create_child(const
std::string& name, bool inden
return child;
}
+RuntimeProfile* RuntimeProfile::get_or_create_child(const std::string& name,
bool indent,
+ bool prepend) {
+ std::lock_guard<std::mutex> l(_children_lock);
+ auto it = _child_map.find(name);
+ if (it != _child_map.end()) {
+ return it->second;
+ }
+
+ RuntimeProfile* child = _pool->add(new RuntimeProfile(name));
+ if (this->is_set_metadata()) {
+ child->set_metadata(this->metadata());
+ }
+ auto* location = !_children.empty() && prepend ? _children.front().first :
nullptr;
+ add_child_unlock(child, indent, location);
+ return child;
+}
+
void RuntimeProfile::add_child_unlock(RuntimeProfile* child, bool indent,
RuntimeProfile* loc) {
DCHECK(child != nullptr);
_child_map[child->_name] = child;
diff --git a/be/src/runtime/runtime_profile.h b/be/src/runtime/runtime_profile.h
index c5fe4aeb78d..54f2e80c89e 100644
--- a/be/src/runtime/runtime_profile.h
+++ b/be/src/runtime/runtime_profile.h
@@ -577,6 +577,11 @@ public:
/// otherwise appended after other child profiles.
RuntimeProfile* create_child(const std::string& name, bool indent = true,
bool prepend = false);
+ /// Returns an existing child profile with 'name', or creates it if
absent. Lookup and creation
+ /// are atomic so concurrent callers cannot race while initializing a
shared profile subtree.
+ RuntimeProfile* get_or_create_child(const std::string& name, bool indent =
true,
+ bool prepend = false);
+
// Merges the src profile into this one, combining counters that have an
identical
// path. Info strings from profiles are not merged. 'src' would be a const
if it
// weren't for locking.
diff --git a/be/test/format_v2/parquet/native_decoder_test.cpp
b/be/test/format_v2/parquet/native_decoder_test.cpp
index 3dfad7f4abd..b71e94222a3 100644
--- a/be/test/format_v2/parquet/native_decoder_test.cpp
+++ b/be/test/format_v2/parquet/native_decoder_test.cpp
@@ -387,6 +387,7 @@ public:
Status read_bytes(const uint8_t** buf, uint64_t offset, size_t
bytes_to_read,
const io::IOContext*) override {
+ ++_read_count;
if (offset > _data.size() || bytes_to_read > _data.size() - offset) {
return Status::IOError("out of bounds");
}
@@ -394,6 +395,7 @@ public:
return Status::OK();
}
Status read_bytes(Slice& slice, uint64_t offset, const io::IOContext*)
override {
+ ++_read_count;
if (offset > _data.size() || slice.size > _data.size() - offset) {
return Status::IOError("out of bounds");
}
@@ -402,9 +404,11 @@ public:
}
std::string path() override { return "memory.parquet"; }
int64_t mtime() const override { return 0; }
+ size_t read_count() const { return _read_count; }
private:
std::vector<uint8_t> _data;
+ size_t _read_count = 0;
};
class NativeDecoderMemoryFileReader final : public io::FileReader {
@@ -420,10 +424,12 @@ public:
size_t size() const override { return _data.size(); }
bool closed() const override { return _closed; }
int64_t mtime() const override { return 1; }
+ size_t read_count() const { return _read_count; }
protected:
Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
const io::IOContext*) override {
+ ++_read_count;
if (offset > _data.size() || result.size > _data.size() - offset) {
return Status::IOError("native decoder memory read exceeds file");
}
@@ -436,6 +442,7 @@ private:
std::vector<uint8_t> _data;
io::Path _path;
bool _closed = false;
+ size_t _read_count = 0;
};
std::shared_ptr<::parquet::ColumnDescriptor> descriptor(::parquet::Type::type
physical_type) {
@@ -487,6 +494,53 @@ std::vector<uint8_t> serialize_page(tparquet::PageHeader
header,
return bytes;
}
+std::vector<uint8_t> serialize_plain_int32_page(const std::vector<int32_t>&
values) {
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ header.__set_compressed_page_size(values.size() * sizeof(int32_t));
+ header.__set_uncompressed_page_size(values.size() * sizeof(int32_t));
+ header.__isset.data_page_header = true;
+ header.data_page_header.__set_num_values(values.size());
+ header.data_page_header.__set_encoding(tparquet::Encoding::PLAIN);
+
header.data_page_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+
header.data_page_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+ const auto* value_bytes = reinterpret_cast<const uint8_t*>(values.data());
+ return serialize_page(
+ header, std::vector<uint8_t>(value_bytes, value_bytes +
header.compressed_page_size));
+}
+
+TEST(ParquetV2NativeDecoderTest, ColumnChunkInitDoesNotReadFirstDataPage) {
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ header.__set_compressed_page_size(sizeof(int32_t));
+ header.__set_uncompressed_page_size(sizeof(int32_t));
+ header.__isset.data_page_header = true;
+ header.data_page_header.__set_num_values(1);
+ header.data_page_header.__set_encoding(tparquet::Encoding::PLAIN);
+
header.data_page_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+
header.data_page_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+ auto bytes = serialize_page(header, std::vector<uint8_t>(sizeof(int32_t)));
+ MemoryBufferedReader stream(bytes);
+
+ 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(1);
+ 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.parquet_schema.__set_type(tparquet::Type::INT32);
+
field.parquet_schema.__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ ParquetPageReadContext context(false, "");
+ ColumnChunkReader<false, false> reader(&stream, &chunk, &field, nullptr,
1, nullptr, context);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(stream.read_count(), 0);
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_GT(stream.read_count(), 0);
+}
+
Status materialize_level_only_page(bool data_page_v2, tparquet::Type::type
physical_type,
tparquet::Encoding::type encoding, bool
all_null) {
constexpr size_t VALUE_COUNT = 3;
@@ -541,6 +595,7 @@ Status materialize_level_only_page(bool data_page_v2,
tparquet::Type::type physi
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, VALUE_COUNT,
nullptr, page_context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
const auto load_status = chunk_reader.load_page_data();
EXPECT_TRUE(load_status.ok()) << load_status;
RETURN_IF_ERROR(load_status);
@@ -617,6 +672,7 @@ Status load_scripted_page(tparquet::PageHeader header,
const std::vector<uint8_t
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, 1, nullptr,
context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
return chunk_reader.load_page_data();
}
@@ -639,6 +695,7 @@ Status load_malformed_nested_page(tparquet::PageHeader
header, const std::vector
ColumnChunkReader<true, false> chunk_reader(&reader, &chunk, &field,
nullptr, 1, nullptr,
context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
RETURN_IF_ERROR(chunk_reader.load_page_data());
std::vector<level_t> rep_levels;
size_t result_rows = 0;
@@ -674,6 +731,7 @@ Status materialize_plain_int96(const
std::vector<ParquetInt96Timestamp>& values,
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, values.size(),
nullptr, page_context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
RETURN_IF_ERROR(chunk_reader.load_page_data());
DataTypeDateTimeV2 type(6);
@@ -730,6 +788,7 @@ Status materialize_selected_plain_fixed(
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, logical_values,
nullptr, page_context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
RETURN_IF_ERROR(chunk_reader.load_page_data());
ParquetDecodeContext decode_context;
@@ -844,6 +903,7 @@ Status materialize_selected_dictionary_fixed(
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, logical_values,
nullptr, page_context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
RETURN_IF_ERROR(chunk_reader.load_page_data());
ParquetDecodeContext decode_context;
@@ -946,6 +1006,7 @@ Status materialize_selected_dictionary_strings(const
std::vector<std::string>& d
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, logical_values,
nullptr, page_context);
RETURN_IF_ERROR(chunk_reader.init());
+ RETURN_IF_ERROR(chunk_reader.parse_page_header());
RETURN_IF_ERROR(chunk_reader.load_page_data());
DataTypeString type;
@@ -993,6 +1054,7 @@ TEST(ParquetV2NativeDecoderTest,
RawExprMapsNullableSparseRowsDirectly) {
ColumnChunkReader<false, false> chunk_reader(&reader, &chunk, &field,
nullptr, LOGICAL_VALUES,
nullptr, page_context);
ASSERT_TRUE(chunk_reader.init().ok());
+ ASSERT_TRUE(chunk_reader.parse_page_header().ok());
ASSERT_TRUE(chunk_reader.load_page_data().ok());
const std::vector<uint16_t> null_runs {1, 1, 2, 1, 2};
@@ -1341,8 +1403,10 @@ TEST(ParquetV2NativeDecoderTest,
DictionaryProbeMaterializesTypedValuesOnlyOnce)
ScalarColumnReader<false, false> reader(row_ranges, 2, chunk, nullptr,
nullptr, nullptr);
ASSERT_TRUE(reader.init(file, &field, bytes.size(), nullptr, "",
ParquetReaderCompat {}, true)
.ok());
+ EXPECT_EQ(file->read_count(), 0);
auto dictionary_result = reader.dictionary_values(field.data_type);
ASSERT_TRUE(dictionary_result.has_value()) << dictionary_result.error();
+ EXPECT_GT(file->read_count(), 0);
EXPECT_EQ(reader.dictionary_materialization_count_for_test(), 1);
FilterMap filter;
@@ -1371,6 +1435,83 @@ TEST(ParquetV2NativeDecoderTest,
DictionaryProbeMaterializesTypedValuesOnlyOnce)
EXPECT_EQ(reader.dictionary_materialization_count_for_test(), 1);
}
+TEST(ParquetV2NativeDecoderTest,
IndexedSkipLoadsDictionaryBeforeJumpingToDataPage) {
+ const std::array<int32_t, 2> dictionary {10, 20};
+ std::vector<uint8_t> dictionary_payload(sizeof(dictionary));
+ memcpy(dictionary_payload.data(), dictionary.data(),
dictionary_payload.size());
+ tparquet::PageHeader dictionary_header;
+ dictionary_header.type = tparquet::PageType::DICTIONARY_PAGE;
+ dictionary_header.__set_compressed_page_size(dictionary_payload.size());
+ dictionary_header.__set_uncompressed_page_size(dictionary_payload.size());
+ dictionary_header.__isset.dictionary_page_header = true;
+
dictionary_header.dictionary_page_header.__set_num_values(dictionary.size());
+
dictionary_header.dictionary_page_header.__set_encoding(tparquet::Encoding::PLAIN);
+
+ std::vector<uint8_t> bytes(1, 0);
+ const auto dictionary_page = serialize_page(dictionary_header,
dictionary_payload);
+ bytes.insert(bytes.end(), dictionary_page.begin(), dictionary_page.end());
+
+ auto make_data_page = [] {
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ header.__set_compressed_page_size(1);
+ header.__set_uncompressed_page_size(1);
+ header.__isset.data_page_header = true;
+ header.data_page_header.__set_num_values(1);
+
header.data_page_header.__set_encoding(tparquet::Encoding::RLE_DICTIONARY);
+
header.data_page_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+
header.data_page_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+ return serialize_page(header, {0});
+ };
+ const size_t first_data_offset = bytes.size();
+ const auto first_data_page = make_data_page();
+ bytes.insert(bytes.end(), first_data_page.begin(), first_data_page.end());
+ const size_t second_data_offset = bytes.size();
+ const auto second_data_page = make_data_page();
+ bytes.insert(bytes.end(), second_data_page.begin(),
second_data_page.end());
+
+ tparquet::OffsetIndex offset_index;
+ tparquet::PageLocation first_location;
+ first_location.__set_offset(first_data_offset);
+ first_location.__set_compressed_page_size(first_data_page.size());
+ first_location.__set_first_row_index(0);
+ tparquet::PageLocation second_location;
+ second_location.__set_offset(second_data_offset);
+ second_location.__set_compressed_page_size(second_data_page.size());
+ second_location.__set_first_row_index(1);
+ offset_index.__set_page_locations({first_location, second_location});
+
+ 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(2);
+ chunk.meta_data.__set_total_compressed_size(bytes.size() - 1);
+ chunk.meta_data.__set_dictionary_page_offset(1);
+ chunk.meta_data.__set_data_page_offset(first_data_offset);
+ NativeFieldSchema field;
+ field.physical_type = tparquet::Type::INT32;
+ field.parquet_schema.__set_type(tparquet::Type::INT32);
+
field.parquet_schema.__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ MemoryBufferedReader stream(bytes);
+ ParquetPageReadContext context(false, "");
+ ColumnChunkReader<false, true> reader(&stream, &chunk, &field,
&offset_index, 2, nullptr,
+ context);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(stream.read_count(), 0);
+ ASSERT_TRUE(reader.next_page().ok());
+ EXPECT_GT(stream.read_count(), 0);
+ bool has_dict = false;
+ const size_t reads_after_seek = stream.read_count();
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_TRUE(has_dict);
+ EXPECT_EQ(stream.read_count(), reads_after_seek);
+ ASSERT_NE(reader.dictionary_decoder(), nullptr);
+ EXPECT_EQ(reader.dictionary_decoder()->dictionary_size(),
dictionary.size());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_EQ(reader.page_start_row(), 1);
+}
+
TEST(ParquetV2NativeDecoderTest,
DictionaryRepeatedRunsGatherDirectlyIntoDestination) {
for (const auto encoding :
{tparquet::Encoding::RLE_DICTIONARY,
tparquet::Encoding::PLAIN_DICTIONARY}) {
@@ -2860,6 +3001,7 @@ TEST(ParquetV2NativeDecoderTest,
DecompressionScratchStaysActiveUntilPageExhaust
static_cast<size_t>(LARGE_VALUE_COUNT) + 1, nullptr,
context);
ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
ASSERT_TRUE(reader.load_page_data().ok());
ASSERT_GT(reader.active_decoder_scratch_bytes(), 1UL << 20);
ASSERT_TRUE(reader.skip_values(LARGE_VALUE_COUNT).ok());
@@ -3083,6 +3225,218 @@ TEST(ParquetV2NativeDecoderTest,
ShiftedOffsetIndexFallsBackToSequentialPages) {
verify_fallback(true);
}
+TEST(ParquetV2NativeDecoderTest,
LazyFlatIndexedSkipValidatesFirstPageCardinality) {
+ const auto first_page = serialize_plain_int32_page({10});
+ const auto second_page = serialize_plain_int32_page({20, 21});
+ const auto third_page = serialize_plain_int32_page({30});
+ const auto fourth_page = serialize_plain_int32_page({40, 41});
+ std::vector<uint8_t> bytes = first_page;
+ const size_t second_offset = bytes.size();
+ bytes.insert(bytes.end(), second_page.begin(), second_page.end());
+ const size_t third_offset = bytes.size();
+ bytes.insert(bytes.end(), third_page.begin(), third_page.end());
+ const size_t fourth_offset = bytes.size();
+ bytes.insert(bytes.end(), fourth_page.begin(), fourth_page.end());
+
+ tparquet::OffsetIndex offset_index;
+ std::vector<tparquet::PageLocation> locations(4);
+ locations[0].__set_offset(0);
+ locations[0].__set_compressed_page_size(first_page.size());
+ locations[0].__set_first_row_index(0);
+ locations[1].__set_offset(second_offset);
+ locations[1].__set_compressed_page_size(second_page.size());
+ locations[1].__set_first_row_index(2);
+ locations[2].__set_offset(third_offset);
+ locations[2].__set_compressed_page_size(third_page.size());
+ locations[2].__set_first_row_index(4);
+ locations[3].__set_offset(fourth_offset);
+ locations[3].__set_compressed_page_size(fourth_page.size());
+ locations[3].__set_first_row_index(5);
+ offset_index.__set_page_locations(locations);
+
+ 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);
+ 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 file = std::make_shared<NativeDecoderMemoryFileReader>(bytes);
+ const auto row_ranges = ::doris::RowRanges::create_single(2, 4);
+ ScalarColumnReader<false, true> reader(row_ranges, 6, chunk,
&offset_index, nullptr, nullptr);
+ ASSERT_TRUE(reader.init(file, &field, bytes.size(), nullptr, "",
ParquetReaderCompat {}, true)
+ .ok());
+
+ FilterMap filter;
+ ASSERT_TRUE(filter.init(nullptr, 2, false).ok());
+ ColumnPtr values = ColumnInt32::create();
+ size_t rows = 0;
+ bool eof = false;
+ const auto status = reader.read_column_data(values, field.data_type,
nullptr, filter, 2, &rows,
+ &eof, false);
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+}
+
+TEST(ParquetV2NativeDecoderTest,
LazyFixedWidthFilterUsesReconciledFirstPageRange) {
+ const auto first_page = serialize_plain_int32_page({10, 11});
+ const auto second_page = serialize_plain_int32_page({20});
+ std::vector<uint8_t> bytes = first_page;
+ bytes.insert(bytes.end(), second_page.begin(), second_page.end());
+ bytes.push_back(0);
+
+ tparquet::OffsetIndex offset_index;
+ tparquet::PageLocation first_location;
+ first_location.__set_offset(0);
+ first_location.__set_compressed_page_size(first_page.size() + 1);
+ first_location.__set_first_row_index(0);
+ tparquet::PageLocation second_location;
+ second_location.__set_offset(first_page.size() + 1);
+ second_location.__set_compressed_page_size(second_page.size());
+ second_location.__set_first_row_index(1);
+ offset_index.__set_page_locations({first_location, second_location});
+
+ 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);
+ 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 file = std::make_shared<NativeDecoderMemoryFileReader>(bytes);
+ const auto row_ranges = ::doris::RowRanges::create_single(0, 2);
+ ScalarColumnReader<false, true> reader(row_ranges, 3, chunk,
&offset_index, nullptr, nullptr);
+ ASSERT_TRUE(reader.init(file, &field, bytes.size(), nullptr, "",
ParquetReaderCompat {}, true)
+ .ok());
+
+ FilterMap filter;
+ ASSERT_TRUE(filter.init(nullptr, 2, false).ok());
+ IColumn::Filter row_filter;
+ size_t rows = 0;
+ bool eof = false;
+ bool used_filter = false;
+ const auto status = reader.read_fixed_width_filter(
+ {create_int32_raw_comparison(0, "ge", TExprOpcode::GE, 0)}, 0,
filter, 2, nullptr,
+ &row_filter, &rows, &eof, &used_filter);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_TRUE(used_filter);
+ EXPECT_EQ(rows, 2);
+ EXPECT_EQ(row_filter, (IColumn::Filter {1, 1}));
+}
+
+TEST(ParquetV2NativeDecoderTest,
LazyFlatIndexedFallbackUsesReconciledFirstPageRange) {
+ const auto first_page = serialize_plain_int32_page({10, 11});
+ const auto second_page = serialize_plain_int32_page({20});
+ std::vector<uint8_t> bytes = first_page;
+ bytes.insert(bytes.end(), second_page.begin(), second_page.end());
+ bytes.push_back(0);
+
+ tparquet::OffsetIndex offset_index;
+ tparquet::PageLocation first_location;
+ first_location.__set_offset(0);
+ first_location.__set_compressed_page_size(first_page.size() + 1);
+ first_location.__set_first_row_index(0);
+ tparquet::PageLocation second_location;
+ second_location.__set_offset(first_page.size() + 1);
+ second_location.__set_compressed_page_size(second_page.size());
+ second_location.__set_first_row_index(1);
+ offset_index.__set_page_locations({first_location, second_location});
+
+ 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.data_type = std::make_shared<DataTypeInt32>();
+ 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);
+ ScalarColumnReader<false, true> reader(row_ranges, 3, chunk,
&offset_index, nullptr, nullptr);
+ ASSERT_TRUE(reader.init(file, &field, bytes.size(), nullptr, "",
ParquetReaderCompat {}, true)
+ .ok());
+
+ FilterMap filter;
+ ASSERT_TRUE(filter.init(nullptr, 2, false).ok());
+ ColumnPtr values = ColumnInt32::create();
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(
+ reader.read_column_data(values, field.data_type, nullptr, filter,
2, &rows, &eof, false)
+ .ok());
+ ASSERT_EQ(rows, 2);
+ EXPECT_EQ(assert_cast<const ColumnInt32&>(*values).get_data(),
+ (ColumnInt32::Container {10, 11}));
+}
+
+TEST(ParquetV2NativeDecoderTest, LazyNestedV2SeekValidatesFirstPageRowRange) {
+ auto make_page = [](int32_t value) {
+ const std::vector<uint8_t> repetition_levels {2, 0};
+ std::vector<uint8_t> payload = repetition_levels;
+ const auto* value_bytes = reinterpret_cast<const uint8_t*>(&value);
+ payload.insert(payload.end(), value_bytes, value_bytes +
sizeof(value));
+ tparquet::PageHeader header;
+ header.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;
+ header.data_page_header_v2.__set_num_values(1);
+ header.data_page_header_v2.__set_num_rows(1);
+ header.data_page_header_v2.__set_num_nulls(0);
+ header.data_page_header_v2.__set_encoding(tparquet::Encoding::PLAIN);
+
header.data_page_header_v2.__set_repetition_levels_byte_length(repetition_levels.size());
+ header.data_page_header_v2.__set_definition_levels_byte_length(0);
+ header.data_page_header_v2.__set_is_compressed(false);
+ return serialize_page(header, payload);
+ };
+ const auto first_page = make_page(10);
+ const auto second_page = make_page(20);
+ std::vector<uint8_t> bytes = first_page;
+ const size_t second_offset = bytes.size();
+ bytes.insert(bytes.end(), second_page.begin(), second_page.end());
+
+ tparquet::OffsetIndex offset_index;
+ tparquet::PageLocation first_location;
+ first_location.__set_offset(0);
+ first_location.__set_compressed_page_size(first_page.size());
+ first_location.__set_first_row_index(0);
+ tparquet::PageLocation second_location;
+ second_location.__set_offset(second_offset);
+ second_location.__set_compressed_page_size(second_page.size());
+ second_location.__set_first_row_index(2);
+ offset_index.__set_page_locations({first_location, second_location});
+
+ 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(2);
+ 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;
+ ParquetPageReadContext context(false, "");
+ MemoryBufferedReader stream(bytes);
+ ColumnChunkReader<true, true> reader(&stream, &chunk, &field,
&offset_index, 3, nullptr,
+ context);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(stream.read_count(), 0);
+ const auto status = reader.seek_to_nested_row(2);
+ EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+}
+
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);
@@ -3108,11 +3462,13 @@ TEST(ParquetV2NativeDecoderTest,
FlatPagesRejectLogicalAndPhysicalCardinalityMis
if (with_offset_index) {
ColumnChunkReader<false, true> reader_with_index(&reader, &chunk,
&field, &offset_index,
1, nullptr,
context);
- return reader_with_index.init();
+ RETURN_IF_ERROR(reader_with_index.init());
+ return reader_with_index.parse_page_header();
}
ColumnChunkReader<false, false> sequential_reader(&reader, &chunk,
&field, nullptr, 1,
nullptr, context);
- return sequential_reader.init();
+ RETURN_IF_ERROR(sequential_reader.init());
+ return sequential_reader.parse_page_header();
};
tparquet::PageHeader v2;
@@ -3181,7 +3537,8 @@ TEST(ParquetV2NativeDecoderTest,
NestedV2PageRejectsOffsetIndexRowSpanMismatch)
ColumnChunkReader<true, true> reader(&stream, &chunk, &field,
&offset_index,
/*total_rows=*/2, nullptr, context);
- const auto status = reader.init();
+ ASSERT_TRUE(reader.init().ok());
+ const auto status = reader.parse_page_header();
EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
}
@@ -3390,6 +3747,7 @@ TEST(ParquetV2NativeDecoderTest,
NestedV1ContinuationRemainsValidAfterFirstRowSt
ColumnChunkReader<true, false> chunk_reader(&reader, &chunk, &field,
nullptr, 1, nullptr,
context);
ASSERT_TRUE(chunk_reader.init().ok());
+ ASSERT_TRUE(chunk_reader.parse_page_header().ok());
ASSERT_TRUE(chunk_reader.load_page_data().ok());
std::vector<level_t> levels;
size_t rows = 0;
@@ -3445,6 +3803,7 @@ TEST(ParquetV2NativeDecoderTest,
NestedV1IgnoresUnverifiableOffsetIndexRows) {
ColumnChunkReader<true, true> chunk_reader(&stream, &chunk, &field,
&offset_index, 1, nullptr,
context);
ASSERT_TRUE(chunk_reader.init().ok());
+ ASSERT_TRUE(chunk_reader.parse_page_header().ok());
ASSERT_TRUE(chunk_reader.load_page_data().ok());
std::vector<level_t> levels;
size_t rows = 0;
@@ -3456,6 +3815,141 @@ TEST(ParquetV2NativeDecoderTest,
NestedV1IgnoresUnverifiableOffsetIndexRows) {
EXPECT_EQ(levels, std::vector<level_t>({0, 1, 1}));
}
+TEST(ParquetV2NativeDecoderTest, LazyNestedV1SeekDoesNotOutrunPhysicalPages) {
+ auto make_page = [] {
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ const std::vector<uint8_t> payload {2, 0, 0, 0, 2, 0, 0, 0, 0, 0};
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ header.__isset.data_page_header = true;
+ header.data_page_header.__set_num_values(1);
+ header.data_page_header.__set_encoding(tparquet::Encoding::PLAIN);
+
header.data_page_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+
header.data_page_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+ return serialize_page(header, payload);
+ };
+ const auto first_page = make_page();
+ const auto second_page = make_page();
+ std::vector<uint8_t> bytes = first_page;
+ bytes.insert(bytes.end(), second_page.begin(), second_page.end());
+
+ MemoryBufferedReader stream(bytes);
+ 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(2);
+ 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;
+ tparquet::OffsetIndex offset_index;
+ tparquet::PageLocation first_location;
+ first_location.__set_offset(0);
+ first_location.__set_compressed_page_size(first_page.size());
+ first_location.__set_first_row_index(0);
+ tparquet::PageLocation second_location;
+ second_location.__set_offset(first_page.size());
+ second_location.__set_compressed_page_size(second_page.size());
+ second_location.__set_first_row_index(1);
+ offset_index.__set_page_locations({first_location, second_location});
+ ParquetPageReadContext context(false, "");
+ ColumnChunkReader<true, true> chunk_reader(&stream, &chunk, &field,
&offset_index, 2, nullptr,
+ context);
+
+ ASSERT_TRUE(chunk_reader.init().ok());
+ EXPECT_EQ(stream.read_count(), 0);
+ ASSERT_TRUE(chunk_reader.seek_to_nested_row(1).ok());
+ std::vector<level_t> levels;
+ size_t rows = 0;
+ bool cross_page = false;
+ ASSERT_TRUE(chunk_reader.load_page_nested_rows(levels, 1, &rows,
&cross_page).ok());
+ if (cross_page) {
+ const auto status = chunk_reader.load_cross_page_nested_row(levels,
&cross_page);
+ ASSERT_TRUE(status.ok()) << status;
+ }
+ EXPECT_EQ(rows, 1);
+ EXPECT_FALSE(cross_page);
+}
+
+TEST(ParquetV2NativeDecoderTest,
LazyDictionaryNestedV1SeekChecksFirstDataPage) {
+ tparquet::PageHeader dictionary_header;
+ dictionary_header.type = tparquet::PageType::DICTIONARY_PAGE;
+ dictionary_header.__set_compressed_page_size(sizeof(int32_t));
+ dictionary_header.__set_uncompressed_page_size(sizeof(int32_t));
+ dictionary_header.__isset.dictionary_page_header = true;
+ dictionary_header.dictionary_page_header.__set_num_values(1);
+
dictionary_header.dictionary_page_header.__set_encoding(tparquet::Encoding::PLAIN);
+ const int32_t dictionary_value = 7;
+ const auto* dictionary_bytes = reinterpret_cast<const
uint8_t*>(&dictionary_value);
+ auto bytes = serialize_page(
+ dictionary_header,
+ std::vector<uint8_t>(dictionary_bytes, dictionary_bytes +
sizeof(dictionary_value)));
+
+ auto make_data_page = [] {
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ const std::vector<uint8_t> payload {2, 0, 0, 0, 2, 0, 0, 0, 0, 0};
+ header.__set_compressed_page_size(payload.size());
+ header.__set_uncompressed_page_size(payload.size());
+ header.__isset.data_page_header = true;
+ header.data_page_header.__set_num_values(1);
+ header.data_page_header.__set_encoding(tparquet::Encoding::PLAIN);
+
header.data_page_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+
header.data_page_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+ return serialize_page(header, payload);
+ };
+ const auto first_page = make_data_page();
+ const size_t first_page_offset = bytes.size();
+ bytes.insert(bytes.end(), first_page.begin(), first_page.end());
+ const auto second_page = make_data_page();
+ const size_t second_page_offset = bytes.size();
+ bytes.insert(bytes.end(), second_page.begin(), second_page.end());
+ const size_t chunk_size = bytes.size();
+ bytes.resize(chunk_size + 16, 0);
+
+ MemoryBufferedReader stream(bytes);
+ 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(2);
+ chunk.meta_data.__set_total_compressed_size(chunk_size);
+ chunk.meta_data.__set_dictionary_page_offset(0);
+ chunk.meta_data.__set_data_page_offset(first_page_offset);
+ NativeFieldSchema field;
+ field.physical_type = tparquet::Type::INT32;
+ field.repetition_level = 1;
+ tparquet::OffsetIndex offset_index;
+ tparquet::PageLocation first_location;
+ first_location.__set_offset(first_page_offset);
+ first_location.__set_compressed_page_size(first_page.size());
+ first_location.__set_first_row_index(0);
+ tparquet::PageLocation second_location;
+ second_location.__set_offset(second_page_offset);
+ second_location.__set_compressed_page_size(second_page.size());
+ second_location.__set_first_row_index(1);
+ offset_index.__set_page_locations({first_location, second_location});
+ ColumnChunkRange padded_range {.offset = 0, .length = bytes.size()};
+ ParquetPageReadContext context(false, "");
+ ColumnChunkReader<true, true> chunk_reader(&stream, &chunk, &field,
&offset_index, 2, nullptr,
+ context, &padded_range);
+
+ ASSERT_TRUE(chunk_reader.init().ok());
+ EXPECT_EQ(stream.read_count(), 0);
+ ASSERT_TRUE(chunk_reader.seek_to_nested_row(1).ok());
+ std::vector<level_t> levels;
+ size_t rows = 0;
+ bool cross_page = false;
+ ASSERT_TRUE(chunk_reader.load_page_nested_rows(levels, 1, &rows,
&cross_page).ok());
+ if (cross_page) {
+ const auto status = chunk_reader.load_cross_page_nested_row(levels,
&cross_page);
+ ASSERT_TRUE(status.ok()) << status;
+ }
+ EXPECT_EQ(rows, 1);
+ EXPECT_FALSE(cross_page);
+}
+
TEST(ParquetV2NativeDecoderTest,
NestedV1DiscardedOffsetIndexStopsAtLogicalChunkEnd) {
tparquet::PageHeader header;
header.type = tparquet::PageType::DATA_PAGE;
@@ -3493,6 +3987,7 @@ TEST(ParquetV2NativeDecoderTest,
NestedV1DiscardedOffsetIndexStopsAtLogicalChunk
ColumnChunkReader<true, true> chunk_reader(&stream, &chunk, &field,
&offset_index, 1, nullptr,
context, &padded_range);
ASSERT_TRUE(chunk_reader.init().ok());
+ ASSERT_TRUE(chunk_reader.parse_page_header().ok());
ASSERT_TRUE(chunk_reader.load_page_data().ok());
std::vector<level_t> levels;
size_t rows = 0;
@@ -3650,6 +4145,7 @@ TEST(ParquetV2NativeDecoderTest,
OptionalV2FixedWidthPageRejectsExtentBeforeAllo
ColumnChunkReader<false, false> reader(&stream, &chunk, &field,
nullptr, 1, nullptr,
context);
ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
EXPECT_TRUE(reader.load_page_data().is<ErrorCode::CORRUPTION>());
EXPECT_LT(reader.retained_decoder_scratch_bytes(), 64UL << 10);
}
@@ -3695,6 +4191,7 @@ TEST(ParquetV2NativeDecoderTest,
RepeatedV2FixedWidthPageRejectsExtentBeforeAllo
ParquetPageReadContext context(false, "");
ColumnChunkReader<true, false> reader(&stream, &chunk, &field, nullptr, 1,
nullptr, context);
ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
EXPECT_TRUE(reader.load_page_data().is<ErrorCode::CORRUPTION>());
EXPECT_LT(reader.retained_decoder_scratch_bytes(), 64UL << 10);
}
@@ -3733,6 +4230,7 @@ TEST(ParquetV2NativeDecoderTest,
VariableWidthDataPagePreflightsCompressedExtent
ParquetPageReadContext context(false, "");
ColumnChunkReader<false, false> reader(&stream, &chunk, &field, nullptr,
1, nullptr, context);
ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
EXPECT_TRUE(reader.load_page_data().is<ErrorCode::CORRUPTION>());
EXPECT_LT(reader.retained_decoder_scratch_bytes(), 64UL << 10);
EXPECT_TRUE(load_scripted_page(header, payload,
tparquet::CompressionCodec::SNAPPY, true,
@@ -3933,6 +4431,7 @@ TEST(ParquetV2NativeDecoderTest,
ColumnChunkSkipsIndexPageBeforeInitializingData
context);
const auto init_status = chunk_reader.init();
ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(chunk_reader.parse_page_header().ok());
EXPECT_EQ(chunk_reader.remaining_num_values(), 1);
ASSERT_TRUE(chunk_reader.load_page_data().ok());
}
@@ -3974,6 +4473,7 @@ TEST(ParquetV2NativeDecoderTest,
ColumnChunkSkipsUnknownAuxiliaryPage) {
context);
const auto init_status = chunk_reader.init();
ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(chunk_reader.parse_page_header().ok());
EXPECT_EQ(chunk_reader.remaining_num_values(), 1);
}
diff --git a/be/test/runtime/runtime_profile_test.cpp
b/be/test/runtime/runtime_profile_test.cpp
index 2734084e9cb..e39658d0b48 100644
--- a/be/test/runtime/runtime_profile_test.cpp
+++ b/be/test/runtime/runtime_profile_test.cpp
@@ -19,8 +19,11 @@
#include <gtest/gtest.h>
+#include <barrier>
#include <cstdlib>
#include <sstream>
+#include <thread>
+#include <vector>
#include "common/exception.h"
#include "common/object_pool.h"
@@ -578,4 +581,32 @@ TEST(RuntimeProfileTest, TestGetChild) {
ASSERT_EQ(child2, root.get_child("Child2"));
}
+TEST(RuntimeProfileTest, ConcurrentGetOrCreateChildReturnsSingleSharedProfile)
{
+ RuntimeProfile root("Root");
+ constexpr size_t kThreadCount = 32;
+ std::barrier start(kThreadCount);
+ std::vector<RuntimeProfile*> children(kThreadCount);
+ std::vector<std::thread> threads;
+ threads.reserve(kThreadCount);
+
+ for (size_t i = 0; i < kThreadCount; ++i) {
+ threads.emplace_back([&, i] {
+ start.arrive_and_wait();
+ children[i] = root.get_or_create_child("SharedChild");
+ });
+ }
+ for (auto& thread : threads) {
+ thread.join();
+ }
+
+ ASSERT_NE(children.front(), nullptr);
+ for (const auto* child : children) {
+ EXPECT_EQ(child, children.front());
+ }
+ std::vector<RuntimeProfile*> profile_children;
+ root.get_children(&profile_children);
+ ASSERT_EQ(profile_children.size(), 1);
+ EXPECT_EQ(profile_children.front(), children.front());
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
