From 4ca7e53553b9264fc80e93e93cb2e7ed1b98fff1 Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Sat, 10 Oct 2026 23:37:12 +0000
Subject: [PATCH 01/11] feat(scan): read Parquet files by task range

FileScanTaskReader previously rejected tasks that read part of a file.
It now passes Parquet task ranges to the reader while keeping the full
file size available for reading the footer.

Use the first column's starting page to assign each row group to a range.
This works when optional row-group offsets are missing and ensures that
adjacent ranges read every row exactly once. Keep row positions and
inherited row IDs relative to the whole file so deletes still remove the
correct rows, even when page offsets are out of order.

Recheck task bounds when opening a file, skip empty row groups, and reject
invalid page offsets and partial tasks for unsupported formats. Add tests
for split boundaries, deletes, row lineage, changed file sizes, and files
with missing row-group offsets.

Validation: all 306 tests in parquet_test and data_test pass. Formatting
and whitespace checks pass. pre-commit could not run because it is not
installed and PyPI DNS resolution failed during installation.
---
 src/iceberg/data/file_scan_task_reader.cc     |  36 ++-
 src/iceberg/data/file_scan_task_reader.h      |   7 +-
 src/iceberg/parquet/parquet_data_util.cc      |  23 ++
 .../parquet/parquet_data_util_internal.h      |   9 +
 src/iceberg/parquet/parquet_reader.cc         | 120 +++++++---
 .../test/file_scan_task_reader_test.cc        | 174 ++++++++++++++-
 src/iceberg/test/parquet_test.cc              | 210 +++++++++++++++++-
 .../split-without-row-group-offsets.md        |  45 ++++
 .../split-without-row-group-offsets.parquet   | Bin 0 -> 797 bytes
 9 files changed, 575 insertions(+), 49 deletions(-)
 create mode 100644 src/iceberg/test/resources/split-without-row-group-offsets.md
 create mode 100644 src/iceberg/test/resources/split-without-row-group-offsets.parquet

diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc
index 7d3a8bc7e3..8c55e31b2b 100644
--- a/src/iceberg/data/file_scan_task_reader.cc
+++ b/src/iceberg/data/file_scan_task_reader.cc
@@ -39,16 +39,21 @@ namespace iceberg {
 
 namespace {
 
-ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptr<FileIO> io,
+ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr<FileIO> io,
                                 std::shared_ptr<Schema> projection,
                                 std::shared_ptr<Expression> filter,
                                 std::shared_ptr<NameMapping> name_mapping,
                                 ReaderProperties properties,
                                 std::optional<int64_t> first_row_id,
                                 std::optional<int64_t> data_sequence_number) {
+  const auto& data_file = *task.data_file();
   return ReaderOptions{
       .path = data_file.file_path,
       .length = static_cast<size_t>(data_file.file_size_in_bytes),
+      .split = task.is_split() ? std::make_optional(
+                                     Split{.offset = static_cast<size_t>(task.start()),
+                                           .length = static_cast<size_t>(task.length())})
+                               : std::nullopt,
       .io = std::move(io),
       .projection = std::move(projection),
       .filter = std::move(filter),
@@ -163,16 +168,29 @@ class FileScanTaskReader::Impl {
     ICEBERG_PRECHECK(data_file->file_size_in_bytes >= 0,
                      "Data file size must not be negative: {}",
                      data_file->file_size_in_bytes);
+    ICEBERG_PRECHECK(std::in_range<size_t>(data_file->file_size_in_bytes),
+                     "Data file size cannot be represented as size_t: {}",
+                     data_file->file_size_in_bytes);
     if (task.is_split()) {
-      return NotSupported(
-          "Reading a partial FileScanTask is not supported yet: {} "
-          "(start={}, length={})",
-          data_file->file_path, task.start(), task.length());
+      ICEBERG_PRECHECK(task.start() >= 0 &&
+                           task.start() <= data_file->file_size_in_bytes &&
+                           task.length() > 0 &&
+                           task.length() <= data_file->file_size_in_bytes - task.start(),
+                       "Task range is outside file {} of size {} (start={}, length={})",
+                       data_file->file_path, data_file->file_size_in_bytes, task.start(),
+                       task.length());
+      // The validated endpoints are bounded by the size_t-representable file size.
+      if (data_file->file_format != FileFormatType::kParquet) {
+        return NotSupported(
+            "Reading a partial FileScanTask requires Parquet: {} "
+            "(start={}, length={})",
+            data_file->file_path, task.start(), task.length());
+      }
     }
 
     if (task.delete_files().empty()) {
       auto options = MakeReaderOptions(
-          *data_file, io_, projected_schema_, task.residual_filter(), name_mapping_,
+          task, io_, projected_schema_, task.residual_filter(), name_mapping_,
           properties_, data_file->first_row_id, data_file->data_sequence_number);
       ICEBERG_ASSIGN_OR_RAISE(
           auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options));
@@ -197,9 +215,9 @@ class FileScanTaskReader::Impl {
                             ProjectionContext::Make(*required_schema, *projected_schema_,
                                                     project_batch_function));
 
-    auto options = MakeReaderOptions(
-        *data_file, io_, required_schema, task.residual_filter(), name_mapping_,
-        properties_, data_file->first_row_id, data_file->data_sequence_number);
+    auto options = MakeReaderOptions(task, io_, required_schema, task.residual_filter(),
+                                     name_mapping_, properties_, data_file->first_row_id,
+                                     data_file->data_sequence_number);
     ICEBERG_ASSIGN_OR_RAISE(auto reader,
                             ReaderFactoryRegistry::Open(data_file->file_format, options));
 
diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h
index c34f3e7aa8..faa62750c6 100644
--- a/src/iceberg/data/file_scan_task_reader.h
+++ b/src/iceberg/data/file_scan_task_reader.h
@@ -68,8 +68,11 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
 
   /// \brief Open a task and return an Arrow C stream for its projected live rows.
   ///
-  /// Partial tasks are rejected until range-aware reading is implemented. The task and
-  /// its shared data/delete-file metadata must not be mutated while the stream uses them.
+  /// Partial tasks are supported for Parquet, with each row group assigned to the
+  /// half-open range containing its first column's starting page. Row positions and
+  /// inherited row IDs remain relative to the whole file. Other formats require
+  /// whole-file tasks. The task and its shared data/delete-file metadata must not be
+  /// mutated while the stream uses them.
   Result<ArrowArrayStream> Open(const FileScanTask& task);
 
   FileScanTaskReader(const FileScanTaskReader&) = delete;
diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc
index 449657df84..b18ff6b8a9 100644
--- a/src/iceberg/parquet/parquet_data_util.cc
+++ b/src/iceberg/parquet/parquet_data_util.cc
@@ -24,6 +24,7 @@
 #include <arrow/compute/api.h>
 #include <arrow/record_batch.h>
 #include <arrow/type.h>
+#include <parquet/metadata.h>
 
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow/literal_util_internal.h"
@@ -37,6 +38,28 @@
 
 namespace iceberg::parquet {
 
+Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
+                              int64_t file_size) {
+  if (row_group.num_columns() == 0) {
+    return Invalid("Cannot split a Parquet row group without columns");
+  }
+  auto column = row_group.ColumnChunk(0);
+  int64_t start = column->data_page_offset();
+  if (column->has_dictionary_page()) {
+    const int64_t dictionary_start = column->dictionary_page_offset();
+    if (dictionary_start > 0 && dictionary_start < start) {
+      start = dictionary_start;
+    }
+  }
+  if (start <= 0 || start >= file_size) {
+    return Invalid(
+        "Cannot split a Parquet row group with missing or out-of-file "
+        "starting page {} (file size {})",
+        start, file_size);
+  }
+  return start;
+}
+
 namespace {
 
 // Forward declaration
diff --git a/src/iceberg/parquet/parquet_data_util_internal.h b/src/iceberg/parquet/parquet_data_util_internal.h
index ad70c223e0..95ab0230ed 100644
--- a/src/iceberg/parquet/parquet_data_util_internal.h
+++ b/src/iceberg/parquet/parquet_data_util_internal.h
@@ -24,12 +24,21 @@
 #include "iceberg/arrow/metadata_column_util_internal.h"
 #include "iceberg/schema_util.h"
 
+namespace parquet {
+class RowGroupMetaData;
+}  // namespace parquet
+
 namespace iceberg::parquet {
 
 /// \brief Whether direct fields need positions for _pos or inherited _row_id.
 bool NeedsRowPositions(const StructType& struct_type,
                        const arrow::MetadataColumnContext& metadata_context);
 
+/// \brief Return the first column's starting page, matching parquet-mr's
+/// BlockMetaData.getStartingPos(). Reject missing or out-of-file starts.
+Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
+                              int64_t file_size);
+
 /// \brief Convert record batch read from a Parquet file to projected Iceberg Schema.
 ///
 /// \param record_batch The record batch to convert.
diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index 90fa064658..f1130afeff 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -20,7 +20,9 @@
 #include "iceberg/parquet/parquet_reader.h"
 
 #include <algorithm>
+#include <limits>
 #include <numeric>
+#include <utility>
 #include <variant>
 #include <vector>
 
@@ -263,6 +265,13 @@ struct ReadContext {
   std::unique_ptr<RowNumberReader> row_number_reader_;
 };
 
+// Consecutive row groups can share a batch reader without introducing gaps in _pos.
+struct RowGroupRange {
+  int begin;
+  int end;
+  int64_t first_row_position;
+};
+
 // TODO(gangwu): list of work items
 // 1. Make the memory pool configurable
 // 2. Catch ParquetException and convert to Status/Result
@@ -276,7 +285,6 @@ class ParquetReader::Impl {
       return InvalidArgument("Projected schema is required by Parquet reader");
     }
 
-    split_ = options.split;
     read_schema_ = options.projection;
 
     // Prepare reader properties
@@ -297,6 +305,7 @@ class ParquetReader::Impl {
     ICEBERG_ARROW_ASSIGN_OR_RETURN(
         reader_, ::parquet::arrow::FileReader::Make(pool_, std::move(file_reader),
                                                     arrow_reader_properties));
+    ICEBERG_RETURN_UNEXPECTED(PlanRowGroupRanges(options));
 
     // Project read schema onto the Parquet file schema
     ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_));
@@ -313,9 +322,17 @@ class ParquetReader::Impl {
       ICEBERG_RETURN_UNEXPECTED(InitReadContext());
     }
 
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, context_->record_batch_reader_->Next());
-    if (!batch) {
-      return std::nullopt;
+    std::shared_ptr<::arrow::RecordBatch> batch;
+    while (true) {
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(batch, context_->record_batch_reader_->Next());
+      if (batch) {
+        break;
+      }
+      if (next_row_group_range_ == row_group_ranges_.size()) {
+        return std::nullopt;
+      }
+      ICEBERG_ARROW_RETURN_NOT_OK(context_->record_batch_reader_->Close());
+      ICEBERG_ASSIGN_OR_RAISE(context_->record_batch_reader_, OpenNextRowGroupRange());
     }
 
     std::shared_ptr<::arrow::Int64Array> row_positions;
@@ -388,36 +405,78 @@ class ParquetReader::Impl {
   }
 
  private:
-  Status InitReadContext() {
-    context_ = std::make_unique<ReadContext>();
-
-    // Row group pruning based on the split
+  Status PlanRowGroupRanges(const ReaderOptions& options) {
     // TODO(gangwu): add row group filtering based on zone map, bloom filter, etc.
     auto metadata = reader_->parquet_reader()->metadata();
-    std::vector<int> row_group_indices;
-    if (split_.has_value()) {
+    if (options.split.has_value()) {
+      const auto& split = *options.split;
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file_size, input_stream_->GetSize());
+      int64_t row_position = 0;
       for (int i = 0; i < metadata->num_row_groups(); ++i) {
-        auto row_group_offset = metadata->RowGroup(i)->file_offset();
-        if (row_group_offset >= split_->offset &&
-            row_group_offset < split_->offset + split_->length) {
-          row_group_indices.push_back(i);
-        } else if (row_group_offset >= split_->offset + split_->length) {
-          break;
+        auto row_group = metadata->RowGroup(i);
+        // Empty row groups have no pages to assign to a split.
+        if (row_group->num_rows() == 0) {
+          continue;
+        }
+        auto start = RowGroupStart(*row_group, file_size);
+        if (!start.has_value()) {
+          return Invalid(
+              "Cannot read Parquet split {} (start={}, length={}), row "
+              "group {}: {}",
+              options.path, split.offset, split.length, i, start.error().message);
+        }
+        // Subtraction avoids overflow for a caller-supplied unbounded split length.
+        if (std::cmp_greater_equal(*start, split.offset) &&
+            static_cast<uint64_t>(*start) - split.offset < split.length) {
+          if (!row_group_ranges_.empty() && row_group_ranges_.back().end == i) {
+            row_group_ranges_.back().end = i + 1;
+          } else {
+            row_group_ranges_.push_back(
+                {.begin = i, .end = i + 1, .first_row_position = row_position});
+          }
+        }
+        if (row_group->num_rows() < 0 ||
+            row_group->num_rows() > std::numeric_limits<int64_t>::max() - row_position) {
+          return Invalid("Invalid Parquet row count in file {}, row group {}",
+                         options.path, i);
         }
+        row_position += row_group->num_rows();
       }
-    } else {
-      row_group_indices.resize(metadata->num_row_groups());
-      std::iota(row_group_indices.begin(), row_group_indices.end(), 0);  // NOLINT
+    } else if (metadata->num_row_groups() > 0) {
+      row_group_ranges_.push_back(
+          {.begin = 0, .end = metadata->num_row_groups(), .first_row_position = 0});
     }
+    return {};
+  }
 
-    // Create the record batch reader, including for empty splits to preserve the schema.
-    auto column_indices = SelectedColumnIndices(projection_);
+  Result<std::unique_ptr<::arrow::RecordBatchReader>> OpenNextRowGroupRange() {
+    if (next_row_group_range_ == row_group_ranges_.size()) {
+      return std::make_unique<EmptyRecordBatchReader>();
+    }
+    const auto& range = row_group_ranges_[next_row_group_range_];
+    std::vector<int> row_group_indices(range.end - range.begin);
+    std::iota(row_group_indices.begin(), row_group_indices.end(), range.begin);  // NOLINT
     ICEBERG_ARROW_ASSIGN_OR_RETURN(
-        context_->record_batch_reader_,
-        reader_->GetRecordBatchReader(row_group_indices, column_indices));
+        auto batch_reader, reader_->GetRecordBatchReader(
+                               row_group_indices, SelectedColumnIndices(projection_)));
+    metadata_context_.next_file_pos = range.first_row_position;
+    ++next_row_group_range_;
+    return batch_reader;
+  }
+
+  Status InitReadContext() {
+    auto context = std::make_unique<ReadContext>();
+    ICEBERG_ASSIGN_OR_RAISE(context->record_batch_reader_, OpenNextRowGroupRange());
 
     if (NeedsRowPositions(*read_schema_, metadata_context_)) {
-      ICEBERG_ASSIGN_OR_RAISE(context_->row_number_reader_,
+      auto metadata = reader_->parquet_reader()->metadata();
+      std::vector<int> row_group_indices;
+      for (const auto& range : row_group_ranges_) {
+        for (int i = range.begin; i < range.end; ++i) {
+          row_group_indices.push_back(i);
+        }
+      }
+      ICEBERG_ASSIGN_OR_RAISE(context->row_number_reader_,
                               RowNumberReader::Make(*metadata, row_group_indices, pool_));
     }
 
@@ -426,7 +485,7 @@ class ParquetReader::Impl {
     // the schema of the file.
     ArrowSchema arrow_schema;
     ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema));
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_,
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(context->output_arrow_schema_,
                                    ::arrow::ImportSchema(&arrow_schema));
 
     // Align the output schema with the arrays the reader actually produces. The reader's
@@ -437,18 +496,19 @@ class ParquetReader::Impl {
     //   3. Mixed list and large_list types in files with stored schemas
     // For each projected field, we use the reader's actual type. For missing fields
     // (columns not in the file), we apply the configured use_large_list preference.
-    context_->output_arrow_schema_ = AlignOutputSchemaToReaderSchema(
-        context_->output_arrow_schema_, context_->record_batch_reader_->schema(),
+    context->output_arrow_schema_ = AlignOutputSchemaToReaderSchema(
+        context->output_arrow_schema_, context->record_batch_reader_->schema(),
         projection_, use_large_list_);
-
+    context_ = std::move(context);
     return {};
   }
 
  private:
   // TODO(gangwu): make memory pool configurable
   ::arrow::MemoryPool* pool_ = ::arrow::default_memory_pool();
-  // The split to read from the Parquet file.
-  std::optional<Split> split_;
+  // Selected row groups in file row order, separated at gaps to preserve absolute _pos.
+  std::vector<RowGroupRange> row_group_ranges_;
+  size_t next_row_group_range_ = 0;
   // Whether to read list columns as large_list (64-bit offsets).
   bool use_large_list_ = false;
   // Schema to read from the Parquet file.
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc
index 11366f72d9..c59da128ed 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -19,9 +19,12 @@
 
 #include "iceberg/data/file_scan_task_reader.h"
 
+#include <algorithm>
+#include <array>
 #include <memory>
 #include <optional>
 #include <string>
+#include <tuple>
 #include <utility>
 #include <vector>
 
@@ -31,11 +34,13 @@
 #include <arrow/record_batch.h>
 #include <arrow/table.h>
 #include <parquet/arrow/writer.h>
+#include <parquet/file_reader.h>
 #include <parquet/metadata.h>
 #include <parquet/properties.h>
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_register.h"
+#include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow_c_data.h"
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/arrow_c_data_util_internal.h"
@@ -56,6 +61,7 @@
 #include "iceberg/schema_internal.h"
 #include "iceberg/table_scan.h"
 #include "iceberg/test/matchers.h"
+#include "iceberg/test/std_io.h"
 #include "iceberg/test/temp_file_test_base.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
@@ -121,6 +127,31 @@ class FileScanTaskReaderTest : public TempFileTestBase {
         table_schema_->schema_id());
   }
 
+  using RowWithPositionAndLineage = std::array<int64_t, 4>;
+
+  Result<std::vector<RowWithPositionAndLineage>> ReadRows(FileScanTaskReader& reader,
+                                                          const FileScanTask& task) {
+    ICEBERG_ASSIGN_OR_RAISE(auto stream, reader.Open(task));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batches,
+                                   ::arrow::ImportRecordBatchReader(&stream));
+    std::vector<RowWithPositionAndLineage> rows;
+    while (true) {
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, batches->Next());
+      if (batch == nullptr) {
+        return rows;
+      }
+      ICEBERG_ARROW_RETURN_NOT_OK(batch->ValidateFull());
+      const auto& ids = static_cast<const ::arrow::Int32Array&>(*batch->column(0));
+      const auto& positions = static_cast<const ::arrow::Int64Array&>(*batch->column(1));
+      const auto& row_ids = static_cast<const ::arrow::Int64Array&>(*batch->column(2));
+      const auto& sequences = static_cast<const ::arrow::Int64Array&>(*batch->column(3));
+      for (int64_t i = 0; i < batch->num_rows(); ++i) {
+        rows.push_back(
+            {ids.Value(i), positions.Value(i), row_ids.Value(i), sequences.Value(i)});
+      }
+    }
+  }
+
   Result<ExportedBatch> MakeBatch(const Schema& schema,
                                   const std::string& json_data) const {
     ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, MakeArrowSchema(schema));
@@ -409,7 +440,7 @@ TEST_F(FileScanTaskReaderTest, OpenWithoutDeletesReadsProjectedSchema) {
       VerifyStream(&stream, R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])"));
 }
 
-TEST_F(FileScanTaskReaderTest, RejectsPartialTaskUntilRangeReadsAreSupported) {
+TEST_F(FileScanTaskReaderTest, RejectsPartialTasksForUnsupportedFormats) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto data_file,
       MakeDataFile(table_schema_,
@@ -423,11 +454,146 @@ TEST_F(FileScanTaskReaderTest, RejectsPartialTaskUntilRangeReadsAreSupported) {
       .projected_schema = projected_schema_,
   };
   ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options)));
-  auto result = reader->Open(*task);
-  EXPECT_THAT(result, IsError(ErrorKind::kNotSupported));
-  EXPECT_THAT(result, HasErrorMessage("Reading a partial FileScanTask"));
+  for (auto format : {FileFormatType::kAvro, FileFormatType::kOrc}) {
+    data_file->file_format = format;
+    auto result = reader->Open(*task);
+    EXPECT_THAT(result, IsError(ErrorKind::kNotSupported));
+    EXPECT_THAT(result,
+                HasErrorMessage("Reading a partial FileScanTask requires Parquet"));
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, RechecksSplitBoundsAfterFileSizeChanges) {
+  ICEBERG_UNWRAP_OR_FAIL(auto data_file,
+                         MakeDataFile(table_schema_, R"([[1, "a", "b"]])", 1));
+  const int64_t file_size = data_file->file_size_in_bytes;
+  ICEBERG_UNWRAP_OR_FAIL(auto task, FileScanTask::MakeSplit(data_file, 4, file_size - 4));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make({
+                                          .io = file_io_,
+                                          .table_schema = table_schema_,
+                                          .projected_schema = projected_schema_,
+                                      }));
+  for (int64_t size : {file_size - 1, int64_t{3}, int64_t{0}}) {
+    data_file->file_size_in_bytes = size;
+    auto result = reader->Open(*task);
+    EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+    EXPECT_THAT(result, HasErrorMessage("Task range is outside file"));
+  }
+  data_file->file_size_in_bytes = -1;
+  EXPECT_THAT(reader->Open(*task),
+              HasErrorMessage("Data file size must not be negative"));
+}
+
+TEST_F(FileScanTaskReaderTest, EmptyParquetFileAndSplitReturnNoRows) {
+  ICEBERG_UNWRAP_OR_FAIL(auto data_file, MakeDataFile(table_schema_, "[]", 0));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make({
+                                          .io = file_io_,
+                                          .table_schema = table_schema_,
+                                          .projected_schema = projected_schema_,
+                                      }));
+  ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 0, 1));
+  for (const auto& task : {std::make_shared<FileScanTask>(data_file), split}) {
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*task));
+    auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+    auto batch = batches->Next();
+    ASSERT_TRUE(batch.ok()) << batch.status();
+    ASSERT_EQ(batch.ValueOrDie(), nullptr);
+  }
+}
+
+class FileScanTaskReaderSplitTest
+    : public FileScanTaskReaderTest,
+      public ::testing::WithParamInterface<std::tuple<std::string, int64_t>> {};
+
+TEST_P(FileScanTaskReaderSplitTest, SplitsMatchWholeFileWithAbsolutePositionsAndLineage) {
+  const auto& [delete_kind, batch_size] = GetParam();
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   std::vector<std::string>{R"([[1, "a", "keep"], [2, "b", "delete"]])",
+                                            R"([[3, "c", "delete"], [4, "d", "delete"]])",
+                                            R"([[5, "e", "keep"], [6, "f", "delete"]])"},
+                   /*record_count=*/6, /*max_row_group_length=*/2));
+  data_file->first_row_id = 100;
+  data_file->data_sequence_number = 7;
+  std::vector<std::shared_ptr<DataFile>> deletes;
+  if (delete_kind == "position" || delete_kind == "mixed") {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto deletion, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                              {1, 2, 3, 5}, data_file->file_path));
+    deletes.push_back(std::move(deletion));
+  }
+  if (delete_kind == "equality" || delete_kind == "mixed") {
+    // The equality field is absent from the output projection.
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto deletion,
+        MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), table_schema_,
+                               R"([[0, "", "delete"]])", {3}));
+    deletes.push_back(std::move(deletion));
+  }
+  if (delete_kind == "dv") {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto deletion, MakeDeletionVectorFile(CreateNewTempFilePathWithSuffix(".puffin"),
+                                              {1, 2, 3, 5}, data_file->file_path));
+    deletes.push_back(std::move(deletion));
+  }
+  auto projection = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()), MetadataColumns::kRowPosition,
+      MetadataColumns::kRowId, MetadataColumns::kLastUpdatedSequenceNumber});
+  // This FileIO honors the supplied length hint, so a split length passed as the
+  // file length would make the footer unreadable.
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader, FileScanTaskReader::Make({
+                       .io = std::make_shared<test::StdFileIO>(),
+                       .table_schema = table_schema_,
+                       .projected_schema = projection,
+                       .properties = {{"read.batch-size", std::to_string(batch_size)}},
+                   }));
+
+  std::vector<RowWithPositionAndLineage> expected;
+  for (int64_t id = 1; id <= 6; ++id) {
+    if (delete_kind == "none" || id == 1 || id == 5) {
+      expected.push_back({id, id - 1, 100 + id - 1, 7});
+    }
+  }
+  ICEBERG_UNWRAP_OR_FAIL(auto whole_rows,
+                         ReadRows(*reader, FileScanTask(data_file, deletes)));
+  ASSERT_EQ(whole_rows, expected);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto input,
+                         arrow::OpenArrowInputStream(file_io_, data_file->file_path));
+  auto metadata = ::parquet::ReadMetaData(input);
+  ASSERT_EQ(metadata->num_row_groups(), 3);
+  std::vector<int64_t> boundaries{0, 1, data_file->file_size_in_bytes};
+  for (int i = 0; i < metadata->num_row_groups(); ++i) {
+    auto column = metadata->RowGroup(i)->ColumnChunk(0);
+    ASSERT_TRUE(column->has_dictionary_page());
+    ASSERT_LT(column->dictionary_page_offset(), column->data_page_offset());
+    // Include exact starts, ranges containing no starts, and boundaries inside groups.
+    boundaries.push_back(column->dictionary_page_offset());
+    boundaries.push_back(column->dictionary_page_offset() + 1);
+    boundaries.push_back(column->data_page_offset());
+  }
+  std::sort(boundaries.begin(), boundaries.end());
+  boundaries.erase(std::unique(boundaries.begin(), boundaries.end()), boundaries.end());
+  std::vector<RowWithPositionAndLineage> split_rows;
+  for (size_t i = 1; i < boundaries.size(); ++i) {
+    SCOPED_TRACE(i);
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto split, FileScanTask::MakeSplit(data_file, boundaries[i - 1],
+                                            boundaries[i] - boundaries[i - 1], deletes));
+    ICEBERG_UNWRAP_OR_FAIL(auto rows, ReadRows(*reader, *split));
+    split_rows.insert(split_rows.end(), rows.begin(), rows.end());
+  }
+  EXPECT_EQ(split_rows, whole_rows);
+  EXPECT_EQ(split_rows, expected);
 }
 
+INSTANTIATE_TEST_SUITE_P(DeleteKindsAndBatchSizes, FileScanTaskReaderSplitTest,
+                         ::testing::Combine(::testing::Values("none", "position",
+                                                              "equality", "dv", "mixed"),
+                                            ::testing::Values(int64_t{1}, int64_t{100})));
+
 TEST_F(FileScanTaskReaderTest, ReadLastUpdatedFromDataSeq) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto data_file,
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index 9fbeae8789..4af203c55f 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -39,7 +39,9 @@
 #include <parquet/arrow/reader.h>
 #include <parquet/arrow/writer.h>
 #include <parquet/file_reader.h>
+#include <parquet/file_writer.h>
 #include <parquet/metadata.h>
+#include <parquet/schema.h>
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
@@ -47,6 +49,7 @@
 #include "iceberg/file_reader.h"
 #include "iceberg/file_writer.h"
 #include "iceberg/metadata_columns.h"
+#include "iceberg/parquet/parquet_data_util_internal.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -158,6 +161,31 @@ std::optional<ParquetCodec> FirstUnavailableParquetCodec() {
   return std::nullopt;
 }
 
+std::unique_ptr<::parquet::FileMetaData> MakeMetadataWithPageOffsets(
+    const std::vector<std::pair<int64_t, std::optional<int64_t>>>& page_offsets) {
+  ::parquet::schema::NodeVector fields;
+  for (size_t i = 0; i < page_offsets.size(); ++i) {
+    fields.push_back(::parquet::schema::PrimitiveNode::Make(
+        "col" + std::to_string(i), ::parquet::Repetition::REQUIRED, nullptr,
+        ::parquet::Type::INT32, /*primitive_length=*/-1, /*field_id=*/i + 1));
+  }
+  ::parquet::SchemaDescriptor schema;
+  schema.Init(::parquet::schema::GroupNode::Make(
+      "schema", ::parquet::Repetition::REQUIRED, fields));
+  auto builder = ::parquet::FileMetaDataBuilder::Make(
+      &schema, ::parquet::default_writer_properties());
+  auto row_group = builder->AppendRowGroup();
+  row_group->set_num_rows(1);
+  for (const auto& [data_offset, dictionary_offset] : page_offsets) {
+    row_group->NextColumnChunk()->Finish(
+        /*num_values=*/1, dictionary_offset.value_or(0), /*index_page_offset=*/0,
+        data_offset, /*compressed_size=*/10, /*uncompressed_size=*/10,
+        dictionary_offset.has_value(), /*dictionary_fallback=*/false, {}, {});
+  }
+  row_group->Finish(10 * page_offsets.size());
+  return builder->Finish();
+}
+
 constexpr std::array<uint8_t, Uuid::kLength> kUuidBytes1 = {
     0x12, 0x3e, 0x45, 0x67, 0xe8, 0x9b, 0x12, 0xd3,
     0xa4, 0x56, 0x42, 0x66, 0x14, 0x17, 0x40, 0x00};
@@ -167,6 +195,28 @@ constexpr std::array<uint8_t, Uuid::kLength> kUuidBytes2 = {
 
 }  // namespace
 
+TEST(ParquetSplitTest, CanonicalStartUsesFirstColumnPages) {
+  for (auto dictionary_offset :
+       {std::optional<int64_t>{}, std::optional<int64_t>{0}, std::optional<int64_t>{80},
+        std::optional<int64_t>{100}, std::optional<int64_t>{120}}) {
+    // The second column starts earlier and must not be used as the assignment key.
+    auto metadata = MakeMetadataWithPageOffsets({{100, dictionary_offset}, {50, 40}});
+    const int64_t expected = dictionary_offset == 80 ? 80 : 100;
+    EXPECT_THAT(RowGroupStart(*metadata->RowGroup(0), 200),
+                HasValue(::testing::Eq(expected)));
+  }
+}
+
+TEST(ParquetSplitTest, CanonicalStartRejectsMissingOrOutOfFilePages) {
+  for (int64_t offset : {-1, 0, 200, 201}) {
+    auto metadata = MakeMetadataWithPageOffsets({{offset, std::nullopt}});
+    EXPECT_THAT(RowGroupStart(*metadata->RowGroup(0), 200), IsError(ErrorKind::kInvalid));
+  }
+  auto no_columns = MakeMetadataWithPageOffsets({});
+  EXPECT_THAT(RowGroupStart(*no_columns->RowGroup(0), 200),
+              HasErrorMessage("without columns"));
+}
+
 class ParquetReaderTest : public TempFileTestBase {
  protected:
   static void SetUpTestSuite() { parquet::RegisterAll(); }
@@ -251,7 +301,7 @@ class ParquetReaderTest : public TempFileTestBase {
                                    .properties = std::move(writer_properties)}));
   }
 
-  void CreateSplitParquetFile() {
+  void CreateSplitParquetFile(int64_t row_group_size = 2, bool dictionary = true) {
     const std::string kParquetFieldIdKey = "PARQUET:field_id";
     auto arrow_schema = ::arrow::schema(
         {::arrow::field("id", ::arrow::int32(), /*nullable=*/false,
@@ -270,12 +320,30 @@ class ParquetReaderTest : public TempFileTestBase {
     auto io = internal::checked_cast<arrow::ArrowFileSystemFileIO&>(*file_io_);
     auto outfile = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie();
 
+    ::parquet::WriterProperties::Builder properties;
+    if (!dictionary) {
+      properties.disable_dictionary();
+    }
     ASSERT_TRUE(::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(),
-                                             outfile, /*chunk_size=*/2)
+                                             outfile, row_group_size, properties.build())
                     .ok());
     ASSERT_TRUE(outfile->Close().ok());
   }
 
+  void ReorderRowGroups(const std::vector<int>& row_groups) {
+    ICEBERG_UNWRAP_OR_FAIL(auto input,
+                           arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
+    auto metadata = ::parquet::ReadMetaData(input);
+    auto file_size = input->GetSize().ValueOrDie();
+    auto pages = input->ReadAt(0, file_size - metadata->size() - 8).ValueOrDie();
+    ASSERT_TRUE(input->Close().ok());
+    ICEBERG_UNWRAP_OR_FAIL(auto output,
+                           arrow::OpenArrowOutputStream(file_io_, temp_parquet_file_));
+    ASSERT_TRUE(output->Write(pages).ok());
+    ::parquet::WriteFileMetaData(*metadata->Subset(row_groups), output.get());
+    ASSERT_TRUE(output->Close().ok());
+  }
+
   std::shared_ptr<Schema> RowLineageSchema() {
     return std::make_shared<Schema>(std::vector<SchemaField>{
         SchemaField::MakeRequired(1, "id", int32()),
@@ -1015,7 +1083,10 @@ TEST_F(ParquetReaderTest, ReadSplit) {
   auto metadata = ::parquet::ReadMetaData(input_stream);
   std::vector<size_t> split_offsets;
   for (int i = 0; i < metadata->num_row_groups(); ++i) {
-    split_offsets.push_back(static_cast<size_t>(metadata->RowGroup(i)->file_offset()));
+    auto column = metadata->RowGroup(i)->ColumnChunk(0);
+    ASSERT_TRUE(column->has_dictionary_page());
+    ASSERT_LT(column->dictionary_page_offset(), column->data_page_offset());
+    split_offsets.push_back(static_cast<size_t>(column->dictionary_page_offset()));
   }
   ASSERT_EQ(split_offsets.size(), 2);
 
@@ -1030,9 +1101,17 @@ TEST_F(ParquetReaderTest, ReadSplit) {
       {.offset = split_offsets[1], .length = 1},
       {.offset = split_offsets[1] + 1, .length = std::numeric_limits<size_t>::max()},
       {.offset = 0, .length = split_offsets[0]},
+      {.offset = split_offsets[0] + 1, .length = split_offsets[1] - split_offsets[0]},
+      {.offset = split_offsets[0], .length = 0},
   };
   std::vector<std::string> expected_json = {
-      R"([[1, 0], [2, 1], [3, 2]])", R"([[1, 0], [2, 1]])", R"([[3, 2]])", "", "",
+      R"([[1, 0], [2, 1], [3, 2]])",
+      R"([[1, 0], [2, 1]])",
+      R"([[3, 2]])",
+      "",
+      "",
+      R"([[3, 2]])",
+      "",
   };
 
   ReaderProperties reader_properties;
@@ -1056,6 +1135,129 @@ TEST_F(ParquetReaderTest, ReadSplit) {
   }
 }
 
+TEST_F(ParquetReaderTest, ReadSplitWithoutDictionaryPages) {
+  ASSERT_NO_FATAL_FAILURE(
+      CreateSplitParquetFile(/*row_group_size=*/2, /*dictionary=*/false));
+  ICEBERG_UNWRAP_OR_FAIL(auto input,
+                         arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
+  auto metadata = ::parquet::ReadMetaData(input);
+  auto column = metadata->RowGroup(1)->ColumnChunk(0);
+  ASSERT_FALSE(column->has_dictionary_page());
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()), MetadataColumns::kRowPosition});
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader,
+      ReaderFactoryRegistry::Open(
+          FileFormatType::kParquet,
+          {.path = temp_parquet_file_,
+           .split = Split{.offset = static_cast<size_t>(column->data_page_offset()),
+                          .length = 1},
+           .io = file_io_,
+           .projection = schema}));
+  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[3, 2]])"));
+  ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
+}
+
+TEST_F(ParquetReaderTest, ReadSplitsWithOmittedRowGroupOffsets) {
+  file_io_ = arrow::ArrowFileSystemFileIO::MakeLocalFileIO();
+  temp_parquet_file_ = GetResourcePath("split-without-row-group-offsets.parquet");
+  ICEBERG_UNWRAP_OR_FAIL(auto input,
+                         arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
+  auto metadata = ::parquet::ReadMetaData(input);
+  ASSERT_EQ(metadata->num_row_groups(), 3);
+  for (int i = 0; i < metadata->num_row_groups(); ++i) {
+    ASSERT_EQ(metadata->RowGroup(i)->file_offset(), 0);
+  }
+  auto boundary = static_cast<size_t>(
+      metadata->RowGroup(1)->ColumnChunk(0)->dictionary_page_offset());
+  auto file_size = static_cast<size_t>(input->GetSize().ValueOrDie());
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()), MetadataColumns::kRowPosition,
+      MetadataColumns::kRowId, MetadataColumns::kLastUpdatedSequenceNumber});
+  const std::vector<std::optional<Split>> splits{
+      std::nullopt, Split{.offset = 0, .length = boundary},
+      Split{.offset = boundary, .length = file_size - boundary}};
+  const std::vector<std::string> expected{
+      R"([[1, 0, 100, 7], [2, 1, 101, 7], [3, 2, 102, 7], [4, 3, 103, 7], [5, 4, 104, 7], [6, 5, 105, 7]])",
+      R"([[1, 0, 100, 7], [2, 1, 101, 7]])",
+      R"([[3, 2, 102, 7], [4, 3, 103, 7], [5, 4, 104, 7], [6, 5, 105, 7]])"};
+  for (size_t i = 0; i < splits.size(); ++i) {
+    SCOPED_TRACE(i);
+    ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                           ReaderFactoryRegistry::Open(FileFormatType::kParquet,
+                                                       {.path = temp_parquet_file_,
+                                                        .length = file_size,
+                                                        .split = splits[i],
+                                                        .io = file_io_,
+                                                        .projection = schema,
+                                                        .first_row_id = 100,
+                                                        .data_sequence_number = 7}));
+    ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, expected[i]));
+    ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
+  }
+}
+
+TEST_F(ParquetReaderTest, RejectsSplitWithInvalidPageOffsetsBeforeReading) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "col0", int32())});
+  for (int64_t offset : {int64_t{-1}, int64_t{0}, std::numeric_limits<int64_t>::max()}) {
+    auto metadata = MakeMetadataWithPageOffsets({{offset, std::nullopt}});
+    ICEBERG_UNWRAP_OR_FAIL(auto output,
+                           arrow::OpenArrowOutputStream(file_io_, temp_parquet_file_));
+    ::parquet::WriteMetaDataFile(*metadata, output.get());
+    ASSERT_TRUE(output->Close().ok());
+    auto reader = ReaderFactoryRegistry::Open(FileFormatType::kParquet,
+                                              {.path = temp_parquet_file_,
+                                               .split = Split{.offset = 0, .length = 1},
+                                               .io = file_io_,
+                                               .projection = schema});
+    EXPECT_THAT(reader, IsError(ErrorKind::kInvalid));
+    EXPECT_THAT(reader, HasErrorMessage("row group 0"));
+    EXPECT_THAT(reader, HasErrorMessage(temp_parquet_file_));
+  }
+}
+
+TEST_F(ParquetReaderTest, ReadSplitWithNonmonotonicRowGroupStarts) {
+  ASSERT_NO_FATAL_FAILURE(CreateSplitParquetFile(/*row_group_size=*/1));
+  // Logical row order becomes [1, 3, 2], while the pages stay in physical order [1, 2,
+  // 3].
+  ASSERT_NO_FATAL_FAILURE(ReorderRowGroups({0, 2, 1}));
+  ICEBERG_UNWRAP_OR_FAIL(auto input,
+                         arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
+  auto metadata = ::parquet::ReadMetaData(input);
+  auto last_start = static_cast<size_t>(
+      metadata->RowGroup(1)->ColumnChunk(0)->dictionary_page_offset());
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()), MetadataColumns::kRowPosition,
+      MetadataColumns::kRowId, MetadataColumns::kLastUpdatedSequenceNumber});
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader,
+      ReaderFactoryRegistry::Open(FileFormatType::kParquet,
+                                  {.path = temp_parquet_file_,
+                                   .split = Split{.offset = 0, .length = last_start},
+                                   .io = file_io_,
+                                   .projection = schema,
+                                   .first_row_id = 100,
+                                   .data_sequence_number = 7}));
+  // A batch must not span the skipped row group, or generated positions would be wrong.
+  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[1, 0, 100, 7]])"));
+  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[2, 2, 102, 7]])"));
+  ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto tail,
+      ReaderFactoryRegistry::Open(FileFormatType::kParquet,
+                                  {.path = temp_parquet_file_,
+                                   .split = Split{.offset = last_start, .length = 1},
+                                   .io = file_io_,
+                                   .projection = schema,
+                                   .first_row_id = 100,
+                                   .data_sequence_number = 7}));
+  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*tail, R"([[3, 1, 101, 7]])"));
+  ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*tail));
+}
+
 TEST_F(ParquetReaderTest, ReadMetadataColumns) {
   temp_parquet_file_ = "metadata_columns.parquet";
   CreateSimpleParquetFile();
diff --git a/src/iceberg/test/resources/split-without-row-group-offsets.md b/src/iceberg/test/resources/split-without-row-group-offsets.md
new file mode 100644
index 0000000000..c6ad010961
--- /dev/null
+++ b/src/iceberg/test/resources/split-without-row-group-offsets.md
@@ -0,0 +1,45 @@
+<!--
+  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.
+-->
+
+# Parquet split fixture with omitted row-group offsets
+
+`split-without-row-group-offsets.parquet` contains a required `int32` column
+named `id`, with Iceberg field ID 1 and values 1 through 6. Its three row groups
+contain two rows each, with dictionary pages before their first data pages.
+The optional `RowGroup.file_offset` field is absent from every row group.
+
+The fixture was generated with PyArrow 12.0.1 using the following write options:
+
+```python
+schema = pa.schema([
+    pa.field("id", pa.int32(), nullable=False, metadata={b"PARQUET:field_id": b"1"})
+])
+table = pa.Table.from_arrays([pa.array([1, 2, 3, 4, 5, 6], type=pa.int32())], schema=schema)
+pq.write_table(table, path, row_group_size=2, compression="NONE",
+               use_dictionary=True, version="2.6", store_schema=False)
+```
+
+The footer was then decoded with Thrift's compact protocol, field 5
+(`file_offset`) was removed from each entry of `FileMetaData.row_groups`, and
+the footer was serialized again. The original data pages were retained and
+the four-byte footer length was updated. A PyArrow read of the resulting file
+was checked against the original table.
+
+This fixture tests that split selection uses the first column's page offsets
+even when the optional row-group offset is unavailable.
diff --git a/src/iceberg/test/resources/split-without-row-group-offsets.parquet b/src/iceberg/test/resources/split-without-row-group-offsets.parquet
new file mode 100644
index 0000000000000000000000000000000000000000..fc84a7c3c92c4ecef7f4025fb067e1f45824cbf1
GIT binary patch
literal 797
zcmb`GzfQwI490zqb7JTaAuhU;ZphM8hqh`FWn$^hP$41asEyPP(2(+H<pDbK7(4(E
z!obL*U^6t(QdlD8bn$2T^G|j%8D9(ynZzXLe8C}r0s!P=z-XhpUrB!PmPEBA`4l?e
zA{JDH;aPcuZj^v_78f%EDWz-0PdEt&f0a9F0tTgJ!WZ}cK@_n<qV5ohYDw}!R7Wfb
z(QCAi=pDlgGE-f$W$+(U%q=vZT_#a2NnR}Lhy`K#i1#r)#o>*ZsV><v_>IYRM_~18
zIxntsnIN`7hJT5Ww%nQXO>RZDNvCUd1i;3P4tHaXV%_MBHr524tjw$1`{J%QEz4eB
eRm;chp{Q1i<xMs`?hpC{=RNP~iofv*{}`Wf97+oS

literal 0
HcmV?d00001


From 760fa0812340c58542761aab3a0fbe7212b49ad4 Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Mon, 12 Oct 2026 00:23:32 +0000
Subject: [PATCH 02/11] docs(parquet): explain row-group split selection

Co-authored-by: Isaac <no-reply@databricks.com>
---
 src/iceberg/parquet/parquet_data_util.cc | 3 +++
 src/iceberg/parquet/parquet_reader.cc    | 5 +++++
 2 files changed, 8 insertions(+)

diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc
index b18ff6b8a9..6f9be73bd6 100644
--- a/src/iceberg/parquet/parquet_data_util.cc
+++ b/src/iceberg/parquet/parquet_data_util.cc
@@ -45,12 +45,15 @@ Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
   }
   auto column = row_group.ColumnChunk(0);
   int64_t start = column->data_page_offset();
+  // A dictionary page comes before the data pages. Like parquet-mr, ignore it
+  // unless its offset is positive and earlier.
   if (column->has_dictionary_page()) {
     const int64_t dictionary_start = column->dictionary_page_offset();
     if (dictionary_start > 0 && dictionary_start < start) {
       start = dictionary_start;
     }
   }
+  // Byte 0 is the "PAR1" magic, so no page can start there.
   if (start <= 0 || start >= file_size) {
     return Invalid(
         "Cannot split a Parquet row group with missing or out-of-file "
diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index f1130afeff..b2cde7b4aa 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -322,6 +322,7 @@ class ParquetReader::Impl {
       ICEBERG_RETURN_UNEXPECTED(InitReadContext());
     }
 
+    // Each row-group range has its own batch reader. When one runs out, open the next.
     std::shared_ptr<::arrow::RecordBatch> batch;
     while (true) {
       ICEBERG_ARROW_ASSIGN_OR_RETURN(batch, context_->record_batch_reader_->Next());
@@ -411,6 +412,7 @@ class ParquetReader::Impl {
     if (options.split.has_value()) {
       const auto& split = *options.split;
       ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file_size, input_stream_->GetSize());
+      // Rows in all earlier row groups, selected or not, so _pos is file-relative.
       int64_t row_position = 0;
       for (int i = 0; i < metadata->num_row_groups(); ++i) {
         auto row_group = metadata->RowGroup(i);
@@ -428,6 +430,8 @@ class ParquetReader::Impl {
         // Subtraction avoids overflow for a caller-supplied unbounded split length.
         if (std::cmp_greater_equal(*start, split.offset) &&
             static_cast<uint64_t>(*start) - split.offset < split.length) {
+          // Extend the last range if this row group follows it; otherwise start a
+          // new range so its _pos can jump past the skipped row groups.
           if (!row_group_ranges_.empty() && row_group_ranges_.back().end == i) {
             row_group_ranges_.back().end = i + 1;
           } else {
@@ -465,6 +469,7 @@ class ParquetReader::Impl {
   }
 
   Status InitReadContext() {
+    // Build locally so a failure leaves context_ unset.
     auto context = std::make_unique<ReadContext>();
     ICEBERG_ASSIGN_OR_RAISE(context->record_batch_reader_, OpenNextRowGroupRange());
 

From acf80e65425fad121910a1f2a2b918735bbd223f Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Mon, 12 Oct 2026 07:31:16 +0000
Subject: [PATCH 03/11] style(test): use ranges algorithms for split boundaries

Co-authored-by: Isaac <no-reply@databricks.com>
---
 src/iceberg/test/file_scan_task_reader_test.cc | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc
index c59da128ed..05416bd7d7 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -574,8 +574,9 @@ TEST_P(FileScanTaskReaderSplitTest, SplitsMatchWholeFileWithAbsolutePositionsAnd
     boundaries.push_back(column->dictionary_page_offset() + 1);
     boundaries.push_back(column->data_page_offset());
   }
-  std::sort(boundaries.begin(), boundaries.end());
-  boundaries.erase(std::unique(boundaries.begin(), boundaries.end()), boundaries.end());
+  std::ranges::sort(boundaries);
+  auto duplicates = std::ranges::unique(boundaries);
+  boundaries.erase(duplicates.begin(), duplicates.end());
   std::vector<RowWithPositionAndLineage> split_rows;
   for (size_t i = 1; i < boundaries.size(); ++i) {
     SCOPED_TRACE(i);

From 1a765a8bfac10d346ba17009c51924eb0fbbbe62 Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Tue, 13 Oct 2026 18:12:36 +0000
Subject: [PATCH 04/11] fix(parquet): consume a row-group range only after its
 reader is installed

InitReadContext advanced to the next row-group range before the output
schema was built. If schema setup failed, a retried Next() skipped the
first range. Opening a range no longer consumes it; ConsumeRowGroupRange()
does that once the reader is in place.

Co-authored-by: Isaac <no-reply@databricks.com>
---
 src/iceberg/parquet/parquet_reader.cc | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index b2cde7b4aa..d2058b1e66 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -334,6 +334,7 @@ class ParquetReader::Impl {
       }
       ICEBERG_ARROW_RETURN_NOT_OK(context_->record_batch_reader_->Close());
       ICEBERG_ASSIGN_OR_RAISE(context_->record_batch_reader_, OpenNextRowGroupRange());
+      ConsumeRowGroupRange();
     }
 
     std::shared_ptr<::arrow::Int64Array> row_positions;
@@ -453,6 +454,8 @@ class ParquetReader::Impl {
     return {};
   }
 
+  // Open the next range without consuming it, so a caller that fails afterwards can
+  // retry it. Call ConsumeRowGroupRange() once the reader is installed.
   Result<std::unique_ptr<::arrow::RecordBatchReader>> OpenNextRowGroupRange() {
     if (next_row_group_range_ == row_group_ranges_.size()) {
       return std::make_unique<EmptyRecordBatchReader>();
@@ -463,13 +466,20 @@ class ParquetReader::Impl {
     ICEBERG_ARROW_ASSIGN_OR_RETURN(
         auto batch_reader, reader_->GetRecordBatchReader(
                                row_group_indices, SelectedColumnIndices(projection_)));
-    metadata_context_.next_file_pos = range.first_row_position;
-    ++next_row_group_range_;
     return batch_reader;
   }
 
+  void ConsumeRowGroupRange() {
+    if (next_row_group_range_ == row_group_ranges_.size()) {
+      return;
+    }
+    metadata_context_.next_file_pos =
+        row_group_ranges_[next_row_group_range_].first_row_position;
+    ++next_row_group_range_;
+  }
+
   Status InitReadContext() {
-    // Build locally so a failure leaves context_ unset.
+    // Build locally so a failure leaves context_ unset and the first range unconsumed.
     auto context = std::make_unique<ReadContext>();
     ICEBERG_ASSIGN_OR_RAISE(context->record_batch_reader_, OpenNextRowGroupRange());
 
@@ -505,6 +515,7 @@ class ParquetReader::Impl {
         context->output_arrow_schema_, context->record_batch_reader_->schema(),
         projection_, use_large_list_);
     context_ = std::move(context);
+    ConsumeRowGroupRange();
     return {};
   }
 

From 9aec9ff1c961165578a6eed918d65964e60ec0ca Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Tue, 13 Oct 2026 18:12:36 +0000
Subject: [PATCH 05/11] feat(parquet): assign row groups to splits by midpoint
 like Java Iceberg

Java Iceberg reads a split through parquet-java's withRange(), which keeps
a row group when its midpoint (first-column start plus half the compressed
size) falls in [start, start + length). Use the same rule so C++ and Java
readers pick the same row groups for splits that cut through a row group.

Co-authored-by: Isaac <no-reply@databricks.com>
---
 src/iceberg/data/file_scan_task_reader.h      |  2 +-
 src/iceberg/parquet/parquet_data_util.cc      | 30 ++++++++
 .../parquet/parquet_data_util_internal.h      |  6 ++
 src/iceberg/parquet/parquet_reader.cc         | 11 +--
 src/iceberg/test/parquet_test.cc              | 73 ++++++++++++++-----
 5 files changed, 96 insertions(+), 26 deletions(-)

diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h
index faa62750c6..33a4b82fff 100644
--- a/src/iceberg/data/file_scan_task_reader.h
+++ b/src/iceberg/data/file_scan_task_reader.h
@@ -69,7 +69,7 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
   /// \brief Open a task and return an Arrow C stream for its projected live rows.
   ///
   /// Partial tasks are supported for Parquet, with each row group assigned to the
-  /// half-open range containing its first column's starting page. Row positions and
+  /// half-open range containing its midpoint, as in Java Iceberg. Row positions and
   /// inherited row IDs remain relative to the whole file. Other formats require
   /// whole-file tasks. The task and its shared data/delete-file metadata must not be
   /// mutated while the stream uses them.
diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc
index 6f9be73bd6..f598c77e6a 100644
--- a/src/iceberg/parquet/parquet_data_util.cc
+++ b/src/iceberg/parquet/parquet_data_util.cc
@@ -63,6 +63,36 @@ Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
   return start;
 }
 
+Result<int64_t> RowGroupMidpoint(const ::parquet::RowGroupMetaData& row_group,
+                                 int64_t file_size) {
+  ICEBERG_ASSIGN_OR_RAISE(auto start, RowGroupStart(row_group, file_size));
+  // The row group's pages are contiguous from its start, so its size is bounded by the
+  // rest of the file. This bound also keeps the sum below from overflowing.
+  const int64_t max_size = file_size - start;
+  // Arrow reports an omitted row-group size as 0. Like parquet-java, fall back to the
+  // sum of the column chunk sizes.
+  int64_t size = row_group.total_compressed_size();
+  if (size == 0) {
+    for (int i = 0; i < row_group.num_columns(); ++i) {
+      const int64_t column_size = row_group.ColumnChunk(i)->total_compressed_size();
+      if (column_size < 0 || column_size > max_size - size) {
+        return Invalid(
+            "Cannot split a Parquet row group with invalid column {} compressed size {} "
+            "(start {}, file size {})",
+            i, column_size, start, file_size);
+      }
+      size += column_size;
+    }
+  }
+  if (size < 0 || size > max_size) {
+    return Invalid(
+        "Cannot split a Parquet row group with invalid compressed size {} "
+        "(start {}, file size {})",
+        size, start, file_size);
+  }
+  return start + size / 2;
+}
+
 namespace {
 
 // Forward declaration
diff --git a/src/iceberg/parquet/parquet_data_util_internal.h b/src/iceberg/parquet/parquet_data_util_internal.h
index 95ab0230ed..7f36765723 100644
--- a/src/iceberg/parquet/parquet_data_util_internal.h
+++ b/src/iceberg/parquet/parquet_data_util_internal.h
@@ -39,6 +39,12 @@ bool NeedsRowPositions(const StructType& struct_type,
 Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
                               int64_t file_size);
 
+/// \brief Return the byte offset that assigns a row group to a split: its start plus
+/// half its compressed size, matching parquet-java's range filter used by Java Iceberg.
+/// Reject sizes that are negative or extend past the end of the file.
+Result<int64_t> RowGroupMidpoint(const ::parquet::RowGroupMetaData& row_group,
+                                 int64_t file_size);
+
 /// \brief Convert record batch read from a Parquet file to projected Iceberg Schema.
 ///
 /// \param record_batch The record batch to convert.
diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index d2058b1e66..7640d975cb 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -421,16 +421,17 @@ class ParquetReader::Impl {
         if (row_group->num_rows() == 0) {
           continue;
         }
-        auto start = RowGroupStart(*row_group, file_size);
-        if (!start.has_value()) {
+        // Like Java Iceberg, a split reads the row groups whose midpoints it contains.
+        auto midpoint = RowGroupMidpoint(*row_group, file_size);
+        if (!midpoint.has_value()) {
           return Invalid(
               "Cannot read Parquet split {} (start={}, length={}), row "
               "group {}: {}",
-              options.path, split.offset, split.length, i, start.error().message);
+              options.path, split.offset, split.length, i, midpoint.error().message);
         }
         // Subtraction avoids overflow for a caller-supplied unbounded split length.
-        if (std::cmp_greater_equal(*start, split.offset) &&
-            static_cast<uint64_t>(*start) - split.offset < split.length) {
+        if (std::cmp_greater_equal(*midpoint, split.offset) &&
+            static_cast<uint64_t>(*midpoint) - split.offset < split.length) {
           // Extend the last range if this row group follows it; otherwise start a
           // new range so its _pos can jump past the skipped row groups.
           if (!row_group_ranges_.empty() && row_group_ranges_.back().end == i) {
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index 4af203c55f..ede91dea49 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -217,6 +217,23 @@ TEST(ParquetSplitTest, CanonicalStartRejectsMissingOrOutOfFilePages) {
               HasErrorMessage("without columns"));
 }
 
+TEST(ParquetSplitTest, MidpointAddsHalfTheRowGroupSize) {
+  // Each column chunk is 10 bytes, so the row group is 20 bytes from its start at 80.
+  auto metadata = MakeMetadataWithPageOffsets({{100, 80}, {50, 40}});
+  ASSERT_EQ(metadata->RowGroup(0)->total_compressed_size(), 20);
+  EXPECT_THAT(RowGroupMidpoint(*metadata->RowGroup(0), 200), HasValue(::testing::Eq(90)));
+  // The size may reach, but not pass, the end of the file.
+  EXPECT_THAT(RowGroupMidpoint(*metadata->RowGroup(0), 100), HasValue(::testing::Eq(90)));
+  EXPECT_THAT(RowGroupMidpoint(*metadata->RowGroup(0), 99),
+              HasErrorMessage("invalid compressed size 20"));
+}
+
+TEST(ParquetSplitTest, MidpointRejectsInvalidStart) {
+  auto metadata = MakeMetadataWithPageOffsets({{0, std::nullopt}});
+  EXPECT_THAT(RowGroupMidpoint(*metadata->RowGroup(0), 200),
+              HasErrorMessage("starting page 0"));
+}
+
 class ParquetReaderTest : public TempFileTestBase {
  protected:
   static void SetUpTestSuite() { parquet::RegisterAll(); }
@@ -1077,18 +1094,24 @@ TEST_F(ParquetReaderTest, EmptySplitKeepsPrunedNestedListTypes) {
 TEST_F(ParquetReaderTest, ReadSplit) {
   CreateSplitParquetFile();
 
-  // Read split offsets
+  // Read row-group starts and midpoints
   auto io = internal::checked_cast<arrow::ArrowFileSystemFileIO&>(*file_io_);
   auto input_stream = io.fs()->OpenInputFile(temp_parquet_file_).ValueOrDie();
   auto metadata = ::parquet::ReadMetaData(input_stream);
-  std::vector<size_t> split_offsets;
+  const int64_t file_size = input_stream->GetSize().ValueOrDie();
+  std::vector<size_t> starts;
+  std::vector<size_t> midpoints;
   for (int i = 0; i < metadata->num_row_groups(); ++i) {
     auto column = metadata->RowGroup(i)->ColumnChunk(0);
     ASSERT_TRUE(column->has_dictionary_page());
     ASSERT_LT(column->dictionary_page_offset(), column->data_page_offset());
-    split_offsets.push_back(static_cast<size_t>(column->dictionary_page_offset()));
+    starts.push_back(static_cast<size_t>(column->dictionary_page_offset()));
+    ICEBERG_UNWRAP_OR_FAIL(auto midpoint,
+                           RowGroupMidpoint(*metadata->RowGroup(i), file_size));
+    ASSERT_GT(static_cast<size_t>(midpoint), starts.back());
+    midpoints.push_back(static_cast<size_t>(midpoint));
   }
-  ASSERT_EQ(split_offsets.size(), 2);
+  ASSERT_EQ(starts.size(), 2);
 
   auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "id", int32()),
@@ -1097,12 +1120,15 @@ TEST_F(ParquetReaderTest, ReadSplit) {
 
   std::vector<Split> splits = {
       {.offset = 0, .length = std::numeric_limits<size_t>::max()},
-      {.offset = split_offsets[0], .length = split_offsets[1] - split_offsets[0]},
-      {.offset = split_offsets[1], .length = 1},
-      {.offset = split_offsets[1] + 1, .length = std::numeric_limits<size_t>::max()},
-      {.offset = 0, .length = split_offsets[0]},
-      {.offset = split_offsets[0] + 1, .length = split_offsets[1] - split_offsets[0]},
-      {.offset = split_offsets[0], .length = 0},
+      {.offset = starts[0], .length = starts[1] - starts[0]},
+      {.offset = midpoints[1], .length = 1},
+      {.offset = midpoints[1] + 1, .length = std::numeric_limits<size_t>::max()},
+      {.offset = 0, .length = midpoints[0]},
+      {.offset = midpoints[0] + 1, .length = midpoints[1] - midpoints[0]},
+      {.offset = starts[0], .length = 0},
+      // A row group belongs to the split containing its midpoint, not its start.
+      {.offset = starts[1], .length = 1},
+      {.offset = starts[0] + 1, .length = midpoints[0] - starts[0]},
   };
   std::vector<std::string> expected_json = {
       R"([[1, 0], [2, 1], [3, 2]])",
@@ -1112,6 +1138,8 @@ TEST_F(ParquetReaderTest, ReadSplit) {
       "",
       R"([[3, 2]])",
       "",
+      "",
+      R"([[1, 0], [2, 1]])",
   };
 
   ReaderProperties reader_properties;
@@ -1143,6 +1171,11 @@ TEST_F(ParquetReaderTest, ReadSplitWithoutDictionaryPages) {
   auto metadata = ::parquet::ReadMetaData(input);
   auto column = metadata->RowGroup(1)->ColumnChunk(0);
   ASSERT_FALSE(column->has_dictionary_page());
+  const int64_t file_size = input->GetSize().ValueOrDie();
+  EXPECT_THAT(RowGroupStart(*metadata->RowGroup(1), file_size),
+              HasValue(::testing::Eq(column->data_page_offset())));
+  ICEBERG_UNWRAP_OR_FAIL(auto midpoint,
+                         RowGroupMidpoint(*metadata->RowGroup(1), file_size));
   auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "id", int32()), MetadataColumns::kRowPosition});
   ICEBERG_UNWRAP_OR_FAIL(
@@ -1150,8 +1183,7 @@ TEST_F(ParquetReaderTest, ReadSplitWithoutDictionaryPages) {
       ReaderFactoryRegistry::Open(
           FileFormatType::kParquet,
           {.path = temp_parquet_file_,
-           .split = Split{.offset = static_cast<size_t>(column->data_page_offset()),
-                          .length = 1},
+           .split = Split{.offset = static_cast<size_t>(midpoint), .length = 1},
            .io = file_io_,
            .projection = schema}));
   ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[3, 2]])"));
@@ -1246,14 +1278,15 @@ TEST_F(ParquetReaderTest, ReadSplitWithNonmonotonicRowGroupStarts) {
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
 
   ICEBERG_UNWRAP_OR_FAIL(
-      auto tail,
-      ReaderFactoryRegistry::Open(FileFormatType::kParquet,
-                                  {.path = temp_parquet_file_,
-                                   .split = Split{.offset = last_start, .length = 1},
-                                   .io = file_io_,
-                                   .projection = schema,
-                                   .first_row_id = 100,
-                                   .data_sequence_number = 7}));
+      auto tail, ReaderFactoryRegistry::Open(
+                     FileFormatType::kParquet,
+                     {.path = temp_parquet_file_,
+                      .split = Split{.offset = last_start,
+                                     .length = std::numeric_limits<size_t>::max()},
+                      .io = file_io_,
+                      .projection = schema,
+                      .first_row_id = 100,
+                      .data_sequence_number = 7}));
   ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*tail, R"([[3, 1, 101, 7]])"));
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*tail));
 }

From 7c4939df6b8938ac97819b5aafab68b005b1b1ce Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Fri, 16 Oct 2026 21:49:23 +0000
Subject: [PATCH 06/11] fix comment from part-1

---
 src/iceberg/table_scan.cc               |  9 +++++----
 src/iceberg/table_scan.h                |  2 --
 src/iceberg/test/file_scan_task_test.cc | 17 +++++++++++++++++
 3 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index 25548bc769..a2f695e036 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -224,7 +224,7 @@ Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
 
   return std::shared_ptr<FileScanTask>(
       new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter),
-                       Range{.start = start, .length = length, .file_size = file_size}));
+                       Range{.start = start, .length = length}));
 }
 
 int64_t FileScanTask::length() const {
@@ -241,16 +241,17 @@ int64_t FileScanTask::estimated_row_count() const {
   }
 
   const int64_t record_count = data_file_->record_count;
-  if (record_count <= 0) {
+  const int64_t file_size = data_file_->file_size_in_bytes;
+  if (record_count <= 0 || file_size <= 0) {
     return 0;
   }
 
   // floor(end * rows / size) - floor(start * rows / size), so splits that cover
   // the file sum exactly to record_count. int128 avoids overflow.
   const auto end = static_cast<int128_t>(range_->start) + range_->length;
-  const auto count_at_end = end * record_count / range_->file_size;
+  const auto count_at_end = end * record_count / file_size;
   const auto count_at_start =
-      static_cast<int128_t>(range_->start) * record_count / range_->file_size;
+      static_cast<int128_t>(range_->start) * record_count / file_size;
   return static_cast<int64_t>(count_at_end - count_at_start);
 }
 
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index 7afaa79d3b..4d55c8607a 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -119,8 +119,6 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   struct Range {
     int64_t start;
     int64_t length;
-    // Size when the range was validated. Always positive, so safe to divide by.
-    int64_t file_size;
   };
 
   FileScanTask(std::shared_ptr<DataFile> data_file,
diff --git a/src/iceberg/test/file_scan_task_test.cc b/src/iceberg/test/file_scan_task_test.cc
index 6394c96fed..4170d7f12c 100644
--- a/src/iceberg/test/file_scan_task_test.cc
+++ b/src/iceberg/test/file_scan_task_test.cc
@@ -155,6 +155,23 @@ TEST(FileScanTaskRangeTest, RowEstimateDoesNotOverflow) {
   EXPECT_EQ(last->estimated_row_count(), 1);
 }
 
+TEST(FileScanTaskRangeTest, RowEstimateHandlesChangedFileSize) {
+  auto data_file = std::make_shared<DataFile>();
+  data_file->file_path = "test.parquet";
+  data_file->file_size_in_bytes = 100;
+  data_file->record_count = 7;
+  ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 0, 40));
+
+  data_file->file_size_in_bytes = 200;
+  EXPECT_EQ(split->estimated_row_count(), 1);
+
+  data_file->file_size_in_bytes = 0;
+  EXPECT_EQ(split->estimated_row_count(), 0);
+
+  data_file->file_size_in_bytes = -1;
+  EXPECT_EQ(split->estimated_row_count(), 0);
+}
+
 class FileScanTaskTest : public TempFileTestBase {
  protected:
   static void SetUpTestSuite() { parquet::RegisterAll(); }

From 7d6323cb4c3ea7d382e1bbf441943576f77c8e89 Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Fri, 16 Oct 2026 23:00:05 +0000
Subject: [PATCH 07/11] fix(scan): simplify split readers and bound stale row
 estimates

---
 src/iceberg/parquet/parquet_reader.cc   | 98 ++++++-------------------
 src/iceberg/table_scan.cc               |  4 +-
 src/iceberg/test/file_scan_task_test.cc | 11 +++
 src/iceberg/test/parquet_test.cc        | 39 +++++++++-
 4 files changed, 74 insertions(+), 78 deletions(-)

diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index 7640d975cb..12da76a9ee 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -265,13 +265,6 @@ struct ReadContext {
   std::unique_ptr<RowNumberReader> row_number_reader_;
 };
 
-// Consecutive row groups can share a batch reader without introducing gaps in _pos.
-struct RowGroupRange {
-  int begin;
-  int end;
-  int64_t first_row_position;
-};
-
 // TODO(gangwu): list of work items
 // 1. Make the memory pool configurable
 // 2. Catch ParquetException and convert to Status/Result
@@ -305,7 +298,7 @@ class ParquetReader::Impl {
     ICEBERG_ARROW_ASSIGN_OR_RETURN(
         reader_, ::parquet::arrow::FileReader::Make(pool_, std::move(file_reader),
                                                     arrow_reader_properties));
-    ICEBERG_RETURN_UNEXPECTED(PlanRowGroupRanges(options));
+    ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options));
 
     // Project read schema onto the Parquet file schema
     ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_));
@@ -322,19 +315,9 @@ class ParquetReader::Impl {
       ICEBERG_RETURN_UNEXPECTED(InitReadContext());
     }
 
-    // Each row-group range has its own batch reader. When one runs out, open the next.
-    std::shared_ptr<::arrow::RecordBatch> batch;
-    while (true) {
-      ICEBERG_ARROW_ASSIGN_OR_RETURN(batch, context_->record_batch_reader_->Next());
-      if (batch) {
-        break;
-      }
-      if (next_row_group_range_ == row_group_ranges_.size()) {
-        return std::nullopt;
-      }
-      ICEBERG_ARROW_RETURN_NOT_OK(context_->record_batch_reader_->Close());
-      ICEBERG_ASSIGN_OR_RAISE(context_->record_batch_reader_, OpenNextRowGroupRange());
-      ConsumeRowGroupRange();
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, context_->record_batch_reader_->Next());
+    if (!batch) {
+      return std::nullopt;
     }
 
     std::shared_ptr<::arrow::Int64Array> row_positions;
@@ -407,14 +390,13 @@ class ParquetReader::Impl {
   }
 
  private:
-  Status PlanRowGroupRanges(const ReaderOptions& options) {
+  Status PlanRowGroups(const ReaderOptions& options) {
     // TODO(gangwu): add row group filtering based on zone map, bloom filter, etc.
     auto metadata = reader_->parquet_reader()->metadata();
     if (options.split.has_value()) {
       const auto& split = *options.split;
       ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file_size, input_stream_->GetSize());
-      // Rows in all earlier row groups, selected or not, so _pos is file-relative.
-      int64_t row_position = 0;
+      int64_t row_count = 0;
       for (int i = 0; i < metadata->num_row_groups(); ++i) {
         auto row_group = metadata->RowGroup(i);
         // Empty row groups have no pages to assign to a split.
@@ -432,68 +414,38 @@ class ParquetReader::Impl {
         // Subtraction avoids overflow for a caller-supplied unbounded split length.
         if (std::cmp_greater_equal(*midpoint, split.offset) &&
             static_cast<uint64_t>(*midpoint) - split.offset < split.length) {
-          // Extend the last range if this row group follows it; otherwise start a
-          // new range so its _pos can jump past the skipped row groups.
-          if (!row_group_ranges_.empty() && row_group_ranges_.back().end == i) {
-            row_group_ranges_.back().end = i + 1;
-          } else {
-            row_group_ranges_.push_back(
-                {.begin = i, .end = i + 1, .first_row_position = row_position});
-          }
+          row_group_indices_.push_back(i);
         }
         if (row_group->num_rows() < 0 ||
-            row_group->num_rows() > std::numeric_limits<int64_t>::max() - row_position) {
+            row_group->num_rows() > std::numeric_limits<int64_t>::max() - row_count) {
           return Invalid("Invalid Parquet row count in file {}, row group {}",
                          options.path, i);
         }
-        row_position += row_group->num_rows();
+        row_count += row_group->num_rows();
       }
     } else if (metadata->num_row_groups() > 0) {
-      row_group_ranges_.push_back(
-          {.begin = 0, .end = metadata->num_row_groups(), .first_row_position = 0});
+      row_group_indices_.resize(metadata->num_row_groups());
+      std::iota(row_group_indices_.begin(), row_group_indices_.end(), 0);  // NOLINT
     }
     return {};
   }
 
-  // Open the next range without consuming it, so a caller that fails afterwards can
-  // retry it. Call ConsumeRowGroupRange() once the reader is installed.
-  Result<std::unique_ptr<::arrow::RecordBatchReader>> OpenNextRowGroupRange() {
-    if (next_row_group_range_ == row_group_ranges_.size()) {
-      return std::make_unique<EmptyRecordBatchReader>();
-    }
-    const auto& range = row_group_ranges_[next_row_group_range_];
-    std::vector<int> row_group_indices(range.end - range.begin);
-    std::iota(row_group_indices.begin(), row_group_indices.end(), range.begin);  // NOLINT
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(
-        auto batch_reader, reader_->GetRecordBatchReader(
-                               row_group_indices, SelectedColumnIndices(projection_)));
-    return batch_reader;
-  }
-
-  void ConsumeRowGroupRange() {
-    if (next_row_group_range_ == row_group_ranges_.size()) {
-      return;
-    }
-    metadata_context_.next_file_pos =
-        row_group_ranges_[next_row_group_range_].first_row_position;
-    ++next_row_group_range_;
-  }
-
   Status InitReadContext() {
-    // Build locally so a failure leaves context_ unset and the first range unconsumed.
+    // Build locally so initialization can be retried without consuming any rows.
     auto context = std::make_unique<ReadContext>();
-    ICEBERG_ASSIGN_OR_RAISE(context->record_batch_reader_, OpenNextRowGroupRange());
+    // Create the record batch reader, including for empty splits to preserve the schema.
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(
+        context->record_batch_reader_,
+        reader_->GetRecordBatchReader(row_group_indices_,
+                                      SelectedColumnIndices(projection_)));
 
     if (NeedsRowPositions(*read_schema_, metadata_context_)) {
       auto metadata = reader_->parquet_reader()->metadata();
-      std::vector<int> row_group_indices;
-      for (const auto& range : row_group_ranges_) {
-        for (int i = range.begin; i < range.end; ++i) {
-          row_group_indices.push_back(i);
-        }
-      }
-      ICEBERG_ASSIGN_OR_RAISE(context->row_number_reader_,
-                              RowNumberReader::Make(*metadata, row_group_indices, pool_));
+      // Share the batch reader's selection so positions remain file-relative even
+      // when a batch spans nonconsecutive row groups.
+      ICEBERG_ASSIGN_OR_RAISE(
+          context->row_number_reader_,
+          RowNumberReader::Make(*metadata, row_group_indices_, pool_));
     }
 
     // Build the output Arrow schema from the projected Iceberg schema. This schema is the
@@ -516,16 +468,14 @@ class ParquetReader::Impl {
         context->output_arrow_schema_, context->record_batch_reader_->schema(),
         projection_, use_large_list_);
     context_ = std::move(context);
-    ConsumeRowGroupRange();
     return {};
   }
 
  private:
   // TODO(gangwu): make memory pool configurable
   ::arrow::MemoryPool* pool_ = ::arrow::default_memory_pool();
-  // Selected row groups in file row order, separated at gaps to preserve absolute _pos.
-  std::vector<RowGroupRange> row_group_ranges_;
-  size_t next_row_group_range_ = 0;
+  // Row groups in footer order, shared by the batch and row-number readers.
+  std::vector<int> row_group_indices_;
   // Whether to read list columns as large_list (64-bit offsets).
   bool use_large_list_ = false;
   // Schema to read from the Parquet file.
diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index a2f695e036..2696ccf60e 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -242,7 +242,9 @@ int64_t FileScanTask::estimated_row_count() const {
 
   const int64_t record_count = data_file_->record_count;
   const int64_t file_size = data_file_->file_size_in_bytes;
-  if (record_count <= 0 || file_size <= 0) {
+  if (record_count <= 0 || file_size <= 0 || range_->start > file_size ||
+      range_->length > file_size - range_->start) {
+    // Shared metadata may have changed since MakeSplit validated the range.
     return 0;
   }
 
diff --git a/src/iceberg/test/file_scan_task_test.cc b/src/iceberg/test/file_scan_task_test.cc
index 4170d7f12c..65156c9d29 100644
--- a/src/iceberg/test/file_scan_task_test.cc
+++ b/src/iceberg/test/file_scan_task_test.cc
@@ -165,6 +165,17 @@ TEST(FileScanTaskRangeTest, RowEstimateHandlesChangedFileSize) {
   data_file->file_size_in_bytes = 200;
   EXPECT_EQ(split->estimated_row_count(), 1);
 
+  data_file->file_size_in_bytes = 40;
+  EXPECT_EQ(split->estimated_row_count(), 7);
+
+  data_file->file_size_in_bytes = 39;
+  EXPECT_EQ(split->estimated_row_count(), 0);
+
+  // An invalidated range must not produce an estimate larger than INT64_MAX.
+  data_file->file_size_in_bytes = 1;
+  data_file->record_count = std::numeric_limits<int64_t>::max();
+  EXPECT_EQ(split->estimated_row_count(), 0);
+
   data_file->file_size_in_bytes = 0;
   EXPECT_EQ(split->estimated_row_count(), 0);
 
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index ede91dea49..0482173878 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -1272,9 +1272,9 @@ TEST_F(ParquetReaderTest, ReadSplitWithNonmonotonicRowGroupStarts) {
                                    .projection = schema,
                                    .first_row_id = 100,
                                    .data_sequence_number = 7}));
-  // A batch must not span the skipped row group, or generated positions would be wrong.
-  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[1, 0, 100, 7]])"));
-  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[2, 2, 102, 7]])"));
+  // A batch can span skipped row groups while keeping absolute positions and lineage.
+  ASSERT_NO_FATAL_FAILURE(
+      VerifyNextBatch(*reader, R"([[1, 0, 100, 7], [2, 2, 102, 7]])"));
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
 
   ICEBERG_UNWRAP_OR_FAIL(
@@ -1291,6 +1291,39 @@ TEST_F(ParquetReaderTest, ReadSplitWithNonmonotonicRowGroupStarts) {
   ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*tail));
 }
 
+TEST_F(ParquetReaderTest, ReadMetadataOnlySplitWithNonmonotonicRowGroups) {
+  ASSERT_NO_FATAL_FAILURE(CreateSplitParquetFile(/*row_group_size=*/1));
+  ASSERT_NO_FATAL_FAILURE(ReorderRowGroups({0, 2, 1}));
+  ICEBERG_UNWRAP_OR_FAIL(auto input,
+                         arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
+  auto metadata = ::parquet::ReadMetaData(input);
+  auto last_start = static_cast<size_t>(
+      metadata->RowGroup(1)->ColumnChunk(0)->dictionary_page_offset());
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{MetadataColumns::kRowPosition, MetadataColumns::kRowId,
+                               MetadataColumns::kLastUpdatedSequenceNumber});
+
+  for (int64_t batch_size : {1, 100}) {
+    SCOPED_TRACE(batch_size);
+    ReaderProperties properties;
+    properties.Set(ReaderProperties::kBatchSize, batch_size);
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto reader,
+        ReaderFactoryRegistry::Open(FileFormatType::kParquet,
+                                    {.path = temp_parquet_file_,
+                                     .split = Split{.offset = 0, .length = last_start},
+                                     .io = file_io_,
+                                     .projection = schema,
+                                     .first_row_id = 100,
+                                     .data_sequence_number = 7,
+                                     .properties = properties}));
+    // Arrow emits one batch per row group when no physical columns are read.
+    ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[0, 100, 7]])"));
+    ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, R"([[2, 102, 7]])"));
+    ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
+  }
+}
+
 TEST_F(ParquetReaderTest, ReadMetadataColumns) {
   temp_parquet_file_ = "metadata_columns.parquet";
   CreateSimpleParquetFile();

From 34139a01964066954a37705f23c3ceb568bbebf1 Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Sun, 18 Oct 2026 02:03:08 +0000
Subject: [PATCH 08/11] fix(scan): preserve split estimates when file size
 changes

---
 src/iceberg/table_scan.cc               | 10 +++----
 src/iceberg/table_scan.h                |  2 ++
 src/iceberg/test/file_scan_task_test.cc | 37 ++++++++++++-------------
 3 files changed, 23 insertions(+), 26 deletions(-)

diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index 2696ccf60e..36b342989a 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -224,7 +224,7 @@ Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
 
   return std::shared_ptr<FileScanTask>(
       new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter),
-                       Range{.start = start, .length = length}));
+                       Range{.start = start, .length = length, .file_size = file_size}));
 }
 
 int64_t FileScanTask::length() const {
@@ -241,15 +241,13 @@ int64_t FileScanTask::estimated_row_count() const {
   }
 
   const int64_t record_count = data_file_->record_count;
-  const int64_t file_size = data_file_->file_size_in_bytes;
-  if (record_count <= 0 || file_size <= 0 || range_->start > file_size ||
-      range_->length > file_size - range_->start) {
-    // Shared metadata may have changed since MakeSplit validated the range.
+  if (record_count <= 0) {
     return 0;
   }
 
+  const int64_t file_size = range_->file_size;
   // floor(end * rows / size) - floor(start * rows / size), so splits that cover
-  // the file sum exactly to record_count. int128 avoids overflow.
+  // the original file size sum exactly to record_count. int128 avoids overflow.
   const auto end = static_cast<int128_t>(range_->start) + range_->length;
   const auto count_at_end = end * record_count / file_size;
   const auto count_at_start =
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index 4d55c8607a..5afe453055 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -119,6 +119,8 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   struct Range {
     int64_t start;
     int64_t length;
+    // Use the size at split creation to keep row estimates consistent across splits.
+    int64_t file_size;
   };
 
   FileScanTask(std::shared_ptr<DataFile> data_file,
diff --git a/src/iceberg/test/file_scan_task_test.cc b/src/iceberg/test/file_scan_task_test.cc
index 65156c9d29..08105bea49 100644
--- a/src/iceberg/test/file_scan_task_test.cc
+++ b/src/iceberg/test/file_scan_task_test.cc
@@ -155,32 +155,29 @@ TEST(FileScanTaskRangeTest, RowEstimateDoesNotOverflow) {
   EXPECT_EQ(last->estimated_row_count(), 1);
 }
 
-TEST(FileScanTaskRangeTest, RowEstimateHandlesChangedFileSize) {
+TEST(FileScanTaskRangeTest, RowEstimatesPreserveSplitTotalAfterFileSizeChanges) {
   auto data_file = std::make_shared<DataFile>();
   data_file->file_path = "test.parquet";
   data_file->file_size_in_bytes = 100;
   data_file->record_count = 7;
-  ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 0, 40));
-
-  data_file->file_size_in_bytes = 200;
-  EXPECT_EQ(split->estimated_row_count(), 1);
-
-  data_file->file_size_in_bytes = 40;
-  EXPECT_EQ(split->estimated_row_count(), 7);
-
-  data_file->file_size_in_bytes = 39;
-  EXPECT_EQ(split->estimated_row_count(), 0);
+  ICEBERG_UNWRAP_OR_FAIL(auto first, FileScanTask::MakeSplit(data_file, 0, 40));
+  ICEBERG_UNWRAP_OR_FAIL(auto second, FileScanTask::MakeSplit(data_file, 40, 60));
+
+  for (int64_t file_size : {200, 40, 39, 1, 0, -1}) {
+    SCOPED_TRACE(file_size);
+    data_file->file_size_in_bytes = file_size;
+    EXPECT_EQ(first->estimated_row_count(), 2);
+    EXPECT_EQ(second->estimated_row_count(), 5);
+    EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(),
+              data_file->record_count);
+  }
 
-  // An invalidated range must not produce an estimate larger than INT64_MAX.
-  data_file->file_size_in_bytes = 1;
+  // The captured size also keeps estimates bounded when the row count is large.
   data_file->record_count = std::numeric_limits<int64_t>::max();
-  EXPECT_EQ(split->estimated_row_count(), 0);
-
-  data_file->file_size_in_bytes = 0;
-  EXPECT_EQ(split->estimated_row_count(), 0);
-
-  data_file->file_size_in_bytes = -1;
-  EXPECT_EQ(split->estimated_row_count(), 0);
+  EXPECT_GE(first->estimated_row_count(), 0);
+  EXPECT_GE(second->estimated_row_count(), 0);
+  EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(),
+            data_file->record_count);
 }
 
 class FileScanTaskTest : public TempFileTestBase {

From c7cfe94d157e46d00ff01987f2d4c50fc79e0b4e Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Sun, 18 Oct 2026 02:23:39 +0000
Subject: [PATCH 09/11] fix(scan): cache split row estimates at task creation

---
 src/iceberg/table_scan.cc               | 36 +++++++++++--------------
 src/iceberg/table_scan.h                | 10 +++----
 src/iceberg/test/file_scan_task_test.cc | 28 ++++++++++++++-----
 3 files changed, 42 insertions(+), 32 deletions(-)

diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index 36b342989a..0ac2d1b738 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -222,9 +222,21 @@ Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
                            data_file->file_path);
   }
 
-  return std::shared_ptr<FileScanTask>(
-      new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter),
-                       Range{.start = start, .length = length, .file_size = file_size}));
+  int64_t estimated_row_count = 0;
+  const int64_t record_count = data_file->record_count;
+  if (record_count > 0) {
+    // floor(end * rows / size) - floor(start * rows / size), so splits that cover
+    // the file sum exactly to record_count at creation. int128 avoids overflow.
+    const auto end = static_cast<int128_t>(start) + length;
+    const auto count_at_end = end * record_count / file_size;
+    const auto count_at_start = static_cast<int128_t>(start) * record_count / file_size;
+    estimated_row_count = static_cast<int64_t>(count_at_end - count_at_start);
+  }
+
+  return std::shared_ptr<FileScanTask>(new FileScanTask(
+      std::move(data_file), std::move(delete_files), std::move(filter),
+      Range{
+          .start = start, .length = length, .estimated_row_count = estimated_row_count}));
 }
 
 int64_t FileScanTask::length() const {
@@ -236,23 +248,7 @@ int64_t FileScanTask::size_bytes() const { return length(); }
 int32_t FileScanTask::files_count() const { return 1; }
 
 int64_t FileScanTask::estimated_row_count() const {
-  if (!range_) {
-    return data_file_->record_count;
-  }
-
-  const int64_t record_count = data_file_->record_count;
-  if (record_count <= 0) {
-    return 0;
-  }
-
-  const int64_t file_size = range_->file_size;
-  // floor(end * rows / size) - floor(start * rows / size), so splits that cover
-  // the original file size sum exactly to record_count. int128 avoids overflow.
-  const auto end = static_cast<int128_t>(range_->start) + range_->length;
-  const auto count_at_end = end * record_count / file_size;
-  const auto count_at_start =
-      static_cast<int128_t>(range_->start) * record_count / file_size;
-  return static_cast<int64_t>(count_at_end - count_at_start);
+  return range_ ? range_->estimated_row_count : data_file_->record_count;
 }
 
 // ChangelogScanTask implementation
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index 5afe453055..4bdd14511f 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -81,9 +81,10 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   /// \brief Make a task for a validated half-open byte range [start, start + length).
   ///
   /// A range covering the entire file is represented as a whole-file task. An empty
-  /// range is valid only for an empty whole file. The data file, delete files, and
-  /// residual filter are shared with the resulting task and must not be mutated
-  /// while a planning or reading stream uses it.
+  /// range is valid only for an empty whole file. Partial tasks retain the row
+  /// estimate calculated at creation. The data file, delete files, and residual
+  /// filter are shared with the resulting task and must not be mutated while a
+  /// planning or reading stream uses it.
   static Result<std::shared_ptr<FileScanTask>> MakeSplit(
       std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
       std::vector<std::shared_ptr<DataFile>> delete_files = {},
@@ -119,8 +120,7 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   struct Range {
     int64_t start;
     int64_t length;
-    // Use the size at split creation to keep row estimates consistent across splits.
-    int64_t file_size;
+    int64_t estimated_row_count;
   };
 
   FileScanTask(std::shared_ptr<DataFile> data_file,
diff --git a/src/iceberg/test/file_scan_task_test.cc b/src/iceberg/test/file_scan_task_test.cc
index 08105bea49..a5baa6945b 100644
--- a/src/iceberg/test/file_scan_task_test.cc
+++ b/src/iceberg/test/file_scan_task_test.cc
@@ -155,7 +155,7 @@ TEST(FileScanTaskRangeTest, RowEstimateDoesNotOverflow) {
   EXPECT_EQ(last->estimated_row_count(), 1);
 }
 
-TEST(FileScanTaskRangeTest, RowEstimatesPreserveSplitTotalAfterFileSizeChanges) {
+TEST(FileScanTaskRangeTest, RowEstimatesAreFixedAtSplitCreation) {
   auto data_file = std::make_shared<DataFile>();
   data_file->file_path = "test.parquet";
   data_file->file_size_in_bytes = 100;
@@ -172,12 +172,26 @@ TEST(FileScanTaskRangeTest, RowEstimatesPreserveSplitTotalAfterFileSizeChanges)
               data_file->record_count);
   }
 
-  // The captured size also keeps estimates bounded when the row count is large.
-  data_file->record_count = std::numeric_limits<int64_t>::max();
-  EXPECT_GE(first->estimated_row_count(), 0);
-  EXPECT_GE(second->estimated_row_count(), 0);
-  EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(),
-            data_file->record_count);
+  for (int64_t record_count :
+       {int64_t{0}, int64_t{-1}, std::numeric_limits<int64_t>::max()}) {
+    SCOPED_TRACE(record_count);
+    data_file->record_count = record_count;
+    EXPECT_EQ(first->estimated_row_count(), 2);
+    EXPECT_EQ(second->estimated_row_count(), 5);
+    EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(), 7);
+  }
+}
+
+TEST(FileScanTaskRangeTest, RowEstimatesHandleNonPositiveRecordCounts) {
+  auto data_file = std::make_shared<DataFile>();
+  data_file->file_size_in_bytes = 100;
+
+  for (int64_t record_count : {0, -1}) {
+    SCOPED_TRACE(record_count);
+    data_file->record_count = record_count;
+    ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 0, 40));
+    EXPECT_EQ(split->estimated_row_count(), 0);
+  }
 }
 
 class FileScanTaskTest : public TempFileTestBase {

From be10adce1570aa3f21cb99ac23c3642406b5dc6b Mon Sep 17 00:00:00 2001
From: Kam Cheung Ting <kamcheung.ting@databricks.com>
Date: Sun, 18 Oct 2026 02:59:15 +0000
Subject: [PATCH 10/11] fix(scan): require stable metadata for row estimates

Require shared metadata to remain unchanged for a task's lifetime, with
tasks rebuilt after metadata changes. Remove the cached row estimate and
derive it from DataFile. Limit the row-count sum guarantee to valid,
non-overlapping splits covering a file with unchanged metadata.

Keep defensive bounds checks for invalidated tasks and distinguish those
checks from the supported metadata contract in the regression tests.

Validation: 472 tests passed across scan_test, data_test, and parquet_test;
10 tests skipped. pre-commit passed.
---
 src/iceberg/table_scan.cc               | 39 +++++++++++++++----------
 src/iceberg/table_scan.h                | 18 +++++++-----
 src/iceberg/test/file_scan_task_test.cc | 32 +++++++-------------
 3 files changed, 44 insertions(+), 45 deletions(-)

diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index 0ac2d1b738..d04e3385b6 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -222,21 +222,9 @@ Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
                            data_file->file_path);
   }
 
-  int64_t estimated_row_count = 0;
-  const int64_t record_count = data_file->record_count;
-  if (record_count > 0) {
-    // floor(end * rows / size) - floor(start * rows / size), so splits that cover
-    // the file sum exactly to record_count at creation. int128 avoids overflow.
-    const auto end = static_cast<int128_t>(start) + length;
-    const auto count_at_end = end * record_count / file_size;
-    const auto count_at_start = static_cast<int128_t>(start) * record_count / file_size;
-    estimated_row_count = static_cast<int64_t>(count_at_end - count_at_start);
-  }
-
-  return std::shared_ptr<FileScanTask>(new FileScanTask(
-      std::move(data_file), std::move(delete_files), std::move(filter),
-      Range{
-          .start = start, .length = length, .estimated_row_count = estimated_row_count}));
+  return std::shared_ptr<FileScanTask>(
+      new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter),
+                       Range{.start = start, .length = length}));
 }
 
 int64_t FileScanTask::length() const {
@@ -248,7 +236,26 @@ int64_t FileScanTask::size_bytes() const { return length(); }
 int32_t FileScanTask::files_count() const { return 1; }
 
 int64_t FileScanTask::estimated_row_count() const {
-  return range_ ? range_->estimated_row_count : data_file_->record_count;
+  if (!range_) {
+    return data_file_->record_count;
+  }
+
+  const int64_t record_count = data_file_->record_count;
+  const int64_t file_size = data_file_->file_size_in_bytes;
+  if (record_count <= 0 || file_size <= 0 || range_->start > file_size ||
+      range_->length > file_size - range_->start) {
+    // Keep arithmetic safe even for tasks invalidated by metadata mutation.
+    return 0;
+  }
+
+  // floor(end * rows / size) - floor(start * rows / size). For non-overlapping
+  // splits covering a file with unchanged metadata, estimates sum to record_count.
+  // int128 avoids overflow.
+  const auto end = static_cast<int128_t>(range_->start) + range_->length;
+  const auto count_at_end = end * record_count / file_size;
+  const auto count_at_start =
+      static_cast<int128_t>(range_->start) * record_count / file_size;
+  return static_cast<int64_t>(count_at_end - count_at_start);
 }
 
 // ChangelogScanTask implementation
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index 4bdd14511f..1307089e39 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -72,8 +72,8 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   /// \param data_file The data file to read.
   /// \param delete_files Delete files that apply to this data file.
   /// \param filter Optional residual filter to apply after reading.
-  /// \note The data file, delete files, and residual filter must not be mutated while
-  /// a planning or reading stream uses this task.
+  /// \note The shared data file, delete files, and residual filter must remain
+  /// unchanged for this task's lifetime. Metadata changes require rebuilding tasks.
   explicit FileScanTask(std::shared_ptr<DataFile> data_file,
                         std::vector<std::shared_ptr<DataFile>> delete_files = {},
                         std::shared_ptr<Expression> filter = nullptr);
@@ -81,10 +81,9 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   /// \brief Make a task for a validated half-open byte range [start, start + length).
   ///
   /// A range covering the entire file is represented as a whole-file task. An empty
-  /// range is valid only for an empty whole file. Partial tasks retain the row
-  /// estimate calculated at creation. The data file, delete files, and residual
-  /// filter are shared with the resulting task and must not be mutated while a
-  /// planning or reading stream uses it.
+  /// range is valid only for an empty whole file. The data file, delete files, and
+  /// residual filter are shared with the resulting task and must remain unchanged
+  /// for its lifetime. Metadata changes require rebuilding tasks.
   static Result<std::shared_ptr<FileScanTask>> MakeSplit(
       std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
       std::vector<std::shared_ptr<DataFile>> delete_files = {},
@@ -93,7 +92,7 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   /// \brief The inclusive byte offset where this task starts, or zero for a whole file.
   int64_t start() const { return range_ ? range_->start : 0; }
 
-  /// \brief The byte length of this task, or the current size of a whole file.
+  /// \brief The byte length of this task, or the size of a whole file.
   int64_t length() const;
 
   /// \brief Whether this task reads a proper subset of its data file.
@@ -113,6 +112,10 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   Kind kind() const override { return Kind::kFileScanTask; }
   int64_t size_bytes() const override;
   int32_t files_count() const override;
+
+  /// \brief The estimated number of rows in this task, prorated by byte range.
+  /// For valid, non-overlapping splits covering a file with unchanged metadata,
+  /// the estimates sum to the file's record_count.
   int64_t estimated_row_count() const override;
 
  private:
@@ -120,7 +123,6 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   struct Range {
     int64_t start;
     int64_t length;
-    int64_t estimated_row_count;
   };
 
   FileScanTask(std::shared_ptr<DataFile> data_file,
diff --git a/src/iceberg/test/file_scan_task_test.cc b/src/iceberg/test/file_scan_task_test.cc
index a5baa6945b..41861caec9 100644
--- a/src/iceberg/test/file_scan_task_test.cc
+++ b/src/iceberg/test/file_scan_task_test.cc
@@ -64,9 +64,10 @@ TEST(FileScanTaskRangeTest, WholeFileAndEmptyFile) {
   EXPECT_EQ(explicit_whole->length(), 100);
   EXPECT_EQ(explicit_whole->estimated_row_count(), 7);
 
-  data_file->file_size_in_bytes = 0;
-  data_file->record_count = 0;
-  ICEBERG_UNWRAP_OR_FAIL(auto empty, FileScanTask::MakeSplit(data_file, 0, 0));
+  auto empty_file = std::make_shared<DataFile>();
+  empty_file->file_size_in_bytes = 0;
+  empty_file->record_count = 0;
+  ICEBERG_UNWRAP_OR_FAIL(auto empty, FileScanTask::MakeSplit(empty_file, 0, 0));
   EXPECT_FALSE(empty->is_split());
   EXPECT_EQ(empty->start(), 0);
   EXPECT_EQ(empty->length(), 0);
@@ -155,30 +156,19 @@ TEST(FileScanTaskRangeTest, RowEstimateDoesNotOverflow) {
   EXPECT_EQ(last->estimated_row_count(), 1);
 }
 
-TEST(FileScanTaskRangeTest, RowEstimatesAreFixedAtSplitCreation) {
+TEST(FileScanTaskRangeTest, RowEstimateIsBoundedForInvalidatedRanges) {
   auto data_file = std::make_shared<DataFile>();
   data_file->file_path = "test.parquet";
   data_file->file_size_in_bytes = 100;
-  data_file->record_count = 7;
-  ICEBERG_UNWRAP_OR_FAIL(auto first, FileScanTask::MakeSplit(data_file, 0, 40));
-  ICEBERG_UNWRAP_OR_FAIL(auto second, FileScanTask::MakeSplit(data_file, 40, 60));
+  data_file->record_count = std::numeric_limits<int64_t>::max();
+  ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 40, 60));
 
-  for (int64_t file_size : {200, 40, 39, 1, 0, -1}) {
+  // Mutation violates the task's lifetime contract, but the estimate should still
+  // avoid overflow and division by zero when the range is invalidated.
+  for (int64_t file_size : {99, 40, 39, 1, 0, -1}) {
     SCOPED_TRACE(file_size);
     data_file->file_size_in_bytes = file_size;
-    EXPECT_EQ(first->estimated_row_count(), 2);
-    EXPECT_EQ(second->estimated_row_count(), 5);
-    EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(),
-              data_file->record_count);
-  }
-
-  for (int64_t record_count :
-       {int64_t{0}, int64_t{-1}, std::numeric_limits<int64_t>::max()}) {
-    SCOPED_TRACE(record_count);
-    data_file->record_count = record_count;
-    EXPECT_EQ(first->estimated_row_count(), 2);
-    EXPECT_EQ(second->estimated_row_count(), 5);
-    EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(), 7);
+    EXPECT_EQ(split->estimated_row_count(), 0);
   }
 }
 

From e6787c0f1c8493e91490beaea965db8340c9f8c8 Mon Sep 17 00:00:00 2001
From: Gang Wu <ustcwg@gmail.com>
Date: Wed, 21 Oct 2026 10:44:13 +0800
Subject: [PATCH 11/11] test(parquet): generate split fixture at runtime

---
 src/iceberg/parquet/parquet_data_util.cc      |   2 +-
 .../parquet/parquet_data_util_internal.h      |   2 +-
 src/iceberg/test/parquet_test.cc              |  62 ++++++++++++++++--
 .../split-without-row-group-offsets.md        |  45 -------------
 .../split-without-row-group-offsets.parquet   | Bin 797 -> 0 bytes
 5 files changed, 60 insertions(+), 51 deletions(-)
 delete mode 100644 src/iceberg/test/resources/split-without-row-group-offsets.md
 delete mode 100644 src/iceberg/test/resources/split-without-row-group-offsets.parquet

diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc
index f598c77e6a..73597cbfb8 100644
--- a/src/iceberg/parquet/parquet_data_util.cc
+++ b/src/iceberg/parquet/parquet_data_util.cc
@@ -45,7 +45,7 @@ Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
   }
   auto column = row_group.ColumnChunk(0);
   int64_t start = column->data_page_offset();
-  // A dictionary page comes before the data pages. Like parquet-mr, ignore it
+  // A dictionary page comes before the data pages. Like parquet-java, ignore it
   // unless its offset is positive and earlier.
   if (column->has_dictionary_page()) {
     const int64_t dictionary_start = column->dictionary_page_offset();
diff --git a/src/iceberg/parquet/parquet_data_util_internal.h b/src/iceberg/parquet/parquet_data_util_internal.h
index 7f36765723..afd0ea397a 100644
--- a/src/iceberg/parquet/parquet_data_util_internal.h
+++ b/src/iceberg/parquet/parquet_data_util_internal.h
@@ -34,7 +34,7 @@ namespace iceberg::parquet {
 bool NeedsRowPositions(const StructType& struct_type,
                        const arrow::MetadataColumnContext& metadata_context);
 
-/// \brief Return the first column's starting page, matching parquet-mr's
+/// \brief Return the first column's starting page, matching parquet-java's
 /// BlockMetaData.getStartingPos(). Reject missing or out-of-file starts.
 Result<int64_t> RowGroupStart(const ::parquet::RowGroupMetaData& row_group,
                               int64_t file_size);
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index 0482173878..1327f03f5f 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -58,7 +58,6 @@
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/std_io.h"
 #include "iceberg/test/temp_file_test_base.h"
-#include "iceberg/test/test_resource.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/macros.h"
@@ -1191,14 +1190,69 @@ TEST_F(ParquetReaderTest, ReadSplitWithoutDictionaryPages) {
 }
 
 TEST_F(ParquetReaderTest, ReadSplitsWithOmittedRowGroupOffsets) {
-  file_io_ = arrow::ArrowFileSystemFileIO::MakeLocalFileIO();
-  temp_parquet_file_ = GetResourcePath("split-without-row-group-offsets.parquet");
+  auto arrow_schema = ::arrow::schema(
+      {::arrow::field("id", ::arrow::int32(), /*nullable=*/false,
+                      ::arrow::KeyValueMetadata::Make({"PARQUET:field_id"}, {"1"}))});
+  auto table = ::arrow::Table::Make(
+      arrow_schema,
+      {::arrow::json::ArrayFromJSONString(::arrow::int32(), "[1, 2, 3, 4, 5, 6]")
+           .ValueOrDie()});
+  ::parquet::WriterProperties::Builder properties_builder;
+  properties_builder.version(::parquet::ParquetVersion::PARQUET_2_6)
+      ->compression(::parquet::Compression::UNCOMPRESSED);
+  auto properties = properties_builder.build();
+  ICEBERG_UNWRAP_OR_FAIL(auto original_output,
+                         arrow::OpenArrowOutputStream(file_io_, temp_parquet_file_));
+  ASSERT_TRUE(::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(),
+                                           original_output, /*chunk_size=*/2, properties)
+                  .ok());
+  ASSERT_TRUE(original_output->Close().ok());
+  ICEBERG_UNWRAP_OR_FAIL(auto original_input,
+                         arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
+  auto original_metadata = ::parquet::ReadMetaData(original_input);
+  auto original_size = original_input->GetSize().ValueOrDie();
+  auto pages = original_input->ReadAt(0, original_size - original_metadata->size() - 8)
+                   .ValueOrDie();
+  ASSERT_TRUE(original_input->Close().ok());
+
+  auto builder =
+      ::parquet::FileMetaDataBuilder::Make(original_metadata->schema(), properties);
+  for (int i = 0; i < original_metadata->num_row_groups(); ++i) {
+    auto original_row_group = original_metadata->RowGroup(i);
+    auto row_group = builder->AppendRowGroup();
+    row_group->set_num_rows(original_row_group->num_rows());
+    for (int j = 0; j < original_row_group->num_columns(); ++j) {
+      auto column = original_row_group->ColumnChunk(j);
+      row_group->NextColumnChunk()->Finish(
+          column->num_values(), column->dictionary_page_offset(),
+          /*index_page_offset=*/0, column->data_page_offset(),
+          column->total_compressed_size(), column->total_uncompressed_size(),
+          column->has_dictionary_page(), /*dictionary_fallback=*/false, {}, {});
+    }
+    // RowGroupMetaDataBuilder::Finish fills in the optional file_offset and
+    // total_compressed_size. Leave them unset to exercise legacy footer handling.
+  }
+  auto rebuilt_metadata = builder->Finish(original_metadata->key_value_metadata());
+  ICEBERG_UNWRAP_OR_FAIL(auto output,
+                         arrow::OpenArrowOutputStream(file_io_, temp_parquet_file_));
+  ASSERT_TRUE(output->Write(pages).ok());
+  ::parquet::WriteFileMetaData(*rebuilt_metadata, output.get());
+  ASSERT_TRUE(output->Close().ok());
+
   ICEBERG_UNWRAP_OR_FAIL(auto input,
                          arrow::OpenArrowInputStream(file_io_, temp_parquet_file_));
   auto metadata = ::parquet::ReadMetaData(input);
+  ASSERT_EQ(metadata->num_columns(), 1);
+  ASSERT_EQ(metadata->num_rows(), 6);
   ASSERT_EQ(metadata->num_row_groups(), 3);
   for (int i = 0; i < metadata->num_row_groups(); ++i) {
-    ASSERT_EQ(metadata->RowGroup(i)->file_offset(), 0);
+    auto row_group = metadata->RowGroup(i);
+    ASSERT_EQ(row_group->num_rows(), 2);
+    ASSERT_EQ(row_group->file_offset(), 0);
+    ASSERT_EQ(row_group->total_compressed_size(), 0);
+    auto column = row_group->ColumnChunk(0);
+    ASSERT_TRUE(column->has_dictionary_page());
+    ASSERT_LT(column->dictionary_page_offset(), column->data_page_offset());
   }
   auto boundary = static_cast<size_t>(
       metadata->RowGroup(1)->ColumnChunk(0)->dictionary_page_offset());
diff --git a/src/iceberg/test/resources/split-without-row-group-offsets.md b/src/iceberg/test/resources/split-without-row-group-offsets.md
deleted file mode 100644
index c6ad010961..0000000000
--- a/src/iceberg/test/resources/split-without-row-group-offsets.md
+++ /dev/null
@@ -1,45 +0,0 @@
-<!--
-  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.
--->
-
-# Parquet split fixture with omitted row-group offsets
-
-`split-without-row-group-offsets.parquet` contains a required `int32` column
-named `id`, with Iceberg field ID 1 and values 1 through 6. Its three row groups
-contain two rows each, with dictionary pages before their first data pages.
-The optional `RowGroup.file_offset` field is absent from every row group.
-
-The fixture was generated with PyArrow 12.0.1 using the following write options:
-
-```python
-schema = pa.schema([
-    pa.field("id", pa.int32(), nullable=False, metadata={b"PARQUET:field_id": b"1"})
-])
-table = pa.Table.from_arrays([pa.array([1, 2, 3, 4, 5, 6], type=pa.int32())], schema=schema)
-pq.write_table(table, path, row_group_size=2, compression="NONE",
-               use_dictionary=True, version="2.6", store_schema=False)
-```
-
-The footer was then decoded with Thrift's compact protocol, field 5
-(`file_offset`) was removed from each entry of `FileMetaData.row_groups`, and
-the footer was serialized again. The original data pages were retained and
-the four-byte footer length was updated. A PyArrow read of the resulting file
-was checked against the original table.
-
-This fixture tests that split selection uses the first column's page offsets
-even when the optional row-group offset is unavailable.
diff --git a/src/iceberg/test/resources/split-without-row-group-offsets.parquet b/src/iceberg/test/resources/split-without-row-group-offsets.parquet
deleted file mode 100644
index fc84a7c3c92c4ecef7f4025fb067e1f45824cbf1..0000000000000000000000000000000000000000
GIT binary patch
literal 0
HcmV?d00001

literal 797
zcmb`GzfQwI490zqb7JTaAuhU;ZphM8hqh`FWn$^hP$41asEyPP(2(+H<pDbK7(4(E
z!obL*U^6t(QdlD8bn$2T^G|j%8D9(ynZzXLe8C}r0s!P=z-XhpUrB!PmPEBA`4l?e
zA{JDH;aPcuZj^v_78f%EDWz-0PdEt&f0a9F0tTgJ!WZ}cK@_n<qV5ohYDw}!R7Wfb
z(QCAi=pDlgGE-f$W$+(U%q=vZT_#a2NnR}Lhy`K#i1#r)#o>*ZsV><v_>IYRM_~18
zIxntsnIN`7hJT5Ww%nQXO>RZDNvCUd1i;3P4tHaXV%_MBHr524tjw$1`{J%QEz4eB
eRm;chp{Q1i<xMs`?hpC{=RNP~iofv*{}`Wf97+oS