This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 87124245bd8 [improvement](parquet) Load dictionary pages lazily
(#66000)
87124245bd8 is described below
commit 87124245bd81b28f06a166d642c30b610f064ca8
Author: foxtail463 <[email protected]>
AuthorDate: Wed Jul 29 10:53:20 2026 +0800
[improvement](parquet) Load dictionary pages lazily (#66000)
Problem Summary:
Parquet column readers eagerly parsed the first page during
initialization, causing unnecessary page-header reads and dictionary
decoding for columns later eliminated by lazy filtering.
Solution:
Defer dictionary probing until dictionary values or page data are
actually needed, while preserving correct sequential and offset-index
page advancement.
Co-authored-by: yangtao555 <[email protected]>
---
.../parquet/vparquet_column_chunk_reader.cpp | 31 +-
.../format/parquet/vparquet_column_chunk_reader.h | 19 +-
be/src/format/parquet/vparquet_column_reader.cpp | 14 +-
be/src/format/parquet/vparquet_column_reader.h | 1 -
.../parquet/parquet_column_chunk_reader_test.cpp | 488 +++++++++++++++++++++
be/test/format/parquet/parquet_page_cache_test.cpp | 41 +-
be/test/format/parquet/parquet_reader_test.cpp | 21 +
7 files changed, 568 insertions(+), 47 deletions(-)
diff --git a/be/src/format/parquet/vparquet_column_chunk_reader.cpp
b/be/src/format/parquet/vparquet_column_chunk_reader.cpp
index b4b919f1870..ca9d290943a 100644
--- a/be/src/format/parquet/vparquet_column_chunk_reader.cpp
+++ b/be/src/format/parquet/vparquet_column_chunk_reader.cpp
@@ -25,6 +25,7 @@
#include <memory>
#include <utility>
+#include "common/check.h"
#include "common/compiler_util.h" // IWYU pragma: keep
#include "core/column/column.h"
#include "core/custom_allocator.h"
@@ -76,7 +77,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();
}
@@ -103,20 +103,28 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::skip_nested_values(
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_parse_first_page_header() {
- RETURN_IF_ERROR(parse_page_header());
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_ensure_dictionary_page_loaded() {
+ if (_dict_checked) {
+ return Status::OK();
+ }
+ DORIS_CHECK(_state == INITIALIZED);
+ 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::DICTIONARY_PAGE) {
- // the first page maybe directory page even if
_metadata.__isset.dictionary_page_offset == false,
- // so we should parse the directory page in next_page()
RETURN_IF_ERROR(_decode_dict_page());
- // parse the real first data page
RETURN_IF_ERROR(_page_reader->dict_next_page());
- _state = INITIALIZED;
}
+ _dict_checked = true;
+ return Status::OK();
+}
+
+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();
}
@@ -125,6 +133,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());
RETURN_IF_ERROR(_page_reader->parse_page_header());
const tparquet::PageHeader* header = nullptr;
@@ -145,8 +154,14 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::parse_page_header() {
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() {
- _state = INITIALIZED;
+ if constexpr (OFFSET_INDEX) {
+ RETURN_IF_ERROR(_ensure_dictionary_page_loaded());
+ } else {
+ // Sequential readers need the current header to locate the next page.
+ DORIS_CHECK(_state == HEADER_PARSED || _state == DATA_LOADED);
+ }
RETURN_IF_ERROR(_page_reader->next_page());
+ _state = INITIALIZED;
return Status::OK();
}
diff --git a/be/src/format/parquet/vparquet_column_chunk_reader.h
b/be/src/format/parquet/vparquet_column_chunk_reader.h
index b117f6c6652..064d28fc115 100644
--- a/be/src/format/parquet/vparquet_column_chunk_reader.h
+++ b/be/src/format/parquet/vparquet_column_chunk_reader.h
@@ -77,17 +77,19 @@ struct ColumnChunkReaderStatistics {
* ColumnChunkReader chunk_reader(BufferedStreamReader* reader,
* tparquet::ColumnChunk* column_chunk,
* FieldSchema* fieldSchema);
- * // Initialize chunk reader
+ * // Initialize chunk reader without reading any page
* chunk_reader.init();
- * while (chunk_reader.has_next_page()) {
- * // Seek to next page header. Only read and parse the page header, not
page data.
- * chunk_reader.next_page();
+ * while (true) {
+ * // Read and parse the current page header, but not page data.
+ * chunk_reader.parse_page_header();
* // Load data to decoder. Load the page data into underlying container.
* // Or, we can call the chunk_reader.skip_page() to skip current page.
* chunk_reader.load_page_data();
* // Decode values into column or slice.
* // Or, we can call chunk_reader.skip_values(num_values) to skip some
values.
* chunk_reader.decode_values(slice, num_values);
+ * if (!chunk_reader.has_next_page()) break;
+ * chunk_reader.next_page();
* }
*/
template <bool IN_COLLECTION, bool OFFSET_INDEX>
@@ -99,7 +101,7 @@ public:
const ParquetPageReadContext& page_read_ctx);
~ColumnChunkReader() = default;
- // Initialize chunk reader, will generate the decoder and codec.
+ // Initialize the page reader and compression codec without reading any
page.
Status init();
// Whether the chunk reader has a more page to read.
@@ -141,7 +143,9 @@ 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; };
+ // Check and load a leading dictionary page if present. When the first
page is a data page,
+ // retain its parsed header for parse_page_header().
+ Status load_dictionary_page(bool* has_dict);
// Get page decoder
Decoder* get_page_decoder() { return _page_decoder; }
@@ -215,8 +219,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);
diff --git a/be/src/format/parquet/vparquet_column_reader.cpp
b/be/src/format/parquet/vparquet_column_reader.cpp
index 53e44babbd8..7fa9a6702ee 100644
--- a/be/src/format/parquet/vparquet_column_reader.cpp
+++ b/be/src/format/parquet/vparquet_column_reader.cpp
@@ -534,9 +534,8 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_nested_column(
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_dict_values_to_column(
MutableColumnPtr& doris_column, bool* has_dict) {
- bool loaded;
- RETURN_IF_ERROR(_try_load_dict_page(&loaded, has_dict));
- if (loaded && *has_dict) {
+ RETURN_IF_ERROR(_chunk_reader->load_dictionary_page(has_dict));
+ if (*has_dict) {
return _chunk_reader->read_dict_values_to_column(doris_column);
}
return Status::OK();
@@ -548,15 +547,6 @@ ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::convert_dict_column_to_string_c
return _chunk_reader->convert_dict_column_to_string_column(dict_column);
}
-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,
diff --git a/be/src/format/parquet/vparquet_column_reader.h
b/be/src/format/parquet/vparquet_column_reader.h
index d6c7d480247..d1855d487c5 100644
--- a/be/src/format/parquet/vparquet_column_reader.h
+++ b/be/src/format/parquet/vparquet_column_reader.h
@@ -325,7 +325,6 @@ private:
Status _read_nested_column(ColumnPtr& doris_column, DataTypePtr& type,
FilterMap& filter_map,
size_t batch_size, size_t* read_rows, bool* eof,
bool is_dict_filter);
- Status _try_load_dict_page(bool* loaded, bool* has_dict);
};
class ArrayColumnReader : public ParquetColumnReader {
diff --git a/be/test/format/parquet/parquet_column_chunk_reader_test.cpp
b/be/test/format/parquet/parquet_column_chunk_reader_test.cpp
new file mode 100644
index 00000000000..be9616c523f
--- /dev/null
+++ b/be/test/format/parquet/parquet_column_chunk_reader_test.cpp
@@ -0,0 +1,488 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "core/assert_cast.h"
+#include "core/column/column_string.h"
+#include "format/parquet/schema_desc.h"
+#include "format/parquet/vparquet_column_chunk_reader.h"
+#include "format/parquet/vparquet_column_reader.h"
+#include "io/fs/buffered_reader.h"
+#include "io/fs/file_reader.h"
+#include "runtime/runtime_state.h"
+#include "util/coding.h"
+#include "util/thrift_util.h"
+
+namespace doris {
+namespace {
+
+class CountingBufferedReader final : public io::BufferedStreamReader {
+public:
+ explicit CountingBufferedReader(std::vector<uint8_t> data) :
_data(std::move(data)) {}
+
+ Status read_bytes(const uint8_t** buf, uint64_t offset, size_t
bytes_to_read,
+ const io::IOContext* io_ctx) override {
+ ++_read_count;
+ if (_read_count == _failed_read) {
+ return Status::IOError("Injected read failure");
+ }
+ if (offset + bytes_to_read > _data.size()) {
+ return Status::IOError("Out of bounds");
+ }
+ *buf = _data.data() + offset;
+ return Status::OK();
+ }
+
+ Status read_bytes(Slice& slice, uint64_t offset, const io::IOContext*
io_ctx) override {
+ ++_read_count;
+ if (_read_count == _failed_read) {
+ return Status::IOError("Injected read failure");
+ }
+ if (offset + slice.size > _data.size()) {
+ return Status::IOError("Out of bounds");
+ }
+ slice.data = reinterpret_cast<char*>(_data.data() + offset);
+ return Status::OK();
+ }
+
+ std::string path() override { return "parquet_column_chunk_reader_test"; }
+ int64_t mtime() const override { return 0; }
+ size_t read_count() const { return _read_count; }
+ void fail_on_read(size_t read_count) { _failed_read = read_count; }
+
+private:
+ std::vector<uint8_t> _data;
+ size_t _read_count = 0;
+ size_t _failed_read = 0;
+};
+
+class CountingFileReader final : public io::FileReader {
+public:
+ explicit CountingFileReader(std::vector<uint8_t> data) :
_data(std::move(data)) {}
+
+ Status close() override {
+ _closed = true;
+ return Status::OK();
+ }
+
+ const io::Path& path() const override { return _path; }
+ size_t size() const override { return _data.size(); }
+ bool closed() const override { return _closed; }
+ int64_t mtime() const override { return 0; }
+ 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* io_ctx) override {
+ ++_read_count;
+ if (offset > _data.size()) {
+ return Status::IOError("Out of bounds");
+ }
+ *bytes_read = std::min(result.size, _data.size() - offset);
+ memcpy(result.data, _data.data() + offset, *bytes_read);
+ return Status::OK();
+ }
+
+private:
+ std::vector<uint8_t> _data;
+ io::Path _path = "parquet_scalar_column_reader_test";
+ size_t _read_count = 0;
+ bool _closed = false;
+};
+
+struct ColumnChunkFixture {
+ std::vector<uint8_t> data;
+ tparquet::ColumnChunk chunk;
+ tparquet::OffsetIndex offset_index;
+ FieldSchema field_schema;
+};
+
+Status append_page(tparquet::PageHeader* header, const std::vector<uint8_t>&
payload,
+ std::vector<uint8_t>& data, int64_t* page_offset, int32_t*
page_size) {
+ std::vector<uint8_t> header_bytes;
+ ThriftSerializer serializer(/*compact=*/true, /*initial_buffer_size=*/256);
+ RETURN_IF_ERROR(serializer.serialize(header, &header_bytes));
+
+ *page_offset = data.size();
+ data.insert(data.end(), header_bytes.begin(), header_bytes.end());
+ data.insert(data.end(), payload.begin(), payload.end());
+ *page_size = cast_set<int32_t>(header_bytes.size() + payload.size());
+ return Status::OK();
+}
+
+std::vector<uint8_t> encode_byte_array_dictionary(const
std::vector<std::string>& values) {
+ size_t size = 0;
+ for (const auto& value : values) {
+ size += sizeof(uint32_t) + value.size();
+ }
+
+ std::vector<uint8_t> data(size);
+ size_t offset = 0;
+ for (const auto& value : values) {
+ encode_fixed32_le(data.data() + offset,
cast_set<uint32_t>(value.size()));
+ offset += sizeof(uint32_t);
+ memcpy(data.data() + offset, value.data(), value.size());
+ offset += value.size();
+ }
+ return data;
+}
+
+tparquet::PageHeader make_data_page_header(tparquet::Encoding::type encoding) {
+ tparquet::DataPageHeader data_header;
+ data_header.__set_num_values(1);
+ data_header.__set_encoding(encoding);
+ data_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+ data_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ header.__set_compressed_page_size(1);
+ header.__set_uncompressed_page_size(1);
+ header.__set_data_page_header(data_header);
+ return header;
+}
+
+Status make_dictionary_fixture(ColumnChunkFixture* fixture) {
+ constexpr size_t PREFIX_SIZE = 16;
+ fixture->data.resize(PREFIX_SIZE, 0);
+
+ const std::vector<std::string> dictionary = {"alice", "bob", "carol"};
+ std::vector<uint8_t> dictionary_data =
encode_byte_array_dictionary(dictionary);
+
+ tparquet::DictionaryPageHeader dictionary_header;
+ dictionary_header.__set_num_values(cast_set<int32_t>(dictionary.size()));
+ dictionary_header.__set_encoding(tparquet::Encoding::PLAIN);
+
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DICTIONARY_PAGE;
+
header.__set_compressed_page_size(cast_set<int32_t>(dictionary_data.size()));
+
header.__set_uncompressed_page_size(cast_set<int32_t>(dictionary_data.size()));
+ header.__set_dictionary_page_header(dictionary_header);
+
+ int64_t dictionary_offset = 0;
+ int32_t dictionary_page_size = 0;
+ RETURN_IF_ERROR(append_page(&header, dictionary_data, fixture->data,
&dictionary_offset,
+ &dictionary_page_size));
+
+ std::vector<int64_t> data_page_offsets;
+ std::vector<int32_t> data_page_sizes;
+ for (int i = 0; i < 2; ++i) {
+ header = make_data_page_header(tparquet::Encoding::RLE_DICTIONARY);
+ int64_t page_offset = 0;
+ int32_t page_size = 0;
+ RETURN_IF_ERROR(append_page(&header, {0}, fixture->data, &page_offset,
&page_size));
+ data_page_offsets.push_back(page_offset);
+ data_page_sizes.push_back(page_size);
+ }
+
+ auto& metadata = fixture->chunk.meta_data;
+ metadata.__set_type(tparquet::Type::BYTE_ARRAY);
+ metadata.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ metadata.__set_num_values(2);
+ metadata.__set_dictionary_page_offset(dictionary_offset);
+ metadata.__set_data_page_offset(data_page_offsets[0]);
+ metadata.__set_total_compressed_size(
+ cast_set<int64_t>(fixture->data.size() - dictionary_offset));
+
+ for (int i = 0; i < 2; ++i) {
+ tparquet::PageLocation location;
+ location.__set_offset(data_page_offsets[i]);
+ location.__set_compressed_page_size(data_page_sizes[i]);
+ location.__set_first_row_index(i);
+ fixture->offset_index.page_locations.push_back(location);
+ }
+
+ fixture->field_schema.physical_type = tparquet::Type::BYTE_ARRAY;
+ return Status::OK();
+}
+
+Status make_plain_fixture(ColumnChunkFixture* fixture, int page_count = 1) {
+ constexpr size_t PREFIX_SIZE = 16;
+ fixture->data.resize(PREFIX_SIZE, 0);
+
+ std::vector<int64_t> data_page_offsets;
+ for (int i = 0; i < page_count; ++i) {
+ tparquet::PageHeader header =
make_data_page_header(tparquet::Encoding::PLAIN);
+ int64_t data_page_offset = 0;
+ int32_t data_page_size = 0;
+ RETURN_IF_ERROR(
+ append_page(&header, {0}, fixture->data, &data_page_offset,
&data_page_size));
+ data_page_offsets.push_back(data_page_offset);
+
+ tparquet::PageLocation location;
+ location.__set_offset(data_page_offset);
+ location.__set_compressed_page_size(data_page_size);
+ location.__set_first_row_index(i);
+ fixture->offset_index.page_locations.push_back(location);
+ }
+
+ auto& metadata = fixture->chunk.meta_data;
+ metadata.__set_type(tparquet::Type::BYTE_ARRAY);
+ metadata.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ metadata.__set_num_values(page_count);
+ metadata.__set_data_page_offset(data_page_offsets.front());
+ metadata.__set_total_compressed_size(
+ cast_set<int64_t>(fixture->data.size() -
data_page_offsets.front()));
+
+ fixture->field_schema.physical_type = tparquet::Type::BYTE_ARRAY;
+ return Status::OK();
+}
+
+void expect_dictionary_values(ColumnChunkReader<false, false>* reader) {
+ MutableColumnPtr column = ColumnString::create();
+ ASSERT_TRUE(reader->read_dict_values_to_column(column).ok());
+ const auto& strings = assert_cast<const ColumnString&>(*column);
+ ASSERT_EQ(strings.size(), 3);
+ EXPECT_EQ(std::string(strings.get_data_at(0)), "alice");
+ EXPECT_EQ(std::string(strings.get_data_at(1)), "bob");
+ EXPECT_EQ(std::string(strings.get_data_at(2)), "carol");
+}
+
+TEST(ParquetColumnChunkReaderTest, DictionaryProbeDoesNotParseDataPageHeader) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, false> reader(&buffered_reader, &fixture.chunk,
&fixture.field_schema,
+ nullptr, 2, nullptr, page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(buffered_reader.read_count(), 0);
+
+ bool has_dict = false;
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_TRUE(has_dict);
+ const size_t dictionary_read_count = buffered_reader.read_count();
+ EXPECT_GT(dictionary_read_count, 0);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 1);
+ expect_dictionary_values(&reader);
+
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_EQ(buffered_reader.read_count(), dictionary_read_count);
+
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_GT(buffered_reader.read_count(), dictionary_read_count);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 2);
+ EXPECT_EQ(reader.page_start_row(), 0);
+ EXPECT_EQ(reader.page_end_row(), 1);
+
+ const size_t data_header_read_count = buffered_reader.read_count();
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_EQ(buffered_reader.read_count(), data_header_read_count);
+}
+
+TEST(ParquetColumnChunkReaderTest, ParsePageHeaderLoadsDictionaryOnFirstUse) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, false> reader(&buffered_reader, &fixture.chunk,
&fixture.field_schema,
+ nullptr, 2, nullptr, page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(buffered_reader.read_count(), 0);
+
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_GT(buffered_reader.read_count(), 0);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 2);
+ EXPECT_EQ(reader.page_start_row(), 0);
+ EXPECT_EQ(reader.page_end_row(), 1);
+
+ const size_t read_count = buffered_reader.read_count();
+ bool has_dict = false;
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_TRUE(has_dict);
+ EXPECT_EQ(buffered_reader.read_count(), read_count);
+ expect_dictionary_values(&reader);
+}
+
+TEST(ParquetColumnChunkReaderTest,
SequentialReaderAdvancesAfterLazyDictionaryLoad) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, false> reader(&buffered_reader, &fixture.chunk,
&fixture.field_schema,
+ nullptr, 2, nullptr, page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_EQ(reader.page_start_row(), 0);
+ EXPECT_EQ(reader.page_end_row(), 1);
+
+ ASSERT_TRUE(reader.next_page().ok());
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_EQ(reader.page_start_row(), 1);
+ EXPECT_EQ(reader.page_end_row(), 2);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 3);
+}
+
+TEST(ParquetColumnChunkReaderTest,
PlainPageHeaderIsReusedAfterDictionaryCheck) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_plain_fixture(&fixture).ok());
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, false> reader(&buffered_reader, &fixture.chunk,
&fixture.field_schema,
+ nullptr, 1, nullptr, page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(buffered_reader.read_count(), 0);
+
+ bool has_dict = true;
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_FALSE(has_dict);
+ const size_t dictionary_probe_read_count = buffered_reader.read_count();
+ EXPECT_GT(dictionary_probe_read_count, 0);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 1);
+
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_EQ(buffered_reader.read_count(), dictionary_probe_read_count);
+ EXPECT_EQ(reader.page_start_row(), 0);
+ EXPECT_EQ(reader.page_end_row(), 1);
+}
+
+TEST(ParquetColumnChunkReaderTest, FailedDictionaryCheckCanBeRetried) {
+ for (size_t failed_read : {1, 2}) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, false> reader(&buffered_reader,
&fixture.chunk,
+ &fixture.field_schema, nullptr,
2, nullptr,
+ page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ buffered_reader.fail_on_read(failed_read);
+
+ bool has_dict = false;
+ EXPECT_FALSE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_FALSE(has_dict);
+
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_TRUE(has_dict);
+ EXPECT_GT(buffered_reader.read_count(), failed_read);
+ }
+}
+
+TEST(ParquetColumnChunkReaderTest, ScalarDictionaryReadUsesExplicitProbe) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+ auto file_reader =
std::make_shared<CountingFileReader>(std::move(fixture.data));
+
+ RowRanges row_ranges;
+ row_ranges.add({0, 2});
+ ScalarColumnReader<false, false> reader(row_ranges, 2, fixture.chunk,
nullptr, nullptr,
+ nullptr);
+
+ TQueryOptions query_options;
+ query_options.__set_enable_parquet_file_page_cache(false);
+ RuntimeState runtime_state(query_options, TQueryGlobals());
+
+ ASSERT_TRUE(reader.init(file_reader, &fixture.field_schema,
+ /*max_buf_size=*/1024 * 1024, &runtime_state)
+ .ok());
+ EXPECT_EQ(file_reader->read_count(), 0);
+
+ MutableColumnPtr column = ColumnString::create();
+ bool has_dict = false;
+ ASSERT_TRUE(reader.read_dict_values_to_column(column, &has_dict).ok());
+ EXPECT_TRUE(has_dict);
+ EXPECT_EQ(reader.column_statistics().parse_page_header_num, 1);
+
+ const auto& strings = assert_cast<const ColumnString&>(*column);
+ ASSERT_EQ(strings.size(), 3);
+ EXPECT_EQ(std::string(strings.get_data_at(0)), "alice");
+ EXPECT_EQ(std::string(strings.get_data_at(1)), "bob");
+ EXPECT_EQ(std::string(strings.get_data_at(2)), "carol");
+}
+
+void expect_offset_index_skip(ColumnChunkFixture fixture) {
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, true> reader(&buffered_reader, &fixture.chunk,
&fixture.field_schema,
+ &fixture.offset_index, 2, nullptr,
page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(buffered_reader.read_count(), 0);
+
+ ASSERT_TRUE(reader.next_page().ok());
+ const size_t dictionary_read_count = buffered_reader.read_count();
+ EXPECT_GT(dictionary_read_count, 0);
+
+ bool has_dict = false;
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_TRUE(has_dict);
+ EXPECT_EQ(buffered_reader.read_count(), dictionary_read_count);
+
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_GT(buffered_reader.read_count(), dictionary_read_count);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 2);
+ EXPECT_EQ(reader.page_start_row(), 1);
+ EXPECT_EQ(reader.page_end_row(), 2);
+}
+
+TEST(ParquetColumnChunkReaderTest,
OffsetIndexSkipLoadsDictionaryBeforeMovingPage) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+ expect_offset_index_skip(std::move(fixture));
+}
+
+TEST(ParquetColumnChunkReaderTest,
OffsetIndexSkipFindsDictionaryWithoutMetadataOffset) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());
+
fixture.chunk.meta_data.__set_data_page_offset(fixture.chunk.meta_data.dictionary_page_offset);
+ fixture.chunk.meta_data.__isset.dictionary_page_offset = false;
+ expect_offset_index_skip(std::move(fixture));
+}
+
+TEST(ParquetColumnChunkReaderTest, OffsetIndexSkipMovesPastPlainFirstPage) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_plain_fixture(&fixture, 2).ok());
+ CountingBufferedReader buffered_reader(std::move(fixture.data));
+ ParquetPageReadContext page_read_ctx(false);
+ ColumnChunkReader<false, true> reader(&buffered_reader, &fixture.chunk,
&fixture.field_schema,
+ &fixture.offset_index, 2, nullptr,
page_read_ctx);
+
+ ASSERT_TRUE(reader.init().ok());
+ EXPECT_EQ(buffered_reader.read_count(), 0);
+
+ ASSERT_TRUE(reader.next_page().ok());
+ const size_t dictionary_probe_read_count = buffered_reader.read_count();
+ EXPECT_GT(dictionary_probe_read_count, 0);
+
+ bool has_dict = true;
+ ASSERT_TRUE(reader.load_dictionary_page(&has_dict).ok());
+ EXPECT_FALSE(has_dict);
+ EXPECT_EQ(buffered_reader.read_count(), dictionary_probe_read_count);
+
+ ASSERT_TRUE(reader.parse_page_header().ok());
+ EXPECT_GT(buffered_reader.read_count(), dictionary_probe_read_count);
+ EXPECT_EQ(reader.statistics().parse_page_header_num, 2);
+ EXPECT_EQ(reader.page_start_row(), 1);
+ EXPECT_EQ(reader.page_end_row(), 2);
+}
+
+} // namespace
+} // namespace doris
diff --git a/be/test/format/parquet/parquet_page_cache_test.cpp
b/be/test/format/parquet/parquet_page_cache_test.cpp
index 8812d942323..664882ff3f3 100644
--- a/be/test/format/parquet/parquet_page_cache_test.cpp
+++ b/be/test/format/parquet/parquet_page_cache_test.cpp
@@ -57,6 +57,11 @@ private:
std::vector<uint8_t> _data;
};
+static Status parse_and_load_page(ColumnChunkReader<false, false>* reader) {
+ RETURN_IF_ERROR(reader->parse_page_header());
+ return reader->load_page_data();
+}
+
TEST(ParquetPageCacheTest, CacheHitReturnsDecompressedPayload) {
ParquetPageReadContext ctx;
ctx.enable_parquet_file_page_cache = true;
@@ -113,7 +118,7 @@ TEST(ParquetPageCacheTest,
CacheHitReturnsDecompressedPayload) {
ColumnChunkReader<false, false> ccr(&reader, &cc, &field_schema, nullptr,
0, nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
// load_page_data should hit the cache and return decompressed payload
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
Slice s = ccr.get_page_data();
ASSERT_EQ(s.size, payload.size());
ASSERT_EQ(0, memcmp(s.data, payload.data(), payload.size()));
@@ -168,14 +173,14 @@ TEST(ParquetPageCacheTest,
DecompressedPageInsertedByColumnChunkReader) {
field_schema.definition_level = 0;
ColumnChunkReader<false, false> ccr(&reader, &cc, &field_schema,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
// Now cache should have an entry; verify by creating a fresh
ColumnChunkReader and hitting cache
ColumnChunkReader<false, false> ccr_check(&reader, &cc, &field_schema,
nullptr, 0, nullptr,
ctx);
ASSERT_TRUE(ccr_check.init().ok());
// ASSERT_TRUE(ccr_check.next_page().ok());
- ASSERT_TRUE(ccr_check.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_check).ok());
Slice s = ccr_check.get_page_data();
ASSERT_EQ(s.size, payload.size());
EXPECT_EQ(0, memcmp(s.data, payload.data(), payload.size()));
@@ -236,13 +241,13 @@ TEST(ParquetPageCacheTest, V2LevelsPreservedInCache) {
{
ColumnChunkReader<false, false> ccr(&reader, &cc, &field_schema,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
// Now cache should have entry; verify by creating a ColumnChunkReader
and hitting cache
ColumnChunkReader<false, false> ccr_check(&reader, &cc, &field_schema,
nullptr, 0, nullptr,
ctx);
ASSERT_TRUE(ccr_check.init().ok());
- ASSERT_TRUE(ccr_check.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_check).ok());
Slice s = ccr_check.get_page_data();
ASSERT_EQ(s.size, payload.size());
EXPECT_EQ(0, memcmp(s.data, payload.data(), payload.size()));
@@ -254,7 +259,7 @@ TEST(ParquetPageCacheTest, V2LevelsPreservedInCache) {
field_schema2.definition_level = 1;
ColumnChunkReader<false, false> ccr2(&reader, &cc, &field_schema2,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr2.init().ok());
- ASSERT_TRUE(ccr2.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr2).ok());
// Level slices should equal the original level bytes
const Slice& rep = ccr2.v2_rep_levels();
const Slice& def = ccr2.v2_def_levels();
@@ -318,7 +323,7 @@ TEST(ParquetPageCacheTest, CompressedV1PageCachedAndHit) {
// Load page to trigger decompression + cache insert
ColumnChunkReader<false, false> ccr(&reader, &cc, &field_schema, nullptr,
0, nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
EXPECT_EQ(ccr.statistics().page_cache_write_counter, 1);
// Now verify a fresh reader hits the cache and returns payload
@@ -326,7 +331,7 @@ TEST(ParquetPageCacheTest, CompressedV1PageCachedAndHit) {
ctx);
ASSERT_TRUE(ccr_check.init().ok());
// ASSERT_TRUE(ccr_check.next_page().ok());
- ASSERT_TRUE(ccr_check.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_check).ok());
Slice s = ccr_check.get_page_data();
ASSERT_EQ(s.size, payload.size());
EXPECT_EQ(0, memcmp(s.data, payload.data(), payload.size()));
@@ -393,7 +398,7 @@ TEST(ParquetPageCacheTest,
CompressedV2LevelsPreservedInCache) {
// Load page to trigger decompression + cache insert
ColumnChunkReader<false, false> ccr(&reader, &cc, &field_schema, nullptr,
0, nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
EXPECT_EQ(ccr.statistics().page_cache_write_counter, 1);
// Now verify a fresh reader hits the cache and v2 levels are preserved
@@ -403,7 +408,7 @@ TEST(ParquetPageCacheTest,
CompressedV2LevelsPreservedInCache) {
ColumnChunkReader<false, false> ccr_check(&reader, &cc, &field_schema2,
nullptr, 0, nullptr,
ctx);
ASSERT_TRUE(ccr_check.init().ok());
- ASSERT_TRUE(ccr_check.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_check).ok());
Slice s = ccr_check.get_page_data();
ASSERT_EQ(s.size, payload.size());
EXPECT_EQ(0, memcmp(s.data, payload.data(), payload.size()));
@@ -498,7 +503,7 @@ TEST(ParquetPageCacheTest, MultiPagesMixedV1V2CacheHit) {
field_schema1.definition_level = 0;
ColumnChunkReader<false, false> ccr1(&reader1, &cc1, &field_schema1,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr1.init().ok());
- ASSERT_TRUE(ccr1.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr1).ok());
Slice s1 = ccr1.get_page_data();
ASSERT_EQ(s1.size, payload1.size());
EXPECT_EQ(0, memcmp(s1.data, payload1.data(), payload1.size()));
@@ -516,7 +521,7 @@ TEST(ParquetPageCacheTest, MultiPagesMixedV1V2CacheHit) {
field_schema2.definition_level = dl;
ColumnChunkReader<false, false> ccr2(&reader2, &cc2, &field_schema2,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr2.init().ok());
- ASSERT_TRUE(ccr2.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr2).ok());
Slice s2 = ccr2.get_page_data();
ASSERT_EQ(s2.size, payload2.size());
EXPECT_EQ(0, memcmp(s2.data, payload2.data(), payload2.size()));
@@ -566,7 +571,7 @@ TEST(ParquetPageCacheTest, CacheMissThenHit) {
// First reader: should not hit cache, but should write cache
ColumnChunkReader<false, false> ccr(&reader, &cc, &fs, nullptr, 0,
nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
auto& statistics = ccr.statistics();
EXPECT_EQ(statistics.page_cache_hit_counter, 0);
EXPECT_EQ(statistics.page_cache_write_counter, 1);
@@ -574,7 +579,7 @@ TEST(ParquetPageCacheTest, CacheMissThenHit) {
// Second reader: should hit cache
ColumnChunkReader<false, false> ccr2(&reader, &cc, &fs, nullptr, 0,
nullptr, ctx);
ASSERT_TRUE(ccr2.init().ok());
- ASSERT_TRUE(ccr2.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr2).ok());
auto& statistics2 = ccr2.statistics();
EXPECT_EQ(statistics2.page_cache_hit_counter, 1);
EXPECT_EQ(statistics2.page_cache_decompressed_hit_counter, 1);
@@ -632,7 +637,7 @@ TEST(ParquetPageCacheTest,
DecompressThresholdCachesCompressed) {
ColumnChunkReader<false, false> ccr_small_thresh(&reader, &cc, &fs,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr_small_thresh.init().ok());
// ASSERT_TRUE(ccr_small_thresh.next_page().ok());
- ASSERT_TRUE(ccr_small_thresh.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_small_thresh).ok());
EXPECT_EQ(ccr_small_thresh.statistics().page_cache_write_counter, 1);
// Inspect cache entry: payload stored should be compressed size
@@ -706,7 +711,7 @@ TEST(ParquetPageCacheTest,
DecompressThresholdCachesDecompressed) {
ColumnChunkReader<false, false> ccr_large_thresh(&reader, &cc, &fs,
nullptr, 0, nullptr, ctx);
ASSERT_TRUE(ccr_large_thresh.init().ok());
// ASSERT_TRUE(ccr_large_thresh.next_page().ok());
- ASSERT_TRUE(ccr_large_thresh.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_large_thresh).ok());
EXPECT_EQ(ccr_large_thresh.statistics().page_cache_write_counter, 1);
// Inspect cache entry for large threshold: payload stored should be
uncompressed size
@@ -726,7 +731,7 @@ TEST(ParquetPageCacheTest,
DecompressThresholdCachesDecompressed) {
ColumnChunkReader<false, false> ccr_check(&reader, &cc, &fs, nullptr, 0,
nullptr, ctx);
ASSERT_TRUE(ccr_check.init().ok());
// ASSERT_TRUE(ccr_check.next_page().ok());
- ASSERT_TRUE(ccr_check.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr_check).ok());
EXPECT_EQ(ccr_check.statistics().page_cache_hit_counter, 1);
// restore config
config::parquet_page_cache_decompress_threshold = old_thresh;
@@ -789,7 +794,7 @@ TEST(ParquetPageCacheTest, MultipleReadersShareCachedEntry)
{
fs.definition_level = dl;
ColumnChunkReader<false, false> ccr(&reader, &cc, &fs, nullptr, 0,
nullptr, ctx);
ASSERT_TRUE(ccr.init().ok());
- ASSERT_TRUE(ccr.load_page_data().ok());
+ ASSERT_TRUE(parse_and_load_page(&ccr).ok());
Slice s = ccr.get_page_data();
ASSERT_EQ(s.size, payload.size());
EXPECT_EQ(0, memcmp(s.data, payload.data(), payload.size()));
diff --git a/be/test/format/parquet/parquet_reader_test.cpp
b/be/test/format/parquet/parquet_reader_test.cpp
index 705ed289c3d..0f64bf5d4ae 100644
--- a/be/test/format/parquet/parquet_reader_test.cpp
+++ b/be/test/format/parquet/parquet_reader_test.cpp
@@ -236,6 +236,11 @@ public:
pq_ctx.slot_id_to_filter_conjuncts = &slot_id_to_expr_ctxs;
pq_ctx.params = &scan_params;
pq_ctx.range = &scan_range;
+ if constexpr (filter_all && enable_lazy) {
+ // Exercise row-level lazy filtering instead of eliminating the
row group from
+ // min/max or page-index metadata before a RowGroupReader is
created.
+ pq_ctx.filter_groups = false;
+ }
st = p_reader->init_reader(&pq_ctx);
EXPECT_TRUE(st.ok()) << st;
@@ -275,6 +280,22 @@ public:
} else {
EXPECT_EQ(total_rows, 10000);
}
+
+ if constexpr (filter_all && enable_lazy) {
+ EXPECT_EQ(p_reader->reader_statistics().lazy_read_filtered_rows,
10000);
+ ASSERT_NE(p_reader->_current_group_reader, nullptr);
+
+ const auto follower_statistics =
+
p_reader->_current_group_reader->_column_readers.at("string_col")
+ ->column_statistics();
+ EXPECT_EQ(follower_statistics.page_read_counter, 0);
+ EXPECT_EQ(follower_statistics.parse_page_header_num, 0);
+
+ const auto predicate_statistics =
+
p_reader->_current_group_reader->_column_readers.at("value_col")
+ ->column_statistics();
+ EXPECT_GT(predicate_statistics.page_read_counter, 0);
+ }
}
};
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
