feat(scan): Support split and parallel scans - Part-2a: read Parquet files by task range

src/iceberg/catalog/rest/json_serde.cc

@@ -18,6 +18,7 @@
  */
 
 #include <iterator>
+#include <limits>
 #include <map>
 #include <memory>
 #include <optional>
@@ -110,6 +111,22 @@ constexpr std::string_view kDeleteFileReferences = "delete-file-references";
 constexpr std::string_view kResidualFilter = "residual-filter";
 constexpr std::string_view kStart = "start";
 constexpr std::string_view kLength = "length";
+constexpr std::string_view kMapKeys = "keys";
+constexpr std::string_view kMapValues = "values";
+
+// Reject values above INT64_MAX; get<int64_t>() would wrap them to negative.
+Result<int64_t> GetRangeValue(const nlohmann::json& json, std::string_view key) {
+  const auto& value = json.at(key);
+  if (!value.is_number_integer()) {
+    return JsonParseError("'{}' must be an integer, but is {}", key, value.type_name());
+  }
+  if (value.is_number_unsigned() &&
+      value.get<uint64_t>() >
+          static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
+    return JsonParseError("'{}' integer is out of range: {}", key, SafeDumpJson(value));
+  }
+  return value.get<int64_t>();
+}
 
 Result<nlohmann::json> StorageCredentialToJson(const StorageCredential& credential) {
   ICEBERG_RETURN_UNEXPECTED(credential.Validate());
@@ -168,14 +185,19 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
     ICEBERG_ASSIGN_OR_RAISE(auto data_file_json,
                             GetJsonValue<nlohmann::json>(task_json, kDataFile));
     ICEBERG_ASSIGN_OR_RAISE(
-        auto data_file,
-        iceberg::DataFileFromJson(data_file_json, partition_spec_by_id, schema));
-    ICEBERG_ASSIGN_OR_RAISE(auto start, GetJsonIntegerOrDefault(task_json, kStart, 0));
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto length,
-        GetJsonIntegerOrDefault(task_json, kLength, data_file.file_size_in_bytes));
-    ICEBERG_RETURN_UNEXPECTED(
-        CheckFileScanTaskNotSplit(start, length, data_file.file_size_in_bytes));
+        auto data_file, DataFileFromJson(data_file_json, partition_spec_by_id, schema));
+    // Neither field means the whole file. Only one field is ambiguous.
+    const bool has_start = task_json.contains(kStart);
+    const bool has_length = task_json.contains(kLength);
+    if (has_start != has_length) {
+      return JsonParseError("'start' and 'length' must be provided together");
+    }
+    int64_t start = 0;
+    int64_t length = data_file.file_size_in_bytes;
+    if (has_start) {
+      ICEBERG_ASSIGN_OR_RAISE(start, GetRangeValue(task_json, kStart));
+      ICEBERG_ASSIGN_OR_RAISE(length, GetRangeValue(task_json, kLength));
+    }
     // FIXME: REST scan-task DataFile JSON currently carries first-row-id,
     // but not the manifest-entry data sequence number. Until the REST API exposes
     // it, REST-planned tasks cannot inherit _last_updated_sequence_number.
@@ -203,9 +225,13 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
       ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json));
     }
 
-    file_scan_tasks.push_back(std::make_shared<FileScanTask>(
-        std::make_shared<DataFile>(std::move(data_file)), std::move(task_delete_files),
-        std::move(residual_filter)));
+    auto task = FileScanTask::MakeSplit(std::make_shared<DataFile>(std::move(data_file)),
+                                        start, length, std::move(task_delete_files),
+                                        std::move(residual_filter));
+    if (!task) {
+      return JsonParseError("Invalid FileScanTask range: {}", task.error().message);
+    }
+    file_scan_tasks.push_back(std::move(task.value()));
   }
   return file_scan_tasks;
 }
@@ -253,10 +279,19 @@ Result<nlohmann::json> ScanTaskFieldsToJson(
       if (!task) continue;
       nlohmann::json task_json;
       if (task->data_file()) {
+        // The shared data file may have changed size; don't write an invalid range.
+        auto range_result =
+            FileScanTask::MakeSplit(task->data_file(), task->start(), task->length());
+        if (!range_result) {
+          return ValidationFailed("Invalid FileScanTask range: {}",
+                                  range_result.error().message);
+        }
         ICEBERG_ASSIGN_OR_RAISE(
             auto data_file_json,
             iceberg::ToJson(*task->data_file(), partition_specs_by_id, schema));
         task_json[kDataFile] = std::move(data_file_json);
+        task_json[kStart] = task->start();
+        task_json[kLength] = task->length();
       }
       if (!task->delete_files().empty()) {
         std::vector<int32_t> refs;

src/iceberg/catalog/rest/types.cc

@@ -192,6 +192,7 @@ bool FileScanTaskEqual(const std::shared_ptr<FileScanTask>& lhs,
     return false;
   }
   return SharedPtrEqual(lhs->data_file(), rhs->data_file()) &&
+         lhs->start() == rhs->start() && lhs->length() == rhs->length() &&
          SharedPtrVectorEqual(lhs->delete_files(), rhs->delete_files()) &&
          ExpressionPtrEqual(lhs->residual_filter(), rhs->residual_filter());
 }

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,10 +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()) {
+      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));
@@ -191,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));
 

src/iceberg/data/file_scan_task_reader.h

@@ -67,6 +67,12 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
   ~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
+  /// 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;

src/iceberg/parquet/parquet_data_util.cc

@@ -22,6 +22,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"
@@ -35,6 +36,31 @@
 
 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();
+  // 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 "
+        "starting page {} (file size {})",
+        start, file_size);
+  }
+  return start;
+}
+
 namespace {
 
 // Forward declaration

src/iceberg/parquet/parquet_data_util_internal.h

@@ -24,8 +24,17 @@
 #include "iceberg/arrow/metadata_column_util_internal.h"
 #include "iceberg/schema_util.h"
 
+namespace parquet {
+class RowGroupMetaData;
+}  // namespace parquet
+
 namespace iceberg::parquet {
 
+/// \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.