This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 ec854151 chore(build): reformat arrow.diff patch sections (#217)
ec854151 is described below
commit ec8541515c0be97056271ac8cc272b6588dd94ea
Author: Yonghao Fang <[email protected]>
AuthorDate: Wed Aug 19 22:10:55 2026 +0800
chore(build): reformat arrow.diff patch sections (#217)
---
cmake_modules/arrow.diff | 1524 +++++++++++++++++++++++-----------------------
1 file changed, 766 insertions(+), 758 deletions(-)
diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff
index f86f36e8..ce63af35 100644
--- a/cmake_modules/arrow.diff
+++ b/cmake_modules/arrow.diff
@@ -1,22 +1,197 @@
-diff --git a/cpp/src/parquet/arrow/schema.cc b/cpp/src/parquet/arrow/schema.cc
-index ec3890a41f..943f69bb6c 100644
---- a/cpp/src/parquet/arrow/schema.cc
-+++ b/cpp/src/parquet/arrow/schema.cc
-@@ -178,7 +178,7 @@ static Status GetTimestampMetadata(const
::arrow::TimestampType& type,
+diff --git a/cpp/cmake_modules/BuildUtils.cmake
b/cpp/cmake_modules/BuildUtils.cmake
+index e7523add27..e079a1ad41 100644
+--- a/cpp/cmake_modules/BuildUtils.cmake
++++ b/cpp/cmake_modules/BuildUtils.cmake
+@@ -112,7 +112,7 @@ function(arrow_create_merged_static_lib output_target)
+ execute_process(COMMAND ${LIBTOOL_MACOS} -V
+ OUTPUT_VARIABLE LIBTOOL_V_OUTPUT
+ OUTPUT_STRIP_TRAILING_WHITESPACE)
+- if(NOT "${LIBTOOL_V_OUTPUT}" MATCHES ".*cctools-([0-9.]+).*")
++ if(NOT "${LIBTOOL_V_OUTPUT}" MATCHES ".*cctools(_ld)?-([0-9.]+).*")
+ message(FATAL_ERROR "libtool found appears to be the incompatible GNU
libtool: ${LIBTOOL_MACOS}"
+ )
+ endif()
+diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake
b/cpp/cmake_modules/ThirdpartyToolchain.cmake
+index 8cb3ec83f5..0765df8fa8 100644
+--- a/cpp/cmake_modules/ThirdpartyToolchain.cmake
++++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake
+@@ -983,6 +983,11 @@ if(CMAKE_TOOLCHAIN_FILE)
+ list(APPEND EP_COMMON_CMAKE_ARGS
-DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE})
+ endif()
- // The user is explicitly asking for Impala int96 encoding, there is no
- // logical type.
-- if (arrow_properties.support_deprecated_int96_timestamps()) {
-+ if (arrow_properties.support_deprecated_int96_timestamps() && target_unit
== ::arrow::TimeUnit::NANO) {
- *physical_type = ParquetType::INT96;
- return Status::OK();
- }
++# Compatibility with bundled dependencies that require old CMake versions.
++if(CMAKE_VERSION VERSION_GREATER_EQUAL "3.30")
++ list(APPEND EP_COMMON_CMAKE_ARGS -DCMAKE_POLICY_VERSION_MINIMUM=3.5)
++endif()
++
+ # and crosscompiling emulator (for try_run() )
+ if(CMAKE_CROSSCOMPILING_EMULATOR)
+ string(REPLACE ";" ${EP_LIST_SEPARATOR} EP_CMAKE_CROSSCOMPILING_EMULATOR
+@@ -1716,6 +1721,7 @@ macro(build_thrift)
+ -DWITH_JAVASCRIPT=OFF
+ -DWITH_LIBEVENT=OFF
+ -DWITH_NODEJS=OFF
++ -DWITH_OPENSSL=OFF
+ -DWITH_PYTHON=OFF
+ -DWITH_QT5=OFF
+ -DWITH_ZLIB=OFF)
+diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h
+index b36c38c6d4..f974a33073 100644
+--- a/cpp/src/arrow/io/interfaces.h
++++ b/cpp/src/arrow/io/interfaces.h
+@@ -210,7 +210,7 @@ class ARROW_EXPORT InputStream : virtual public
FileInterface, virtual public Re
+ /// \brief Advance or skip stream indicated number of bytes
+ /// \param[in] nbytes the number to move forward
+ /// \return Status
+- Status Advance(int64_t nbytes);
++ virtual Status Advance(int64_t nbytes);
+ /// \brief Return zero-copy string_view to upcoming bytes.
+ ///
diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc
-index 285e2a5973..aa6f92f077 100644
+index 285e2a5973..db919d7ef8 100644
--- a/cpp/src/parquet/arrow/reader.cc
+++ b/cpp/src/parquet/arrow/reader.cc
-@@ -1013,25 +1013,32 @@ Status FileReaderImpl::GetRecordBatchReader(const
std::vector<int>& row_groups,
+@@ -254,6 +254,11 @@ class FileReaderImpl : public FileReader {
+ return GetColumn(i, AllRowGroupsFactory(), out);
+ }
+
++ ::arrow::Status GetColumn(
++ int i, const std::vector<int>& column_indices,
++ FileColumnIteratorFactory iterator_factory,
++ std::unique_ptr<ColumnReader>* out) override;
++
+ Status GetSchema(std::shared_ptr<::arrow::Schema>* out) override {
+ return FromParquetSchema(reader_->metadata()->schema(),
reader_properties_,
+ reader_->metadata()->key_value_metadata(), out);
+@@ -493,10 +498,40 @@ class LeafReader : public ColumnReaderImpl {
+
+ ::arrow::Status BuildArray(int64_t length_upper_bound,
+ std::shared_ptr<::arrow::ChunkedArray>* out)
final {
++ if (!out_) {
++ BEGIN_PARQUET_CATCH_EXCEPTIONS
++ RETURN_NOT_OK(
++ TransferColumnData(record_reader_.get(), field_, descr_,
ctx_->pool, &out_));
++ END_PARQUET_CATCH_EXCEPTIONS
++ }
+ *out = out_;
+ return Status::OK();
+ }
+
++ std::vector<int> LeafColumnIndices() const final {
++ return {input_->column_index()};
++ }
++
++ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) final {
++ if (col_idx != input_->column_index()) return Status::OK();
++ BEGIN_PARQUET_CATCH_EXCEPTIONS
++ out_ = nullptr;
++ record_reader_->Reset();
++ record_reader_->Reserve(reserve);
++ return Status::OK();
++ END_PARQUET_CATCH_EXCEPTIONS
++ }
++
++ int64_t SkipRecords(int col_idx, int64_t num_records) final {
++ if (col_idx != input_->column_index() || num_records <= 0) return 0;
++ return record_reader_->SkipRecords(num_records);
++ }
++
++ int64_t ReadRecords(int col_idx, int64_t num_records) final {
++ if (col_idx != input_->column_index() || num_records <= 0) return 0;
++ return record_reader_->ReadRecords(num_records);
++ }
++
+ const std::shared_ptr<Field> field() override { return field_; }
+
+ private:
+@@ -532,6 +567,22 @@ class ExtensionReader : public ColumnReaderImpl {
+ return storage_reader_->LoadBatch(number_of_records);
+ }
+
++ std::vector<int> LeafColumnIndices() const final {
++ return storage_reader_->LeafColumnIndices();
++ }
++
++ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) final {
++ return storage_reader_->ResetLeaf(col_idx, reserve);
++ }
++
++ int64_t SkipRecords(int col_idx, int64_t num_records) final {
++ return storage_reader_->SkipRecords(col_idx, num_records);
++ }
++
++ int64_t ReadRecords(int col_idx, int64_t num_records) final {
++ return storage_reader_->ReadRecords(col_idx, num_records);
++ }
++
+ Status BuildArray(int64_t length_upper_bound,
+ std::shared_ptr<ChunkedArray>* out) override {
+ std::shared_ptr<ChunkedArray> storage;
+@@ -576,6 +627,22 @@ class ListReader : public ColumnReaderImpl {
+ return item_reader_->LoadBatch(number_of_records);
+ }
+
++ std::vector<int> LeafColumnIndices() const final {
++ return item_reader_->LeafColumnIndices();
++ }
++
++ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) final {
++ return item_reader_->ResetLeaf(col_idx, reserve);
++ }
++
++ int64_t SkipRecords(int col_idx, int64_t num_records) final {
++ return item_reader_->SkipRecords(col_idx, num_records);
++ }
++
++ int64_t ReadRecords(int col_idx, int64_t num_records) final {
++ return item_reader_->ReadRecords(col_idx, num_records);
++ }
++
+ virtual ::arrow::Result<std::shared_ptr<ChunkedArray>> AssembleArray(
+ std::shared_ptr<ArrayData> data) {
+ if (field_->type()->id() == ::arrow::Type::MAP) {
+@@ -709,6 +776,39 @@ class PARQUET_NO_EXPORT StructReader : public
ColumnReaderImpl {
+ }
+ return Status::OK();
+ }
++
++ std::vector<int> LeafColumnIndices() const override {
++ std::vector<int> indices;
++ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
++ std::vector<int> child_indices = reader->LeafColumnIndices();
++ indices.insert(indices.end(), child_indices.begin(),
child_indices.end());
++ }
++ return indices;
++ }
++
++ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) override {
++ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
++ RETURN_NOT_OK(reader->ResetLeaf(col_idx, reserve));
++ }
++ return Status::OK();
++ }
++
++ int64_t SkipRecords(int col_idx, int64_t num_records) override {
++ int64_t skipped = 0;
++ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
++ skipped += reader->SkipRecords(col_idx, num_records);
++ }
++ return skipped;
++ }
++
++ int64_t ReadRecords(int col_idx, int64_t num_records) override {
++ int64_t read = 0;
++ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
++ read += reader->ReadRecords(col_idx, num_records);
++ }
++ return read;
++ }
++
+ Status BuildArray(int64_t length_upper_bound,
+ std::shared_ptr<ChunkedArray>* out) override;
+ Status GetDefLevels(const int16_t** data, int64_t* length) override;
+@@ -1013,25 +1113,32 @@ Status FileReaderImpl::GetRecordBatchReader(const
std::vector<int>& row_groups,
return Status::OK();
}
@@ -55,581 +230,49 @@ index 285e2a5973..aa6f92f077 100644
RETURN_NOT_OK(::arrow::internal::OptionalParallelFor(
reader_properties_.use_threads(),
static_cast<int>(readers.size()),
-diff --git a/cpp/src/parquet/arrow/writer.cc b/cpp/src/parquet/arrow/writer.cc
-index 4fd7ef1b47..87326a54f1 100644
---- a/cpp/src/parquet/arrow/writer.cc
-+++ b/cpp/src/parquet/arrow/writer.cc
-@@ -314,6 +314,14 @@ class FileWriterImpl : public FileWriter {
- return Status::OK();
- }
+@@ -1224,6 +1331,23 @@ Status FileReaderImpl::GetColumn(int i,
FileColumnIteratorFactory iterator_facto
+ return Status::OK();
+ }
-+ int64_t GetBufferedSize() override {
-+ if (row_group_writer_ == nullptr) {
-+ return 0;
-+ }
-+ return row_group_writer_->total_compressed_bytes() +
-+ row_group_writer_->total_compressed_bytes_written();
-+ }
++::arrow::Status FileReaderImpl::GetColumn(
++ int i, const std::vector<int>& column_indices,
++ FileColumnIteratorFactory iterator_factory,
++ std::unique_ptr<ColumnReader>* out) {
++ RETURN_NOT_OK(BoundsCheckColumn(i));
++ auto ctx = std::make_shared<ReaderContext>();
++ ctx->reader = reader_.get();
++ ctx->pool = pool_;
++ ctx->iterator_factory = iterator_factory;
++ ctx->filter_leaves = true;
++ ctx->included_leaves = VectorToSharedSet(column_indices);
++ std::unique_ptr<ColumnReaderImpl> result;
++ RETURN_NOT_OK(GetReader(manifest_.schema_fields[i], ctx, &result));
++ *out = std::move(result);
++ return Status::OK();
++}
+
- Status Close() override {
- if (!closed_) {
- // Make idempotent
-@@ -418,10 +426,13 @@ class FileWriterImpl : public FileWriter {
+ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
+ const std::vector<int>& column_indices,
+ std::shared_ptr<Table>* out) {
+diff --git a/cpp/src/parquet/arrow/reader.h b/cpp/src/parquet/arrow/reader.h
+index 6e46ca43f7..e86ff0ef52 100644
+--- a/cpp/src/parquet/arrow/reader.h
++++ b/cpp/src/parquet/arrow/reader.h
+@@ -21,6 +21,7 @@
+ // N.B. we don't include async_generator.h as it's relatively heavy
+ #include <functional>
+ #include <memory>
++#include <utility>
+ #include <vector>
- // Max number of rows allowed in a row group.
- const int64_t max_row_group_length =
this->properties().max_row_group_length();
-+ const int64_t max_row_group_size =
this->properties().max_row_group_size();
+ #include "parquet/file_reader.h"
+@@ -48,9 +49,13 @@ namespace arrow {
- // Initialize a new buffered row group writer if necessary.
- if (row_group_writer_ == nullptr || !row_group_writer_->buffered() ||
-- row_group_writer_->num_rows() >= max_row_group_length) {
-+ row_group_writer_->num_rows() >= max_row_group_length ||
-+ (row_group_writer_->total_compressed_bytes_written() +
-+ row_group_writer_->total_compressed_bytes() >= max_row_group_size)) {
- RETURN_NOT_OK(NewBufferedRowGroup());
- }
-
-diff --git a/cpp/src/parquet/arrow/writer.h b/cpp/src/parquet/arrow/writer.h
-index 4a1a033a7b..0f13d05e44 100644
---- a/cpp/src/parquet/arrow/writer.h
-+++ b/cpp/src/parquet/arrow/writer.h
-@@ -138,6 +138,9 @@ class PARQUET_EXPORT FileWriter {
- /// option in this case.
- virtual ::arrow::Status WriteRecordBatch(const ::arrow::RecordBatch& batch)
= 0;
-
-+ /// \brief Return the buffered size in bytes.
-+ virtual int64_t GetBufferedSize() = 0;
-+
- /// \brief Write the footer and close the file.
- virtual ::arrow::Status Close() = 0;
- virtual ~FileWriter();
-diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h
-index 4d3acb491e..3906ff3c59 100644
---- a/cpp/src/parquet/properties.h
-+++ b/cpp/src/parquet/properties.h
-@@ -139,6 +139,7 @@ static constexpr bool DEFAULT_IS_DICTIONARY_ENABLED = true;
- static constexpr int64_t DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT =
kDefaultDataPageSize;
- static constexpr int64_t DEFAULT_WRITE_BATCH_SIZE = 1024;
- static constexpr int64_t DEFAULT_MAX_ROW_GROUP_LENGTH = 1024 * 1024;
-+static constexpr int64_t DEFAULT_MAX_ROW_GROUP_SIZE = 128 * 1024 * 1024;
- static constexpr bool DEFAULT_ARE_STATISTICS_ENABLED = true;
- static constexpr int64_t DEFAULT_MAX_STATISTICS_SIZE = 4096;
- static constexpr Encoding::type DEFAULT_ENCODING = Encoding::UNKNOWN;
-@@ -232,6 +233,7 @@ class PARQUET_EXPORT WriterProperties {
- dictionary_pagesize_limit_(DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT),
- write_batch_size_(DEFAULT_WRITE_BATCH_SIZE),
- max_row_group_length_(DEFAULT_MAX_ROW_GROUP_LENGTH),
-+ max_row_group_size_(DEFAULT_MAX_ROW_GROUP_SIZE),
- pagesize_(kDefaultDataPageSize),
- version_(ParquetVersion::PARQUET_2_6),
- data_page_version_(ParquetDataPageVersion::V1),
-@@ -244,6 +246,7 @@ class PARQUET_EXPORT WriterProperties {
- dictionary_pagesize_limit_(properties.dictionary_pagesize_limit()),
- write_batch_size_(properties.write_batch_size()),
- max_row_group_length_(properties.max_row_group_length()),
-+ max_row_group_size_(properties.max_row_group_size()),
- pagesize_(properties.data_pagesize()),
- version_(properties.version()),
- data_page_version_(properties.data_page_version()),
-@@ -321,6 +324,13 @@ class PARQUET_EXPORT WriterProperties {
- return this;
- }
-
-+ /// Specify the max bytes size to put in a single row group.
-+ /// Default 128 M.
-+ Builder* max_row_group_size(int64_t max_row_group_size) {
-+ max_row_group_size_ = max_row_group_size;
-+ return this;
-+ }
-+
- /// Specify the data page size.
- /// Default 1MB.
- Builder* data_pagesize(int64_t pg_size) {
-@@ -664,7 +674,7 @@ class PARQUET_EXPORT WriterProperties {
-
- return std::shared_ptr<WriterProperties>(new WriterProperties(
- pool_, dictionary_pagesize_limit_, write_batch_size_,
max_row_group_length_,
-- pagesize_, version_, created_by_, page_checksum_enabled_,
-+ max_row_group_size_, pagesize_, version_, created_by_,
page_checksum_enabled_,
- std::move(file_encryption_properties_), default_column_properties_,
- column_properties, data_page_version_, store_decimal_as_integer_,
- std::move(sorting_columns_)));
-@@ -675,6 +685,7 @@ class PARQUET_EXPORT WriterProperties {
- int64_t dictionary_pagesize_limit_;
- int64_t write_batch_size_;
- int64_t max_row_group_length_;
-+ int64_t max_row_group_size_;
- int64_t pagesize_;
- ParquetVersion::type version_;
- ParquetDataPageVersion data_page_version_;
-@@ -705,6 +716,8 @@ class PARQUET_EXPORT WriterProperties {
-
- inline int64_t max_row_group_length() const { return max_row_group_length_;
}
-
-+ inline int64_t max_row_group_size() const { return max_row_group_size_; }
-+
- inline int64_t data_pagesize() const { return pagesize_; }
-
- inline ParquetDataPageVersion data_page_version() const {
-@@ -810,7 +823,7 @@ class PARQUET_EXPORT WriterProperties {
- private:
- explicit WriterProperties(
- MemoryPool* pool, int64_t dictionary_pagesize_limit, int64_t
write_batch_size,
-- int64_t max_row_group_length, int64_t pagesize, ParquetVersion::type
version,
-+ int64_t max_row_group_length, int64_t max_row_group_size, int64_t
pagesize, ParquetVersion::type version,
- const std::string& created_by, bool page_write_checksum_enabled,
- std::shared_ptr<FileEncryptionProperties> file_encryption_properties,
- const ColumnProperties& default_column_properties,
-@@ -821,6 +834,7 @@ class PARQUET_EXPORT WriterProperties {
- dictionary_pagesize_limit_(dictionary_pagesize_limit),
- write_batch_size_(write_batch_size),
- max_row_group_length_(max_row_group_length),
-+ max_row_group_size_(max_row_group_size),
- pagesize_(pagesize),
- parquet_data_page_version_(data_page_version),
- parquet_version_(version),
-@@ -836,6 +850,7 @@ class PARQUET_EXPORT WriterProperties {
- int64_t dictionary_pagesize_limit_;
- int64_t write_batch_size_;
- int64_t max_row_group_length_;
-+ int64_t max_row_group_size_;
- int64_t pagesize_;
- ParquetDataPageVersion parquet_data_page_version_;
- ParquetVersion::type parquet_version_;
-
---- a/cpp/src/parquet/file_reader.h
-+++ b/cpp/src/parquet/file_reader.h
-@@ -210,6 +210,17 @@
- ::arrow::Future<> WhenBuffered(const std::vector<int>& row_groups,
- const std::vector<int>& column_indices)
const;
-
-+ /// Pre-buffer arbitrary byte ranges (e.g., page-level ranges from
OffsetIndex).
-+ /// Unlike PreBuffer(), this does NOT set the column bitmap, so
-+ /// GetColumnPageReader will use CachedInputStream (page-level cache path).
-+ void PreBufferRanges(const std::vector<::arrow::io::ReadRange>& ranges,
-+ const ::arrow::io::IOContext& ctx,
-+ const ::arrow::io::CacheOptions& options);
-+
-+ /// Wait for arbitrary byte ranges to be pre-buffered.
-+ ::arrow::Future<> WhenBufferedRanges(
-+ const std::vector<::arrow::io::ReadRange>& ranges) const;
-+
- private:
- // Holds a pointer to an instance of Contents implementation
- std::unique_ptr<Contents> contents_;
-
---- a/cpp/src/parquet/file_reader.cc
-+++ b/cpp/src/parquet/file_reader.cc
-@@ -207,6 +207,117 @@
- return {col_start, col_length};
- }
-
-+// CachedInputStream: InputStream adapter that reads through ReadRangeCache
with
-+// zero-cost skip for non-cached pages. Used for page-level caching where only
-+// specific pages are pre-buffered.
-+//
-+// Key behavior:
-+// - Read(): On cache hit, returns cached data. On cache miss, returns
zero-filled
-+// buffer (zero I/O). This makes InputStream::Advance() (which calls Read()
and
-+// discards) effectively free for skipped pages.
-+// - Peek(): Always falls back to source on cache miss, because PageReader
uses
-+// Peek() to read Thrift page headers (~30 bytes) which must have real data.
-+class CachedInputStream : public ::arrow::io::InputStream {
-+ public:
-+ CachedInputStream(
-+ std::shared_ptr<::arrow::io::internal::ReadRangeCache> cache,
-+ std::shared_ptr<ArrowInputFile> source,
-+ int64_t offset, int64_t length)
-+ : cache_(std::move(cache)),
-+ source_(std::move(source)),
-+ base_offset_(offset),
-+ length_(length) {}
-+
-+ ::arrow::Status Close() override {
-+ closed_ = true;
-+ return ::arrow::Status::OK();
-+ }
-+
-+ bool closed() const override { return closed_; }
-+
-+ ::arrow::Result<int64_t> Tell() const override { return position_; }
-+
-+ ::arrow::Result<std::string_view> Peek(int64_t nbytes) override {
-+ int64_t to_read = std::min(nbytes, length_ - position_);
-+ if (to_read <= 0) {
-+ return std::string_view();
-+ }
-+ ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
-+ auto result = cache_->Read(range);
-+ if (result.ok()) {
-+ peek_buffer_ = *result;
-+ } else {
-+ // Peek is used for Thrift page headers (~30 bytes) — must read real
data
-+ ARROW_ASSIGN_OR_RAISE(peek_buffer_,
-+ source_->ReadAt(range.offset, range.length));
-+ }
-+ return std::string_view(
-+ reinterpret_cast<const char*>(peek_buffer_->data()),
-+ static_cast<size_t>(peek_buffer_->size()));
-+ }
-+
-+ ::arrow::Result<int64_t> Read(int64_t nbytes, void* out) override {
-+ int64_t to_read = std::min(nbytes, length_ - position_);
-+ if (to_read <= 0) return 0;
-+ ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
-+ auto result = cache_->Read(range);
-+ if (result.ok()) {
-+ auto& buf = *result;
-+ memcpy(out, buf->data(), static_cast<size_t>(buf->size()));
-+ position_ += buf->size();
-+ return buf->size();
-+ }
-+ // Cache miss: fall back to real I/O from source
-+ ARROW_ASSIGN_OR_RAISE(auto buf, source_->ReadAt(range.offset,
range.length));
-+ memcpy(out, buf->data(), static_cast<size_t>(buf->size()));
-+ position_ += buf->size();
-+ return buf->size();
-+ }
-+
-+ ::arrow::Result<std::shared_ptr<::arrow::Buffer>> Read(int64_t nbytes)
override {
-+ int64_t to_read = std::min(nbytes, length_ - position_);
-+ if (to_read <= 0) {
-+ return std::make_shared<::arrow::Buffer>(nullptr, 0);
-+ }
-+ ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
-+ auto result = cache_->Read(range);
-+ if (result.ok()) {
-+ position_ += (*result)->size();
-+ return *result;
-+ }
-+ // Cache miss: fall back to real I/O from source
-+ ARROW_ASSIGN_OR_RAISE(auto buf, source_->ReadAt(range.offset,
range.length));
-+ position_ += buf->size();
-+ return std::shared_ptr<::arrow::Buffer>(std::move(buf));
-+ }
-+
-+ // Override Advance to avoid real I/O for skipped pages.
-+ // The default InputStream::Advance() calls Read() and discards the result,
-+ // which would trigger source_->ReadAt() on cache miss — defeating
page-level
-+ // I/O skipping via data_page_filter. Since Advance() is only used to skip
-+ // over data that will not be consumed, we can safely just move the
position.
-+ ::arrow::Status Advance(int64_t nbytes) override {
-+ if (nbytes <= 0) {
-+ return ::arrow::Status::OK();
-+ }
-+ int64_t remaining = length_ - position_;
-+ if (remaining <= 0) {
-+ return ::arrow::Status::OK();
-+ }
-+ position_ += std::min(nbytes, remaining);
-+ return ::arrow::Status::OK();
-+ }
-+
-+ private:
-+ std::shared_ptr<::arrow::io::internal::ReadRangeCache> cache_;
-+ std::shared_ptr<ArrowInputFile> source_;
-+ int64_t base_offset_;
-+ int64_t length_;
-+ int64_t position_ = 0;
-+ bool closed_ = false;
-+ std::shared_ptr<::arrow::Buffer> peek_buffer_;
-+};
-+
- // RowGroupReader::Contents implementation for the Parquet file specification
- class SerializedRowGroup : public RowGroupReader::Contents {
- public:
-@@ -242,6 +343,11 @@
- // segments.
- PARQUET_ASSIGN_OR_THROW(auto buffer, cached_source_->Read(col_range));
- stream = std::make_shared<::arrow::io::BufferReader>(buffer);
-+ } else if (cached_source_) {
-+ // Page-level caching: read through cache with fallback to source.
-+ // Advance() is zero-cost for skipped pages via data_page_filter.
-+ stream = std::make_shared<CachedInputStream>(
-+ cached_source_, source_, col_range.offset, col_range.length);
- } else {
- stream = properties_.GetStream(source_, col_range.offset,
col_range.length);
- }
-@@ -417,6 +523,26 @@
- return cached_source_->WaitFor(ranges);
- }
-
-+ void PreBufferRanges(const std::vector<::arrow::io::ReadRange>& ranges,
-+ const ::arrow::io::IOContext& ctx,
-+ const ::arrow::io::CacheOptions& options) {
-+ cached_source_ =
-+ std::make_shared<::arrow::io::internal::ReadRangeCache>(source_, ctx,
options);
-+ // Do NOT set prebuffered_column_chunks_ bitmap — GetColumnPageReader will
-+ // use CachedInputStream path instead of full-chunk BufferReader path.
-+ prebuffered_column_chunks_.clear();
-+ PARQUET_THROW_NOT_OK(cached_source_->Cache(ranges));
-+ }
-+
-+ ::arrow::Future<> WhenBufferedRanges(
-+ const std::vector<::arrow::io::ReadRange>& ranges) const {
-+ if (!cached_source_) {
-+ return ::arrow::Status::Invalid(
-+ "Must call PreBufferRanges before WhenBufferedRanges");
-+ }
-+ return cached_source_->WaitFor(ranges);
-+ }
-+
- // Metadata/footer parsing. Divided up to separate sync/async paths, and to
use
- // exceptions for error handling (with the async path converting to
Future/Status).
-
-@@ -911,6 +1037,22 @@
- return file->WhenBuffered(row_groups, column_indices);
- }
-
-+void ParquetFileReader::PreBufferRanges(
-+ const std::vector<::arrow::io::ReadRange>& ranges,
-+ const ::arrow::io::IOContext& ctx,
-+ const ::arrow::io::CacheOptions& options) {
-+ SerializedFile* file =
-+ ::arrow::internal::checked_cast<SerializedFile*>(contents_.get());
-+ file->PreBufferRanges(ranges, ctx, options);
-+}
-+
-+::arrow::Future<> ParquetFileReader::WhenBufferedRanges(
-+ const std::vector<::arrow::io::ReadRange>& ranges) const {
-+ SerializedFile* file =
-+ ::arrow::internal::checked_cast<SerializedFile*>(contents_.get());
-+ return file->WhenBufferedRanges(ranges);
-+}
-+
- // ----------------------------------------------------------------------
- // File metadata helpers
-
-diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake
b/cpp/cmake_modules/ThirdpartyToolchain.cmake
---- a/cpp/cmake_modules/ThirdpartyToolchain.cmake
-+++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake
-@@ -981,6 +981,11 @@ if(CMAKE_TOOLCHAIN_FILE)
- list(APPEND EP_COMMON_CMAKE_ARGS
-DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE})
- endif()
-
-+# Compatibility with bundled dependencies that require old CMake versions.
-+if(CMAKE_VERSION VERSION_GREATER_EQUAL "3.30")
-+ list(APPEND EP_COMMON_CMAKE_ARGS -DCMAKE_POLICY_VERSION_MINIMUM=3.5)
-+endif()
-+
- # and crosscompiling emulator (for try_run() )
- if(CMAKE_CROSSCOMPILING_EMULATOR)
- string(REPLACE ";" ${EP_LIST_SEPARATOR} EP_CMAKE_CROSSCOMPILING_EMULATOR
-@@ -1720,6 +1725,7 @@ macro(build_thrift)
- -DWITH_JAVASCRIPT=OFF
- -DWITH_LIBEVENT=OFF
- -DWITH_NODEJS=OFF
-+ -DWITH_OPENSSL=OFF
- -DWITH_PYTHON=OFF
- -DWITH_QT5=OFF
- -DWITH_ZLIB=OFF)
-diff --git a/cpp/cmake_modules/BuildUtils.cmake
b/cpp/cmake_modules/BuildUtils.cmake
---- a/cpp/cmake_modules/BuildUtils.cmake
-+++ b/cpp/cmake_modules/BuildUtils.cmake
-@@ -112,7 +112,7 @@ function(arrow_create_merged_static_lib output_target)
- execute_process(COMMAND ${LIBTOOL_MACOS} -V
- OUTPUT_VARIABLE LIBTOOL_V_OUTPUT
- OUTPUT_STRIP_TRAILING_WHITESPACE)
-- if(NOT "${LIBTOOL_V_OUTPUT}" MATCHES ".*cctools-([0-9.]+).*")
-+ if(NOT "${LIBTOOL_V_OUTPUT}" MATCHES ".*cctools(_ld)?-([0-9.]+).*")
- message(FATAL_ERROR "libtool found appears to be the incompatible GNU
libtool: ${LIBTOOL_MACOS}"
- )
- endif()
-
-diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h
---- a/cpp/src/arrow/io/interfaces.h
-+++ b/cpp/src/arrow/io/interfaces.h
-@@ -210,7 +210,7 @@
- /// \brief Advance or skip stream indicated number of bytes
- /// \param[in] nbytes the number to move forward
- /// \return Status
-- Status Advance(int64_t nbytes);
-+ virtual Status Advance(int64_t nbytes);
-
- /// \brief Return zero-copy string_view to upcoming bytes.
- ///
---- a/cpp/src/parquet/arrow/reader.cc
-+++ b/cpp/src/parquet/arrow/reader.cc
-@@ -254,6 +254,11 @@
- return GetColumn(i, AllRowGroupsFactory(), out);
- }
-
-+ ::arrow::Status GetColumn(
-+ int i, const std::vector<int>& column_indices,
-+ FileColumnIteratorFactory iterator_factory,
-+ std::unique_ptr<ColumnReader>* out) override;
-+
- Status GetSchema(std::shared_ptr<::arrow::Schema>* out) override {
- return FromParquetSchema(reader_->metadata()->schema(),
reader_properties_,
- reader_->metadata()->key_value_metadata(), out);
-@@ -493,10 +498,40 @@
-
- ::arrow::Status BuildArray(int64_t length_upper_bound,
- std::shared_ptr<::arrow::ChunkedArray>* out)
final {
-+ if (!out_) {
-+ BEGIN_PARQUET_CATCH_EXCEPTIONS
-+ RETURN_NOT_OK(
-+ TransferColumnData(record_reader_.get(), field_, descr_,
ctx_->pool, &out_));
-+ END_PARQUET_CATCH_EXCEPTIONS
-+ }
- *out = out_;
- return Status::OK();
- }
-
-+ std::vector<int> LeafColumnIndices() const final {
-+ return {input_->column_index()};
-+ }
-+
-+ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) final {
-+ if (col_idx != input_->column_index()) return Status::OK();
-+ BEGIN_PARQUET_CATCH_EXCEPTIONS
-+ out_ = nullptr;
-+ record_reader_->Reset();
-+ record_reader_->Reserve(reserve);
-+ return Status::OK();
-+ END_PARQUET_CATCH_EXCEPTIONS
-+ }
-+
-+ int64_t SkipRecords(int col_idx, int64_t num_records) final {
-+ if (col_idx != input_->column_index() || num_records <= 0) return 0;
-+ return record_reader_->SkipRecords(num_records);
-+ }
-+
-+ int64_t ReadRecords(int col_idx, int64_t num_records) final {
-+ if (col_idx != input_->column_index() || num_records <= 0) return 0;
-+ return record_reader_->ReadRecords(num_records);
-+ }
-+
- const std::shared_ptr<Field> field() override { return field_; }
-
- private:
-@@ -532,6 +567,22 @@
- return storage_reader_->LoadBatch(number_of_records);
- }
-
-+ std::vector<int> LeafColumnIndices() const final {
-+ return storage_reader_->LeafColumnIndices();
-+ }
-+
-+ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) final {
-+ return storage_reader_->ResetLeaf(col_idx, reserve);
-+ }
-+
-+ int64_t SkipRecords(int col_idx, int64_t num_records) final {
-+ return storage_reader_->SkipRecords(col_idx, num_records);
-+ }
-+
-+ int64_t ReadRecords(int col_idx, int64_t num_records) final {
-+ return storage_reader_->ReadRecords(col_idx, num_records);
-+ }
-+
- Status BuildArray(int64_t length_upper_bound,
- std::shared_ptr<ChunkedArray>* out) override {
- std::shared_ptr<ChunkedArray> storage;
-@@ -576,6 +627,22 @@
- return item_reader_->LoadBatch(number_of_records);
- }
-
-+ std::vector<int> LeafColumnIndices() const final {
-+ return item_reader_->LeafColumnIndices();
-+ }
-+
-+ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) final {
-+ return item_reader_->ResetLeaf(col_idx, reserve);
-+ }
-+
-+ int64_t SkipRecords(int col_idx, int64_t num_records) final {
-+ return item_reader_->SkipRecords(col_idx, num_records);
-+ }
-+
-+ int64_t ReadRecords(int col_idx, int64_t num_records) final {
-+ return item_reader_->ReadRecords(col_idx, num_records);
-+ }
-+
- virtual ::arrow::Result<std::shared_ptr<ChunkedArray>> AssembleArray(
- std::shared_ptr<ArrayData> data) {
- if (field_->type()->id() == ::arrow::Type::MAP) {
-@@ -709,6 +776,39 @@
- }
- return Status::OK();
- }
-+
-+ std::vector<int> LeafColumnIndices() const override {
-+ std::vector<int> indices;
-+ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
-+ std::vector<int> child_indices = reader->LeafColumnIndices();
-+ indices.insert(indices.end(), child_indices.begin(),
child_indices.end());
-+ }
-+ return indices;
-+ }
-+
-+ ::arrow::Status ResetLeaf(int col_idx, int64_t reserve) override {
-+ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
-+ RETURN_NOT_OK(reader->ResetLeaf(col_idx, reserve));
-+ }
-+ return Status::OK();
-+ }
-+
-+ int64_t SkipRecords(int col_idx, int64_t num_records) override {
-+ int64_t skipped = 0;
-+ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
-+ skipped += reader->SkipRecords(col_idx, num_records);
-+ }
-+ return skipped;
-+ }
-+
-+ int64_t ReadRecords(int col_idx, int64_t num_records) override {
-+ int64_t read = 0;
-+ for (const std::unique_ptr<ColumnReaderImpl>& reader : children_) {
-+ read += reader->ReadRecords(col_idx, num_records);
-+ }
-+ return read;
-+ }
-+
- Status BuildArray(int64_t length_upper_bound,
- std::shared_ptr<ChunkedArray>* out) override;
- Status GetDefLevels(const int16_t** data, int64_t* length) override;
-@@ -1228,6 +1328,23 @@
- std::unique_ptr<ColumnReaderImpl> result;
- RETURN_NOT_OK(GetReader(manifest_.schema_fields[i], ctx, &result));
- *out = std::move(result);
-+ return Status::OK();
-+}
-+
-+::arrow::Status FileReaderImpl::GetColumn(
-+ int i, const std::vector<int>& column_indices,
-+ FileColumnIteratorFactory iterator_factory,
-+ std::unique_ptr<ColumnReader>* out) {
-+ RETURN_NOT_OK(BoundsCheckColumn(i));
-+ auto ctx = std::make_shared<ReaderContext>();
-+ ctx->reader = reader_.get();
-+ ctx->pool = pool_;
-+ ctx->iterator_factory = iterator_factory;
-+ ctx->filter_leaves = true;
-+ ctx->included_leaves = VectorToSharedSet(column_indices);
-+ std::unique_ptr<ColumnReaderImpl> result;
-+ RETURN_NOT_OK(GetReader(manifest_.schema_fields[i], ctx, &result));
-+ *out = std::move(result);
- return Status::OK();
- }
-
---- a/cpp/src/parquet/arrow/reader.h
-+++ b/cpp/src/parquet/arrow/reader.h
-@@ -21,6 +21,7 @@
- // N.B. we don't include async_generator.h as it's relatively heavy
- #include <functional>
- #include <memory>
-+#include <utility>
- #include <vector>
-
- #include "parquet/file_reader.h"
-@@ -48,9 +49,13 @@
-
- class ColumnChunkReader;
- class ColumnReader;
-+class FileColumnIterator;
- struct SchemaManifest;
- class RowGroupReader;
+ class ColumnChunkReader;
+ class ColumnReader;
++class FileColumnIterator;
+ struct SchemaManifest;
+ class RowGroupReader;
+using FileColumnIteratorFactory =
+ std::function<FileColumnIterator*(int, ParquetFileReader*)>;
@@ -637,7 +280,7 @@ diff --git a/cpp/src/arrow/io/interfaces.h
b/cpp/src/arrow/io/interfaces.h
/// \brief Arrow read adapter class for deserializing Parquet files as Arrow
row batches.
///
/// This interfaces caters for different use cases and thus provides different
-@@ -136,6 +141,27 @@
+@@ -136,6 +141,27 @@ class PARQUET_EXPORT FileReader {
// The indicated column index is relative to the schema
virtual ::arrow::Status GetColumn(int i, std::unique_ptr<ColumnReader>*
out) = 0;
@@ -665,7 +308,7 @@ diff --git a/cpp/src/arrow/io/interfaces.h
b/cpp/src/arrow/io/interfaces.h
/// \brief Return arrow schema for all the columns.
virtual ::arrow::Status GetSchema(std::shared_ptr<::arrow::Schema>* out) =
0;
-@@ -316,6 +342,43 @@
+@@ -316,6 +342,43 @@ class PARQUET_EXPORT ColumnReader {
// the data available in the file.
virtual ::arrow::Status NextBatch(int64_t batch_size,
std::shared_ptr<::arrow::ChunkedArray>*
out) = 0;
@@ -698,50 +341,269 @@ diff --git a/cpp/src/arrow/io/interfaces.h
b/cpp/src/arrow/io/interfaces.h
+ /// error; callers convert it to Status at the public boundary.
+ virtual int64_t ReadRecords(int col_idx, int64_t num_records) { return 0; }
+
-+ /// \brief Build the Arrow array from previously loaded data.
-+ /// For leaf readers, calls TransferColumnData if not already done.
-+ /// For nested readers, assembles the nested array from child arrays.
-+ virtual ::arrow::Status BuildArray(
-+ int64_t length_upper_bound,
-+ std::shared_ptr<::arrow::ChunkedArray>* out) {
-+ return ::arrow::Status::NotImplemented("BuildArray not implemented");
-+ }
- };
-
- /// \brief Experimental helper class for bindings (like Python) that struggle
---- a/cpp/src/parquet/arrow/reader_internal.h
-+++ b/cpp/src/parquet/arrow/reader_internal.h
-@@ -26,6 +26,7 @@
- #include <utility>
- #include <vector>
-
-+#include "parquet/arrow/reader.h"
- #include "parquet/arrow/schema.h"
- #include "parquet/column_reader.h"
- #include "parquet/file_reader.h"
-@@ -70,7 +71,10 @@
++ /// \brief Build the Arrow array from previously loaded data.
++ /// For leaf readers, calls TransferColumnData if not already done.
++ /// For nested readers, assembles the nested array from child arrays.
++ virtual ::arrow::Status BuildArray(
++ int64_t length_upper_bound,
++ std::shared_ptr<::arrow::ChunkedArray>* out) {
++ return ::arrow::Status::NotImplemented("BuildArray not implemented");
++ }
+ };
+
+ /// \brief Experimental helper class for bindings (like Python) that struggle
+diff --git a/cpp/src/parquet/arrow/reader_internal.h
b/cpp/src/parquet/arrow/reader_internal.h
+index cf9dbb8657..9216f18289 100644
+--- a/cpp/src/parquet/arrow/reader_internal.h
++++ b/cpp/src/parquet/arrow/reader_internal.h
+@@ -26,6 +26,7 @@
+ #include <utility>
+ #include <vector>
+
++#include "parquet/arrow/reader.h"
+ #include "parquet/arrow/schema.h"
+ #include "parquet/column_reader.h"
+ #include "parquet/file_reader.h"
+@@ -70,7 +71,10 @@ class FileColumnIterator {
+
+ virtual ~FileColumnIterator() {}
+
+- std::unique_ptr<::parquet::PageReader> NextChunk() {
++ /// \brief Fetch the PageReader for the next row group in this iterator's
++ /// range. Virtual so subclasses can decorate the returned PageReader, e.g.
++ /// to install a data_page_filter for I/O-level page skipping.
++ virtual std::unique_ptr<::parquet::PageReader> NextChunk() {
+ if (row_groups_.empty()) {
+ return nullptr;
+ }
+@@ -95,9 +99,6 @@ class FileColumnIterator {
+ std::deque<int> row_groups_;
+ };
+
+-using FileColumnIteratorFactory =
+- std::function<FileColumnIterator*(int, ParquetFileReader*)>;
+-
+ Status TransferColumnData(::parquet::internal::RecordReader* reader,
+ const std::shared_ptr<::arrow::Field>& value_field,
+ const ColumnDescriptor* descr, ::arrow::MemoryPool*
pool,
+diff --git a/cpp/src/parquet/arrow/schema.cc b/cpp/src/parquet/arrow/schema.cc
+index ec3890a41f..943f69bb6c 100644
+--- a/cpp/src/parquet/arrow/schema.cc
++++ b/cpp/src/parquet/arrow/schema.cc
+@@ -178,7 +178,7 @@ static Status GetTimestampMetadata(const
::arrow::TimestampType& type,
+
+ // The user is explicitly asking for Impala int96 encoding, there is no
+ // logical type.
+- if (arrow_properties.support_deprecated_int96_timestamps()) {
++ if (arrow_properties.support_deprecated_int96_timestamps() && target_unit
== ::arrow::TimeUnit::NANO) {
+ *physical_type = ParquetType::INT96;
+ return Status::OK();
+ }
+diff --git a/cpp/src/parquet/arrow/writer.cc b/cpp/src/parquet/arrow/writer.cc
+index 4fd7ef1b47..87326a54f1 100644
+--- a/cpp/src/parquet/arrow/writer.cc
++++ b/cpp/src/parquet/arrow/writer.cc
+@@ -314,6 +314,14 @@ class FileWriterImpl : public FileWriter {
+ return Status::OK();
+ }
+
++ int64_t GetBufferedSize() override {
++ if (row_group_writer_ == nullptr) {
++ return 0;
++ }
++ return row_group_writer_->total_compressed_bytes() +
++ row_group_writer_->total_compressed_bytes_written();
++ }
++
+ Status Close() override {
+ if (!closed_) {
+ // Make idempotent
+@@ -418,10 +426,13 @@ class FileWriterImpl : public FileWriter {
+
+ // Max number of rows allowed in a row group.
+ const int64_t max_row_group_length =
this->properties().max_row_group_length();
++ const int64_t max_row_group_size =
this->properties().max_row_group_size();
+
+ // Initialize a new buffered row group writer if necessary.
+ if (row_group_writer_ == nullptr || !row_group_writer_->buffered() ||
+- row_group_writer_->num_rows() >= max_row_group_length) {
++ row_group_writer_->num_rows() >= max_row_group_length ||
++ (row_group_writer_->total_compressed_bytes_written() +
++ row_group_writer_->total_compressed_bytes() >= max_row_group_size)) {
+ RETURN_NOT_OK(NewBufferedRowGroup());
+ }
+
+diff --git a/cpp/src/parquet/arrow/writer.h b/cpp/src/parquet/arrow/writer.h
+index 4a1a033a7b..0f13d05e44 100644
+--- a/cpp/src/parquet/arrow/writer.h
++++ b/cpp/src/parquet/arrow/writer.h
+@@ -138,6 +138,9 @@ class PARQUET_EXPORT FileWriter {
+ /// option in this case.
+ virtual ::arrow::Status WriteRecordBatch(const ::arrow::RecordBatch& batch)
= 0;
+
++ /// \brief Return the buffered size in bytes.
++ virtual int64_t GetBufferedSize() = 0;
++
+ /// \brief Write the footer and close the file.
+ virtual ::arrow::Status Close() = 0;
+ virtual ~FileWriter();
+diff --git a/cpp/src/parquet/column_reader.cc
b/cpp/src/parquet/column_reader.cc
+index ebf9515f27..0abc7d2320 100644
+--- a/cpp/src/parquet/column_reader.cc
++++ b/cpp/src/parquet/column_reader.cc
+@@ -208,6 +208,39 @@ ReaderProperties default_reader_properties() {
+ return default_reader_properties;
+ }
+
++void PageReader::set_data_page_read_plan(
++ int64_t first_data_page_offset,
++ std::vector<DataPageReadPlanEntry> data_pages) {
++ if (data_page_filter_) {
++ throw ParquetException(
++ "data_page_filter and data_page_read_plan cannot be enabled
together");
++ }
++ if (first_data_page_offset < 0) {
++ throw ParquetException("Invalid negative first data page offset");
++ }
++
++ int64_t previous_end = first_data_page_offset;
++ int32_t previous_ordinal = -1;
++ for (const auto& page : data_pages) {
++ int64_t page_end;
++ if (page.page_ordinal < 0 || page.offset < first_data_page_offset ||
++ page.compressed_page_size <= 0 ||
++ AddWithOverflow(page.offset, page.compressed_page_size, &page_end)) {
++ throw ParquetException("Invalid data page read plan entry");
++ }
++ if (page.offset < previous_end || page.page_ordinal <= previous_ordinal) {
++ throw ParquetException("Data page read plan entries must be ordered");
++ }
++ previous_end = page_end;
++ previous_ordinal = page.page_ordinal;
++ }
++
++ data_page_read_plan_enabled_ = true;
++ first_data_page_offset_ = first_data_page_offset;
++ data_page_read_plan_ = std::move(data_pages);
++ next_data_page_ = 0;
++}
++
+ namespace {
+
+ // Extracts encoded statistics from V1 and V2 data page headers
+@@ -430,9 +463,43 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+
+ // Loop here because there may be unhandled page types that we skip until
+ // finding a page that we do know what to do with
+- while (seen_num_values_ < total_num_values_) {
++ while (data_page_read_plan_enabled_ || seen_num_values_ <
total_num_values_) {
++ const DataPageReadPlanEntry* planned_data_page = nullptr;
++ uint32_t page_header_limit = max_page_header_size_;
++
++ if (data_page_read_plan_enabled_) {
++ if (next_data_page_ >= data_page_read_plan_.size()) {
++ return nullptr;
++ }
++
++ PARQUET_ASSIGN_OR_THROW(int64_t current_position, stream_->Tell());
++ if (current_position < first_data_page_offset_) {
++ page_header_limit = static_cast<uint32_t>(std::min<int64_t>(
++ page_header_limit, first_data_page_offset_ - current_position));
++ } else {
++ planned_data_page = &data_page_read_plan_[next_data_page_];
++ if (current_position > planned_data_page->offset) {
++ throw ParquetException("Data page read plan points behind stream
position");
++ }
++ PARQUET_THROW_NOT_OK(
++ stream_->Advance(planned_data_page->offset - current_position));
++ PARQUET_ASSIGN_OR_THROW(int64_t target_position, stream_->Tell());
++ if (target_position != planned_data_page->offset) {
++ throw ParquetException("Failed to seek to planned data page");
++ }
++ page_ordinal_ = planned_data_page->page_ordinal;
++ page_header_limit = static_cast<uint32_t>(std::min<int64_t>(
++ page_header_limit, planned_data_page->compressed_page_size));
++ }
++ }
++
++ if (page_header_limit == 0) {
++ throw ParquetException("No bytes available for page header");
++ }
++
+ uint32_t header_size = 0;
+- uint32_t allowed_page_size = kDefaultPageHeaderSize;
++ uint32_t allowed_page_size =
++ std::min<uint32_t>(kDefaultPageHeaderSize, page_header_limit);
- virtual ~FileColumnIterator() {}
+ // Page headers can be very large because of page statistics
+ // We try to deserialize a larger buffer progressively
+@@ -458,11 +525,12 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+ // Failed to deserialize. Double the allowed page header size and try
again
+ std::stringstream ss;
+ ss << e.what();
+- allowed_page_size *= 2;
+- if (allowed_page_size > max_page_header_size_) {
++ if (allowed_page_size >= page_header_limit) {
+ ss << "Deserializing page header failed.\n";
+ throw ParquetException(ss.str());
+ }
++ allowed_page_size =
++ std::min<uint32_t>(allowed_page_size * 2, page_header_limit);
+ }
+ }
+ // Advance the stream offset
+@@ -474,6 +542,20 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+ throw ParquetException("Invalid page header");
+ }
-- std::unique_ptr<::parquet::PageReader> NextChunk() {
-+ /// \brief Fetch the PageReader for the next row group in this iterator's
-+ /// range. Virtual so subclasses can decorate the returned PageReader, e.g.
-+ /// to install a data_page_filter for I/O-level page skipping.
-+ virtual std::unique_ptr<::parquet::PageReader> NextChunk() {
- if (row_groups_.empty()) {
- return nullptr;
++ const PageType::type page_type = LoadEnumSafe(¤t_page_header_.type);
++ if (planned_data_page != nullptr) {
++ if (page_type != PageType::DATA_PAGE && page_type !=
PageType::DATA_PAGE_V2) {
++ throw ParquetException("Data page read plan points to a non-data
page");
++ }
++ int64_t total_compressed_size;
++ if (AddWithOverflow(static_cast<int64_t>(header_size),
++ static_cast<int64_t>(compressed_len),
++ &total_compressed_size) ||
++ total_compressed_size != planned_data_page->compressed_page_size) {
++ throw ParquetException("Planned data page size does not match page
header");
++ }
++ }
++
+ EncodedStatistics data_page_statistics;
+ if (ShouldSkipPage(&data_page_statistics)) {
+ PARQUET_THROW_NOT_OK(stream_->Advance(compressed_len));
+@@ -494,8 +576,6 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+ ParquetException::EofException(ss.str());
}
-@@ -95,9 +99,6 @@
- std::deque<int> row_groups_;
- };
--using FileColumnIteratorFactory =
-- std::function<FileColumnIterator*(int, ParquetFileReader*)>;
+- const PageType::type page_type = LoadEnumSafe(¤t_page_header_.type);
-
- Status TransferColumnData(::parquet::internal::RecordReader* reader,
- const std::shared_ptr<::arrow::Field>& value_field,
- const ColumnDescriptor* descr, ::arrow::MemoryPool*
pool,
+ if (properties_.page_checksum_verification() &&
current_page_header_.__isset.crc &&
+ PageCanUseChecksum(page_type)) {
+ // verify crc
+@@ -534,6 +614,9 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+
LoadEnumSafe(&dict_header.encoding),
+ is_sorted);
+ } else if (page_type == PageType::DATA_PAGE) {
++ if (planned_data_page != nullptr) {
++ ++next_data_page_;
++ }
+ ++page_ordinal_;
+ const format::DataPageHeader& header =
current_page_header_.data_page_header;
+ page_buffer =
+@@ -545,6 +628,9 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+ LoadEnumSafe(&header.repetition_level_encoding), uncompressed_len,
+ std::move(data_page_statistics));
+ } else if (page_type == PageType::DATA_PAGE_V2) {
++ if (planned_data_page != nullptr) {
++ ++next_data_page_;
++ }
+ ++page_ordinal_;
+ const format::DataPageHeaderV2& header =
current_page_header_.data_page_header_v2;
+
diff --git a/cpp/src/parquet/column_reader.h b/cpp/src/parquet/column_reader.h
+index 29e1b2a25e..386e574644 100644
--- a/cpp/src/parquet/column_reader.h
+++ b/cpp/src/parquet/column_reader.h
@@ -76,6 +76,18 @@ struct PARQUET_EXPORT DataPageStats {
@@ -797,156 +659,302 @@ diff --git a/cpp/src/parquet/column_reader.h
b/cpp/src/parquet/column_reader.h
};
class PARQUET_EXPORT ColumnReader {
-diff --git a/cpp/src/parquet/column_reader.cc
b/cpp/src/parquet/column_reader.cc
---- a/cpp/src/parquet/column_reader.cc
-+++ b/cpp/src/parquet/column_reader.cc
-@@ -207,6 +207,39 @@ ReaderProperties default_reader_properties() {
- return default_reader_properties;
+diff --git a/cpp/src/parquet/file_reader.cc b/cpp/src/parquet/file_reader.cc
+index 3e9eeea6c6..671ebe4644 100644
+--- a/cpp/src/parquet/file_reader.cc
++++ b/cpp/src/parquet/file_reader.cc
+@@ -207,6 +207,117 @@ const RowGroupMetaData* RowGroupReader::metadata() const
{ return contents_->met
+ return {col_start, col_length};
}
-+void PageReader::set_data_page_read_plan(
-+ int64_t first_data_page_offset,
-+ std::vector<DataPageReadPlanEntry> data_pages) {
-+ if (data_page_filter_) {
-+ throw ParquetException(
-+ "data_page_filter and data_page_read_plan cannot be enabled
together");
++// CachedInputStream: InputStream adapter that reads through ReadRangeCache
with
++// zero-cost skip for non-cached pages. Used for page-level caching where only
++// specific pages are pre-buffered.
++//
++// Key behavior:
++// - Read(): On cache hit, returns cached data. On cache miss, returns
zero-filled
++// buffer (zero I/O). This makes InputStream::Advance() (which calls Read()
and
++// discards) effectively free for skipped pages.
++// - Peek(): Always falls back to source on cache miss, because PageReader
uses
++// Peek() to read Thrift page headers (~30 bytes) which must have real data.
++class CachedInputStream : public ::arrow::io::InputStream {
++ public:
++ CachedInputStream(
++ std::shared_ptr<::arrow::io::internal::ReadRangeCache> cache,
++ std::shared_ptr<ArrowInputFile> source,
++ int64_t offset, int64_t length)
++ : cache_(std::move(cache)),
++ source_(std::move(source)),
++ base_offset_(offset),
++ length_(length) {}
++
++ ::arrow::Status Close() override {
++ closed_ = true;
++ return ::arrow::Status::OK();
+ }
-+ if (first_data_page_offset < 0) {
-+ throw ParquetException("Invalid negative first data page offset");
++
++ bool closed() const override { return closed_; }
++
++ ::arrow::Result<int64_t> Tell() const override { return position_; }
++
++ ::arrow::Result<std::string_view> Peek(int64_t nbytes) override {
++ int64_t to_read = std::min(nbytes, length_ - position_);
++ if (to_read <= 0) {
++ return std::string_view();
++ }
++ ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
++ auto result = cache_->Read(range);
++ if (result.ok()) {
++ peek_buffer_ = *result;
++ } else {
++ // Peek is used for Thrift page headers (~30 bytes) — must read real
data
++ ARROW_ASSIGN_OR_RAISE(peek_buffer_,
++ source_->ReadAt(range.offset, range.length));
++ }
++ return std::string_view(
++ reinterpret_cast<const char*>(peek_buffer_->data()),
++ static_cast<size_t>(peek_buffer_->size()));
++ }
++
++ ::arrow::Result<int64_t> Read(int64_t nbytes, void* out) override {
++ int64_t to_read = std::min(nbytes, length_ - position_);
++ if (to_read <= 0) return 0;
++ ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
++ auto result = cache_->Read(range);
++ if (result.ok()) {
++ auto& buf = *result;
++ memcpy(out, buf->data(), static_cast<size_t>(buf->size()));
++ position_ += buf->size();
++ return buf->size();
++ }
++ // Cache miss: fall back to real I/O from source
++ ARROW_ASSIGN_OR_RAISE(auto buf, source_->ReadAt(range.offset,
range.length));
++ memcpy(out, buf->data(), static_cast<size_t>(buf->size()));
++ position_ += buf->size();
++ return buf->size();
++ }
++
++ ::arrow::Result<std::shared_ptr<::arrow::Buffer>> Read(int64_t nbytes)
override {
++ int64_t to_read = std::min(nbytes, length_ - position_);
++ if (to_read <= 0) {
++ return std::make_shared<::arrow::Buffer>(nullptr, 0);
++ }
++ ::arrow::io::ReadRange range{base_offset_ + position_, to_read};
++ auto result = cache_->Read(range);
++ if (result.ok()) {
++ position_ += (*result)->size();
++ return *result;
++ }
++ // Cache miss: fall back to real I/O from source
++ ARROW_ASSIGN_OR_RAISE(auto buf, source_->ReadAt(range.offset,
range.length));
++ position_ += buf->size();
++ return std::shared_ptr<::arrow::Buffer>(std::move(buf));
++ }
++
++ // Override Advance to avoid real I/O for skipped pages.
++ // The default InputStream::Advance() calls Read() and discards the result,
++ // which would trigger source_->ReadAt() on cache miss — defeating
page-level
++ // I/O skipping via data_page_filter. Since Advance() is only used to skip
++ // over data that will not be consumed, we can safely just move the
position.
++ ::arrow::Status Advance(int64_t nbytes) override {
++ if (nbytes <= 0) {
++ return ::arrow::Status::OK();
++ }
++ int64_t remaining = length_ - position_;
++ if (remaining <= 0) {
++ return ::arrow::Status::OK();
++ }
++ position_ += std::min(nbytes, remaining);
++ return ::arrow::Status::OK();
++ }
++
++ private:
++ std::shared_ptr<::arrow::io::internal::ReadRangeCache> cache_;
++ std::shared_ptr<ArrowInputFile> source_;
++ int64_t base_offset_;
++ int64_t length_;
++ int64_t position_ = 0;
++ bool closed_ = false;
++ std::shared_ptr<::arrow::Buffer> peek_buffer_;
++};
++
+ // RowGroupReader::Contents implementation for the Parquet file specification
+ class SerializedRowGroup : public RowGroupReader::Contents {
+ public:
+@@ -242,6 +353,11 @@ class SerializedRowGroup : public
RowGroupReader::Contents {
+ // segments.
+ PARQUET_ASSIGN_OR_THROW(auto buffer, cached_source_->Read(col_range));
+ stream = std::make_shared<::arrow::io::BufferReader>(buffer);
++ } else if (cached_source_) {
++ // Page-level caching: read through cache with fallback to source.
++ // Advance() is zero-cost for skipped pages via data_page_filter.
++ stream = std::make_shared<CachedInputStream>(
++ cached_source_, source_, col_range.offset, col_range.length);
+ } else {
+ stream = properties_.GetStream(source_, col_range.offset,
col_range.length);
+ }
+@@ -417,6 +533,26 @@ class SerializedFile : public ParquetFileReader::Contents
{
+ return cached_source_->WaitFor(ranges);
+ }
+
++ void PreBufferRanges(const std::vector<::arrow::io::ReadRange>& ranges,
++ const ::arrow::io::IOContext& ctx,
++ const ::arrow::io::CacheOptions& options) {
++ cached_source_ =
++ std::make_shared<::arrow::io::internal::ReadRangeCache>(source_, ctx,
options);
++ // Do NOT set prebuffered_column_chunks_ bitmap — GetColumnPageReader will
++ // use CachedInputStream path instead of full-chunk BufferReader path.
++ prebuffered_column_chunks_.clear();
++ PARQUET_THROW_NOT_OK(cached_source_->Cache(ranges));
+ }
+
-+ int64_t previous_end = first_data_page_offset;
-+ int32_t previous_ordinal = -1;
-+ for (const auto& page : data_pages) {
-+ int64_t page_end;
-+ if (page.page_ordinal < 0 || page.offset < first_data_page_offset ||
-+ page.compressed_page_size <= 0 ||
-+ AddWithOverflow(page.offset, page.compressed_page_size, &page_end)) {
-+ throw ParquetException("Invalid data page read plan entry");
-+ }
-+ if (page.offset < previous_end || page.page_ordinal <= previous_ordinal) {
-+ throw ParquetException("Data page read plan entries must be ordered");
++ ::arrow::Future<> WhenBufferedRanges(
++ const std::vector<::arrow::io::ReadRange>& ranges) const {
++ if (!cached_source_) {
++ return ::arrow::Status::Invalid(
++ "Must call PreBufferRanges before WhenBufferedRanges");
+ }
-+ previous_end = page_end;
-+ previous_ordinal = page.page_ordinal;
++ return cached_source_->WaitFor(ranges);
+ }
+
-+ data_page_read_plan_enabled_ = true;
-+ first_data_page_offset_ = first_data_page_offset;
-+ data_page_read_plan_ = std::move(data_pages);
-+ next_data_page_ = 0;
+ // Metadata/footer parsing. Divided up to separate sync/async paths, and to
use
+ // exceptions for error handling (with the async path converting to
Future/Status).
+
+@@ -911,6 +1047,22 @@ void ParquetFileReader::PreBuffer(const
std::vector<int>& row_groups,
+ return file->WhenBuffered(row_groups, column_indices);
+ }
+
++void ParquetFileReader::PreBufferRanges(
++ const std::vector<::arrow::io::ReadRange>& ranges,
++ const ::arrow::io::IOContext& ctx,
++ const ::arrow::io::CacheOptions& options) {
++ SerializedFile* file =
++ ::arrow::internal::checked_cast<SerializedFile*>(contents_.get());
++ file->PreBufferRanges(ranges, ctx, options);
+}
+
- namespace {
++::arrow::Future<> ParquetFileReader::WhenBufferedRanges(
++ const std::vector<::arrow::io::ReadRange>& ranges) const {
++ SerializedFile* file =
++ ::arrow::internal::checked_cast<SerializedFile*>(contents_.get());
++ return file->WhenBufferedRanges(ranges);
++}
++
+ // ----------------------------------------------------------------------
+ // File metadata helpers
- // Extracts encoded statistics from V1 and V2 data page headers
-@@ -430,9 +463,43 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
+diff --git a/cpp/src/parquet/file_reader.h b/cpp/src/parquet/file_reader.h
+index b59b59f95c..657a438a3a 100644
+--- a/cpp/src/parquet/file_reader.h
++++ b/cpp/src/parquet/file_reader.h
+@@ -210,6 +210,17 @@ class PARQUET_EXPORT ParquetFileReader {
+ ::arrow::Future<> WhenBuffered(const std::vector<int>& row_groups,
+ const std::vector<int>& column_indices)
const;
- // Loop here because there may be unhandled page types that we skip until
- // finding a page that we do know what to do with
-- while (seen_num_values_ < total_num_values_) {
-+ while (data_page_read_plan_enabled_ || seen_num_values_ <
total_num_values_) {
-+ const DataPageReadPlanEntry* planned_data_page = nullptr;
-+ uint32_t page_header_limit = max_page_header_size_;
-+
-+ if (data_page_read_plan_enabled_) {
-+ if (next_data_page_ >= data_page_read_plan_.size()) {
-+ return nullptr;
-+ }
++ /// Pre-buffer arbitrary byte ranges (e.g., page-level ranges from
OffsetIndex).
++ /// Unlike PreBuffer(), this does NOT set the column bitmap, so
++ /// GetColumnPageReader will use CachedInputStream (page-level cache path).
++ void PreBufferRanges(const std::vector<::arrow::io::ReadRange>& ranges,
++ const ::arrow::io::IOContext& ctx,
++ const ::arrow::io::CacheOptions& options);
+
-+ PARQUET_ASSIGN_OR_THROW(int64_t current_position, stream_->Tell());
-+ if (current_position < first_data_page_offset_) {
-+ page_header_limit = static_cast<uint32_t>(std::min<int64_t>(
-+ page_header_limit, first_data_page_offset_ - current_position));
-+ } else {
-+ planned_data_page = &data_page_read_plan_[next_data_page_];
-+ if (current_position > planned_data_page->offset) {
-+ throw ParquetException("Data page read plan points behind stream
position");
-+ }
-+ PARQUET_THROW_NOT_OK(
-+ stream_->Advance(planned_data_page->offset - current_position));
-+ PARQUET_ASSIGN_OR_THROW(int64_t target_position, stream_->Tell());
-+ if (target_position != planned_data_page->offset) {
-+ throw ParquetException("Failed to seek to planned data page");
-+ }
-+ page_ordinal_ = planned_data_page->page_ordinal;
-+ page_header_limit = static_cast<uint32_t>(std::min<int64_t>(
-+ page_header_limit, planned_data_page->compressed_page_size));
-+ }
-+ }
++ /// Wait for arbitrary byte ranges to be pre-buffered.
++ ::arrow::Future<> WhenBufferedRanges(
++ const std::vector<::arrow::io::ReadRange>& ranges) const;
+
-+ if (page_header_limit == 0) {
-+ throw ParquetException("No bytes available for page header");
+ private:
+ // Holds a pointer to an instance of Contents implementation
+ std::unique_ptr<Contents> contents_;
+diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h
+index 4d3acb491e..3906ff3c59 100644
+--- a/cpp/src/parquet/properties.h
++++ b/cpp/src/parquet/properties.h
+@@ -139,6 +139,7 @@ static constexpr bool DEFAULT_IS_DICTIONARY_ENABLED = true;
+ static constexpr int64_t DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT =
kDefaultDataPageSize;
+ static constexpr int64_t DEFAULT_WRITE_BATCH_SIZE = 1024;
+ static constexpr int64_t DEFAULT_MAX_ROW_GROUP_LENGTH = 1024 * 1024;
++static constexpr int64_t DEFAULT_MAX_ROW_GROUP_SIZE = 128 * 1024 * 1024;
+ static constexpr bool DEFAULT_ARE_STATISTICS_ENABLED = true;
+ static constexpr int64_t DEFAULT_MAX_STATISTICS_SIZE = 4096;
+ static constexpr Encoding::type DEFAULT_ENCODING = Encoding::UNKNOWN;
+@@ -232,6 +233,7 @@ class PARQUET_EXPORT WriterProperties {
+ dictionary_pagesize_limit_(DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT),
+ write_batch_size_(DEFAULT_WRITE_BATCH_SIZE),
+ max_row_group_length_(DEFAULT_MAX_ROW_GROUP_LENGTH),
++ max_row_group_size_(DEFAULT_MAX_ROW_GROUP_SIZE),
+ pagesize_(kDefaultDataPageSize),
+ version_(ParquetVersion::PARQUET_2_6),
+ data_page_version_(ParquetDataPageVersion::V1),
+@@ -244,6 +246,7 @@ class PARQUET_EXPORT WriterProperties {
+ dictionary_pagesize_limit_(properties.dictionary_pagesize_limit()),
+ write_batch_size_(properties.write_batch_size()),
+ max_row_group_length_(properties.max_row_group_length()),
++ max_row_group_size_(properties.max_row_group_size()),
+ pagesize_(properties.data_pagesize()),
+ version_(properties.version()),
+ data_page_version_(properties.data_page_version()),
+@@ -321,6 +324,13 @@ class PARQUET_EXPORT WriterProperties {
+ return this;
+ }
+
++ /// Specify the max bytes size to put in a single row group.
++ /// Default 128 M.
++ Builder* max_row_group_size(int64_t max_row_group_size) {
++ max_row_group_size_ = max_row_group_size;
++ return this;
+ }
+
- uint32_t header_size = 0;
-- uint32_t allowed_page_size = kDefaultPageHeaderSize;
-+ uint32_t allowed_page_size =
-+ std::min<uint32_t>(kDefaultPageHeaderSize, page_header_limit);
+ /// Specify the data page size.
+ /// Default 1MB.
+ Builder* data_pagesize(int64_t pg_size) {
+@@ -664,7 +674,7 @@ class PARQUET_EXPORT WriterProperties {
- // Page headers can be very large because of page statistics
- // We try to deserialize a larger buffer progressively
-@@ -458,11 +525,12 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
- // Failed to deserialize. Double the allowed page header size and try
again
- std::stringstream ss;
- ss << e.what();
-- allowed_page_size *= 2;
-- if (allowed_page_size > max_page_header_size_) {
-+ if (allowed_page_size >= page_header_limit) {
- ss << "Deserializing page header failed.\n";
- throw ParquetException(ss.str());
- }
-+ allowed_page_size =
-+ std::min<uint32_t>(allowed_page_size * 2, page_header_limit);
- }
- }
- // Advance the stream offset
-@@ -474,6 +542,20 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
- throw ParquetException("Invalid page header");
- }
+ return std::shared_ptr<WriterProperties>(new WriterProperties(
+ pool_, dictionary_pagesize_limit_, write_batch_size_,
max_row_group_length_,
+- pagesize_, version_, created_by_, page_checksum_enabled_,
++ max_row_group_size_, pagesize_, version_, created_by_,
page_checksum_enabled_,
+ std::move(file_encryption_properties_), default_column_properties_,
+ column_properties, data_page_version_, store_decimal_as_integer_,
+ std::move(sorting_columns_)));
+@@ -675,6 +685,7 @@ class PARQUET_EXPORT WriterProperties {
+ int64_t dictionary_pagesize_limit_;
+ int64_t write_batch_size_;
+ int64_t max_row_group_length_;
++ int64_t max_row_group_size_;
+ int64_t pagesize_;
+ ParquetVersion::type version_;
+ ParquetDataPageVersion data_page_version_;
+@@ -705,6 +716,8 @@ class PARQUET_EXPORT WriterProperties {
-+ const PageType::type page_type = LoadEnumSafe(¤t_page_header_.type);
-+ if (planned_data_page != nullptr) {
-+ if (page_type != PageType::DATA_PAGE && page_type !=
PageType::DATA_PAGE_V2) {
-+ throw ParquetException("Data page read plan points to a non-data
page");
-+ }
-+ int64_t total_compressed_size;
-+ if (AddWithOverflow(static_cast<int64_t>(header_size),
-+ static_cast<int64_t>(compressed_len),
-+ &total_compressed_size) ||
-+ total_compressed_size != planned_data_page->compressed_page_size) {
-+ throw ParquetException("Planned data page size does not match page
header");
-+ }
-+ }
+ inline int64_t max_row_group_length() const { return max_row_group_length_;
}
+
++ inline int64_t max_row_group_size() const { return max_row_group_size_; }
+
- EncodedStatistics data_page_statistics;
- if (ShouldSkipPage(&data_page_statistics)) {
- PARQUET_THROW_NOT_OK(stream_->Advance(compressed_len));
-@@ -494,8 +576,6 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
- ParquetException::EofException(ss.str());
- }
+ inline int64_t data_pagesize() const { return pagesize_; }
-- const PageType::type page_type = LoadEnumSafe(¤t_page_header_.type);
--
- if (properties_.page_checksum_verification() &&
current_page_header_.__isset.crc &&
- PageCanUseChecksum(page_type)) {
- // verify crc
-@@ -534,6 +614,9 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
-
LoadEnumSafe(&dict_header.encoding),
- is_sorted);
- } else if (page_type == PageType::DATA_PAGE) {
-+ if (planned_data_page != nullptr) {
-+ ++next_data_page_;
-+ }
- ++page_ordinal_;
- const format::DataPageHeader& header =
current_page_header_.data_page_header;
- page_buffer =
-@@ -545,6 +628,9 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
- LoadEnumSafe(&header.repetition_level_encoding), uncompressed_len,
- std::move(data_page_statistics));
- } else if (page_type == PageType::DATA_PAGE_V2) {
-+ if (planned_data_page != nullptr) {
-+ ++next_data_page_;
-+ }
- ++page_ordinal_;
- const format::DataPageHeaderV2& header =
current_page_header_.data_page_header_v2;
+ inline ParquetDataPageVersion data_page_version() const {
+@@ -810,7 +823,7 @@ class PARQUET_EXPORT WriterProperties {
+ private:
+ explicit WriterProperties(
+ MemoryPool* pool, int64_t dictionary_pagesize_limit, int64_t
write_batch_size,
+- int64_t max_row_group_length, int64_t pagesize, ParquetVersion::type
version,
++ int64_t max_row_group_length, int64_t max_row_group_size, int64_t
pagesize, ParquetVersion::type version,
+ const std::string& created_by, bool page_write_checksum_enabled,
+ std::shared_ptr<FileEncryptionProperties> file_encryption_properties,
+ const ColumnProperties& default_column_properties,
+@@ -821,6 +834,7 @@ class PARQUET_EXPORT WriterProperties {
+ dictionary_pagesize_limit_(dictionary_pagesize_limit),
+ write_batch_size_(write_batch_size),
+ max_row_group_length_(max_row_group_length),
++ max_row_group_size_(max_row_group_size),
+ pagesize_(pagesize),
+ parquet_data_page_version_(data_page_version),
+ parquet_version_(version),
+@@ -836,6 +850,7 @@ class PARQUET_EXPORT WriterProperties {
+ int64_t dictionary_pagesize_limit_;
+ int64_t write_batch_size_;
+ int64_t max_row_group_length_;
++ int64_t max_row_group_size_;
+ int64_t pagesize_;
+ ParquetDataPageVersion parquet_data_page_version_;
+ ParquetVersion::type parquet_version_;
