arrow.diff

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/reader.cc b/cpp/src/parquet/arrow/reader.cc
index 285e2a5973..aa6f92f077 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,
     return Status::OK();
   }

-  int64_t num_rows = 0;
+  std::vector<int64_t> num_rows;
   for (int row_group : row_groups) {
-    num_rows += parquet_reader()->metadata()->RowGroup(row_group)->num_rows();
+    num_rows.push_back(parquet_reader()->metadata()->RowGroup(row_group)->num_rows());
   }

   using ::arrow::RecordBatchIterator;
+  int row_group_idx = 0;

   // NB: This lambda will be invoked outside the scope of this call to
   // `GetRecordBatchReader()`, so it must capture `readers` and `batch_schema` by value.
   // `this` is a non-owning pointer so we are relying on the parent FileReader outliving
   // this RecordBatchReader.
   ::arrow::Iterator<RecordBatchIterator> batches = ::arrow::MakeFunctionIterator(
-      [readers, batch_schema, num_rows,
+      [readers, batch_schema, num_rows, row_group_idx,
        this]() mutable -> ::arrow::Result<RecordBatchIterator> {
         ::arrow::ChunkedArrayVector columns(readers.size());

-        // don't reserve more rows than necessary
-        int64_t batch_size = std::min(properties().batch_size(), num_rows);
-        num_rows -= batch_size;
+        int64_t batch_size = 0;
+        if (!num_rows.empty()) {
+          // don't reserve more rows than necessary
+          batch_size = std::min(properties().batch_size(), num_rows[row_group_idx]);
+          num_rows[row_group_idx] -= batch_size;
+          if (num_rows[row_group_idx] == 0 && (num_rows.size() - 1) != row_group_idx) {
+            row_group_idx++;
+          }
+        }

         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();
   }

+  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/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
@@ -211,7 +211,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);