This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new ded7cfb8 perf(read): reduce point-lookup metadata I/O (#285)
ded7cfb8 is described below
commit ded7cfb84005a6315f4778253411dc78f7fa4cb3
Author: wangyong9999 <[email protected]>
AuthorDate: Tue Sep 8 13:38:34 2026 +0800
perf(read): reduce point-lookup metadata I/O (#285)
---
src/paimon/core/utils/snapshot_manager.cpp | 6 +-
src/paimon/core/utils/snapshot_manager_test.cpp | 28 +++++
src/paimon/format/parquet/file_reader_wrapper.cpp | 31 ++++-
.../format/parquet/file_reader_wrapper_test.cpp | 38 ++++++
.../parquet/parquet_file_batch_reader_test.cpp | 116 +++++++++++++++++++
src/paimon/format/parquet/parquet_input_stream.h | 127 +++++++++++++++++++++
src/paimon/format/parquet/parquet_reader_builder.h | 15 +--
7 files changed, 349 insertions(+), 12 deletions(-)
diff --git a/src/paimon/core/utils/snapshot_manager.cpp
b/src/paimon/core/utils/snapshot_manager.cpp
index 3d38a695..9211d966 100644
--- a/src/paimon/core/utils/snapshot_manager.cpp
+++ b/src/paimon/core/utils/snapshot_manager.cpp
@@ -141,10 +141,6 @@ Result<std::optional<int64_t>>
SnapshotManager::FindEarliest(
Result<std::optional<int64_t>> SnapshotManager::FindLatest(
const std::string& dir, const std::string& prefix,
const std::function<std::string(int64_t)>& path_func) const {
- PAIMON_ASSIGN_OR_RAISE(bool is_exist, fs_->Exists(dir));
- if (!is_exist) {
- return std::optional<int64_t>();
- }
std::optional<int64_t> snapshot_id = ReadHint(LATEST, dir);
if (snapshot_id != std::nullopt && snapshot_id.value() > 0) {
int64_t next_snapshot = snapshot_id.value() + 1;
@@ -155,6 +151,8 @@ Result<std::optional<int64_t>> SnapshotManager::FindLatest(
return snapshot_id;
}
}
+ // A valid hint needs no parent-directory probe. The listing fallback
checks
+ // directory existence itself, including tables without any snapshots yet.
return FindByListFiles([](int64_t lhs, int64_t rhs) -> int64_t { return
std::max(lhs, rhs); },
dir, prefix);
}
diff --git a/src/paimon/core/utils/snapshot_manager_test.cpp
b/src/paimon/core/utils/snapshot_manager_test.cpp
index a4794e0e..d1bc8309 100644
--- a/src/paimon/core/utils/snapshot_manager_test.cpp
+++ b/src/paimon/core/utils/snapshot_manager_test.cpp
@@ -20,6 +20,7 @@
#include <filesystem>
#include <limits>
+#include <vector>
#include "gtest/gtest.h"
#include "paimon/common/utils/path_util.h"
@@ -139,6 +140,33 @@ TEST(SnapshotManagerTest, TestPathNotExist) {
ASSERT_EQ(snapshot, std::nullopt);
}
+TEST(SnapshotManagerTest, LatestSnapshotWithStaleMissingOrInvalidHint) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fs = std::make_shared<LocalFileSystem>();
+ SnapshotManager mgr(fs, dir->Str());
+ ASSERT_OK(fs->Mkdirs(mgr.SnapshotDirectory()));
+ ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(1), "{}", true));
+ ASSERT_OK(mgr.CommitLatestHint(1));
+
+ ASSERT_OK_AND_ASSIGN(std::optional<int64_t> latest,
mgr.LatestSnapshotId());
+ ASSERT_EQ(latest, 1);
+
+ // A commit can publish its snapshot before updating the hint.
+ ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(2), "{}", true));
+ ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());
+ ASSERT_EQ(latest, 2);
+
+ const std::string hint_path = PathUtil::JoinPath(mgr.SnapshotDirectory(),
"LATEST");
+ ASSERT_OK(fs->Delete(hint_path));
+ ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());
+ ASSERT_EQ(latest, 2);
+
+ ASSERT_OK(fs->WriteFile(hint_path, "invalid", true));
+ ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());
+ ASSERT_EQ(latest, 2);
+}
+
TEST(SnapshotManagerTest, TestEarlierOrEqualTimeMillisExactMatch) {
std::string test_data_path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
auto file_system = std::make_shared<LocalFileSystem>();
diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp
b/src/paimon/format/parquet/file_reader_wrapper.cpp
index fb5b6574..f69b89d4 100644
--- a/src/paimon/format/parquet/file_reader_wrapper.cpp
+++ b/src/paimon/format/parquet/file_reader_wrapper.cpp
@@ -29,10 +29,12 @@
#include "fmt/format.h"
#include "paimon/common/utils/arrow/arrow_utils.h"
#include "paimon/common/utils/math.h"
+#include "paimon/common/utils/scope_guard.h"
#include "paimon/format/parquet/column_index_filter.h"
#include "paimon/format/parquet/page_filtered_row_group_reader.h"
#include "paimon/format/parquet/parquet_format_defs.h"
#include "paimon/macros.h"
+#include "paimon/predicate/predicate_utils.h"
#include "parquet/arrow/reader.h"
#include "parquet/arrow/schema.h"
#include "parquet/file_reader.h"
@@ -602,8 +604,35 @@ Result<RowRanges>
FileReaderWrapper::CalculateFilteredRowRanges(
return RowRanges::CreateSingle(row_count);
}
+ auto page_index_reader = GetPageIndexReader();
+ if (!page_index_reader) {
+ return RowRanges::CreateSingle(row_count);
+ }
+ std::set<std::string> field_names;
+ PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate,
&field_names));
+ std::vector<int32_t> predicate_columns;
+ for (const auto& name : field_names) {
+ auto it = column_name_to_index.find(name);
+ if (it == column_name_to_index.end()) {
+ return Status::Invalid(
+ fmt::format("column '{}' not found in
column_name_to_index", name));
+ }
+ predicate_columns.push_back(it->second);
+ }
+ if (predicate_columns.empty()) {
+ return RowRanges::CreateSingle(row_count);
+ }
+
+ // Arrow otherwise reads the column-index envelope of every column,
including
+ // potentially large min/max values in payload columns not used by the
predicate.
+ const std::vector<int32_t> row_groups = {row_group_index};
+ ScopeGuard clear_hint([&]() {
page_index_reader->WillNotNeed(row_groups); });
+ page_index_reader->WillNeed(row_groups, predicate_columns,
+ {/*column_index=*/true,
/*offset_index=*/true});
+ // Keep this restricted reader separate: projected payload columns
still need
+ // their offset indexes when the data reader is initialized later.
return ColumnIndexFilter::CalculateRowRanges(predicate,
-
GetRowGroupPageIndexReader(row_group_index),
+
page_index_reader->RowGroup(row_group_index),
column_name_to_index,
row_count);
}
PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::CalculateFilteredRowRanges")
diff --git a/src/paimon/format/parquet/file_reader_wrapper_test.cpp
b/src/paimon/format/parquet/file_reader_wrapper_test.cpp
index 41dcde19..e2335107 100644
--- a/src/paimon/format/parquet/file_reader_wrapper_test.cpp
+++ b/src/paimon/format/parquet/file_reader_wrapper_test.cpp
@@ -46,9 +46,14 @@
#include "paimon/fs/file_system.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate_builder.h"
#include "paimon/record_batch.h"
#include "paimon/testing/utils/testharness.h"
#include "parquet/arrow/reader.h"
+#include "parquet/file_reader.h"
+#include "parquet/metadata.h"
+#include "parquet/page_index.h"
#include "parquet/properties.h"
namespace arrow {
@@ -273,6 +278,39 @@ TEST_F(FileReaderWrapperTest, NullFileReader) {
"file reader wrapper create failed. file reader is
nullptr");
}
+TEST_F(FileReaderWrapperTest,
PredicateReadsOnlyItsPageIndexesAndKeepsPayloadReadable) {
+ std::string file_path = PathUtil::JoinPath(dir_->Str(),
"predicate-index.parquet");
+ PrepareParquetFile(file_path, /*row_count=*/1000,
/*enable_page_index=*/true);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input,
fs_->Open(file_path));
+ auto tracking_input = std::make_shared<ReadTrackingInputStream>(input);
+ ASSERT_OK_AND_ASSIGN(auto reader,
PrepareReaderWrapperOnStream(tracking_input));
+ auto metadata = reader->GetFileReader()->parquet_reader()->metadata();
+ auto index_ranges =
+
::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup(*metadata->RowGroup(0),
{1});
+ ASSERT_TRUE(index_ranges.column_index.has_value());
+ ASSERT_TRUE(index_ranges.offset_index.has_value());
+ int64_t bytes_before = tracking_input->GetPositionalReadBytes();
+ auto predicate = PredicateBuilder::Equal(1, "col2", FieldType::INT,
Literal(25));
+ ASSERT_OK_AND_ASSIGN(auto ranges, reader->CalculateFilteredRowRanges(
+ 0, predicate, {{"col1", 0}, {"col2",
1}, {"col3", 2}}));
+ ASSERT_EQ(10, ranges.RowCount());
+ ASSERT_EQ(index_ranges.column_index->length +
index_ranges.offset_index->length,
+ tracking_input->GetPositionalReadBytes() - bytes_before);
+
+ // The predicate's read hint must not restrict subsequent payload
offset-index reads.
+ ASSERT_OK(reader->PrepareForReading(
+ {TargetRowGroup(/*rg_index=*/0, /*is_partially_matched=*/true,
ranges)}, {0, 1, 2}));
+ ASSERT_OK_AND_ASSIGN(auto batch, reader->Next());
+ ASSERT_TRUE(batch);
+ auto expected = PrepareArray(PrepareArrowSchema().second,
/*record_batch_size=*/10,
+ /*offset=*/20);
+ auto expected_batch = arrow::RecordBatch::FromStructArray(expected);
+ ASSERT_TRUE(expected_batch.ok());
+ ASSERT_TRUE(batch->Equals(*expected_batch.ValueOrDie()));
+ ASSERT_OK_AND_ASSIGN(auto end, reader->Next());
+ ASSERT_FALSE(end);
+}
+
TEST_F(FileReaderWrapperTest, Simple) {
std::string file_path = PathUtil::JoinPath(dir_->Str(), "test.parquet");
PrepareParquetFile(file_path, /*row_count=*/5500);
diff --git a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
index 3e4824e3..c52ceba9 100644
--- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
+++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
@@ -343,6 +343,122 @@ TEST_F(ParquetFileBatchReaderTest,
TestParquetMetadataCacheBypassesWhenGetUriFai
ASSERT_EQ(0, cache->Size());
}
+TEST_F(ParquetFileBatchReaderTest, TestPageIndexBytesSurviveReaderClose) {
+ WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
+ /*enable_dictionary=*/false, /*max_row_group_length=*/3,
+ /*max_page_size=*/1);
+ auto cache =
std::make_shared<paimon::test::CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
+ 128 *
1024 * 1024);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> raw_input,
fs_->Open(file_path_));
+ ASSERT_OK_AND_ASSIGN(int64_t length, raw_input->Length());
+ auto raw = std::make_shared<ArrowInputStreamAdapter>(raw_input, length,
pool_);
+ auto raw_reader = ::parquet::ParquetFileReader::Open(raw);
+ auto metadata = raw_reader->metadata();
+ int64_t expected_index_reads = 0;
+ std::weak_ptr<MemoryPool> cached_pool;
+ for (int32_t round = 0; round < 3; ++round) {
+ if (round == 2) {
+ cache->InvalidateAll();
+ ASSERT_TRUE(cached_pool.expired());
+ }
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input,
fs_->Open(file_path_));
+ std::shared_ptr<MemoryPool> query_pool = GetMemoryPool();
+ auto stream = std::make_shared<ParquetInputStream>(input, length,
pool_, query_pool, cache,
+ file_path_);
+ stream->SetPageIndexRanges(*metadata);
+ for (int32_t rg = 0; rg < metadata->num_row_groups(); ++rg) {
+ auto ranges =
::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup(
+ *metadata->RowGroup(rg), {});
+ for (const auto& range : {ranges.column_index,
ranges.offset_index}) {
+ ASSERT_TRUE(range.has_value());
+ auto expected = raw->ReadAt(range->offset, range->length);
+ ASSERT_TRUE(expected.ok()) << expected.status().ToString();
+ auto actual = stream->ReadAt(range->offset, range->length);
+ ASSERT_TRUE(actual.ok()) << actual.status().ToString();
+
ASSERT_TRUE(actual.ValueOrDie()->Equals(*expected.ValueOrDie()));
+ if (round != 1) {
+ ++expected_index_reads;
+ }
+ }
+ }
+ ASSERT_EQ(expected_index_reads, cache->SupplierCallCount());
+ // Reader 2 serves all indexes from the shared cache without storage
IO.
+ if (round == 1) {
+ ASSERT_EQ(0, stream->StorageReadBytes()->load());
+ } else {
+ ASSERT_GT(stream->StorageReadBytes()->load(), 0);
+ }
+ // Ordinary file bytes must still use storage, even on a cache hit
round.
+ uint64_t bytes_before = stream->StorageReadBytes()->load();
+ auto magic = stream->ReadAt(0, 4);
+ ASSERT_TRUE(magic.ok()) << magic.status().ToString();
+ ASSERT_EQ("PAR1", magic.ValueOrDie()->ToString());
+ ASSERT_EQ(bytes_before + 4, stream->StorageReadBytes()->load());
+ ASSERT_EQ(expected_index_reads, cache->SupplierCallCount());
+ ASSERT_TRUE(stream->Close().ok());
+ if (round != 1) {
+ cached_pool = query_pool;
+ }
+ stream.reset();
+ query_pool.reset();
+ ASSERT_FALSE(cached_pool.expired());
+ }
+ cache->InvalidateAll();
+ ASSERT_TRUE(cached_pool.expired());
+}
+
+TEST_F(ParquetFileBatchReaderTest,
TestCachedFooterKeepsAllocatorAliveUntilEviction) {
+ WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
+ /*enable_dictionary=*/false, /*max_row_group_length=*/3);
+ auto cache =
std::make_shared<paimon::test::CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
+ 128 *
1024 * 1024);
+ std::weak_ptr<MemoryPool> cached_pool;
+ {
+ std::shared_ptr<MemoryPool> query_pool = GetMemoryPool();
+ cached_pool = query_pool;
+ ParquetReaderBuilder builder({}, 10);
+ builder.WithMemoryPool(query_pool)->WithCache(cache);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input,
fs_->Open(file_path_));
+ ASSERT_OK_AND_ASSIGN(auto reader, builder.Build(input));
+ }
+ ASSERT_FALSE(cached_pool.expired());
+ cache->InvalidateAll();
+ ASSERT_TRUE(cached_pool.expired());
+}
+
+TEST_F(ParquetFileBatchReaderTest, TestPointReadReusesFooterAndPageIndexes) {
+ WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
+ /*enable_dictionary=*/false, /*max_row_group_length=*/3,
+ /*max_page_size=*/1);
+ auto cache =
std::make_shared<paimon::test::CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
+ 128 *
1024 * 1024);
+ auto projection = arrow::schema({schema_->GetFieldByName("f4"),
schema_->GetFieldByName("f8")});
+ auto predicate = PredicateBuilder::Equal(0, "f4", FieldType::INT,
Literal(300002));
+ int64_t cold_reads = 0;
+ std::shared_ptr<arrow::ChunkedArray> cold_result;
+ for (int32_t round = 0; round < 2; ++round) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input,
fs_->Open(file_path_));
+ ParquetReaderBuilder builder({{PARQUET_READ_ENABLE_PAGE_INDEX_FILTER,
"true"}}, 10);
+ builder.WithCache(cache);
+ ASSERT_OK_AND_ASSIGN(auto reader, builder.Build(input));
+ ArrowSchema c_schema;
+ ASSERT_TRUE(arrow::ExportSchema(*projection, &c_schema).ok());
+ ASSERT_OK(reader->SetReadSchema(&c_schema, predicate, std::nullopt));
+ ASSERT_OK_AND_ASSIGN(auto result,
+
paimon::test::ReadResultCollector::CollectResult(reader.get()));
+ ASSERT_EQ(1, result->length());
+ if (round == 0) {
+ cold_reads = cache->SupplierCallCount();
+ ASSERT_GT(cold_reads, 1); // Footer plus actual page-index ranges.
+ cold_result = result;
+ } else {
+ ASSERT_TRUE(result->Equals(cold_result));
+ ASSERT_EQ(cold_reads, cache->SupplierCallCount());
+ ASSERT_GT(cache->GetCount(), cold_reads);
+ }
+ }
+}
+
TEST_F(ParquetFileBatchReaderTest,
TestReadBinaryWrittenFromBinaryAndLargeBinary) {
auto check_binary_read_result = [&](const
std::shared_ptr<arrow::DataType>& write_type,
const std::string& file_name) {
diff --git a/src/paimon/format/parquet/parquet_input_stream.h
b/src/paimon/format/parquet/parquet_input_stream.h
new file mode 100644
index 00000000..fbf45f24
--- /dev/null
+++ b/src/paimon/format/parquet/parquet_input_stream.h
@@ -0,0 +1,127 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstring>
+#include <limits>
+#include <map>
+#include <memory>
+#include <string>
+
+#include "paimon/cache/cache.h"
+#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/memory/memory_pool.h"
+#include "parquet/metadata.h"
+#include "parquet/page_index.h"
+
+namespace paimon::parquet {
+
+inline MemorySegment AllocateParquetCacheSegment(int32_t size,
+ const
std::shared_ptr<MemoryPool>& pool) {
+ struct PooledBytes {
+ PooledBytes(int32_t size, const std::shared_ptr<MemoryPool>&
memory_pool)
+ : pool(memory_pool), bytes(size, memory_pool.get()) {}
+ // Destroy bytes before releasing their allocator.
+ std::shared_ptr<MemoryPool> pool;
+ Bytes bytes;
+ };
+ auto owner = std::make_shared<PooledBytes>(size, pool);
+ auto* bytes = &owner->bytes;
+ return MemorySegment::Wrap(std::shared_ptr<Bytes>(std::move(owner),
bytes));
+}
+
+// Cache immutable page-index bytes, not Arrow readers: the latter borrow their
+// input stream, reader properties and decryptor from the current file reader.
+class ParquetInputStream : public ArrowInputStreamAdapter {
+ public:
+ ParquetInputStream(const std::shared_ptr<paimon::InputStream>& input,
int64_t file_size,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool,
+ const std::shared_ptr<MemoryPool>& pool, const
std::shared_ptr<Cache>& cache,
+ const std::string& file_uri)
+ : ArrowInputStreamAdapter(input, file_size, arrow_pool),
+ pool_(pool),
+ cache_(cache),
+ file_uri_(file_uri) {}
+
+ // Called before the stream is published to the file reader. Only index
+ // ranges described by its footer are eligible; data pages remain uncached.
+ void SetPageIndexRanges(const ::parquet::FileMetaData& metadata) {
+ for (int32_t rg = 0; rg < metadata.num_row_groups(); ++rg) {
+ auto ranges =
::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup(
+ *metadata.RowGroup(rg), {});
+ for (const auto& range : {ranges.column_index,
ranges.offset_index}) {
+ if (range.has_value()) {
+ index_ranges_.emplace(range->offset, range->length);
+ }
+ }
+ }
+ }
+
+ using ArrowInputStreamAdapter::ReadAt;
+
+ arrow::Result<int64_t> ReadAt(int64_t position, int64_t nbytes, void* out)
override {
+ if (!cache_ || file_uri_.empty() || nbytes <= 0 ||
+ nbytes > std::numeric_limits<int32_t>::max()) {
+ return ArrowInputStreamAdapter::ReadAt(position, nbytes, out);
+ }
+ auto range = index_ranges_.upper_bound(position);
+ if (range == index_ranges_.begin()) {
+ return ArrowInputStreamAdapter::ReadAt(position, nbytes, out);
+ }
+ --range;
+ if (position - range->first > range->second ||
+ nbytes > range->second - (position - range->first)) {
+ return ArrowInputStreamAdapter::ReadAt(position, nbytes, out);
+ }
+ auto key = CacheKey::ForKind(file_uri_, position,
static_cast<int32_t>(nbytes),
+ CacheKind::DATA_FILE_FOOTER);
+ auto value = cache_->Get(
+ key,
+ [this, position,
+ nbytes](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ MemorySegment segment =
+ AllocateParquetCacheSegment(static_cast<int32_t>(nbytes),
pool_);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ int64_t size,
+ ArrowInputStreamAdapter::ReadAt(position, nbytes,
segment.MutableData()));
+ if (size != nbytes) {
+ return Status::IOError("Short read of Parquet page index");
+ }
+ return std::make_shared<CacheValue>(segment, CacheCallback());
+ });
+ if (!value.ok()) {
+ return ToArrowStatus(value.status());
+ }
+ if (!value.value() || value.value()->GetSegment().Size() != nbytes) {
+ return arrow::Status::IOError("Invalid Parquet page-index cache
value");
+ }
+ std::memcpy(out, value.value()->GetSegment().Data(), nbytes);
+ return nbytes;
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ std::shared_ptr<Cache> cache_;
+ std::string file_uri_;
+ std::map<int64_t, int64_t> index_ranges_;
+};
+
+} // namespace paimon::parquet
diff --git a/src/paimon/format/parquet/parquet_reader_builder.h
b/src/paimon/format/parquet/parquet_reader_builder.h
index e4d2ddb4..fad961ac 100644
--- a/src/paimon/format/parquet/parquet_reader_builder.h
+++ b/src/paimon/format/parquet/parquet_reader_builder.h
@@ -35,6 +35,7 @@
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/format/parquet/parquet_file_batch_reader.h"
#include "paimon/format/parquet/parquet_format_defs.h"
+#include "paimon/format/parquet/parquet_input_stream.h"
#include "paimon/format/read_hints.h"
#include "paimon/format/reader_builder.h"
#include "paimon/memory/memory_pool.h"
@@ -88,13 +89,14 @@ class ParquetReaderBuilder : public ReaderBuilder {
file_uri = std::move(file_uri_result).value();
}
}
- auto unique_input_stream =
- std::make_unique<ArrowInputStreamAdapter>(path, file_length,
arrow_pool_);
- auto storage_read_bytes = unique_input_stream->StorageReadBytes();
- std::shared_ptr<arrow::io::RandomAccessFile> input_stream(
- std::move(unique_input_stream));
+ auto input_stream = std::make_shared<ParquetInputStream>(path,
file_length, arrow_pool_,
+ pool_,
cache_, file_uri);
+ auto storage_read_bytes = input_stream->StorageReadBytes();
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<::parquet::FileMetaData>
file_metadata,
GetCachedParquetMetadata(input_stream,
file_uri, arrow_pool_));
+ if (file_metadata) {
+ input_stream->SetPageIndexRanges(*file_metadata);
+ }
return ParquetFileBatchReader::Create(
std::move(input_stream), options_, batch_size_,
std::move(file_metadata),
std::move(storage_read_bytes), arrow_pool_, hints_);
@@ -123,8 +125,7 @@ class ParquetReaderBuilder : public ReaderBuilder {
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Buffer>
metadata_footer,
output_stream->Finish());
- MemorySegment segment =
- MemorySegment::AllocateHeapMemory(metadata_footer->size(),
pool_.get());
+ MemorySegment segment =
AllocateParquetCacheSegment(metadata_footer->size(), pool_);
std::memcpy(segment.MutableData(), metadata_footer->data(),
metadata_footer->size());
return segment;
}
