feat(scan): the parquet ingestible honors a per-file byte range

src/include/op/scan/parquet_gpu_ingestible.hpp

@@ -35,13 +35,15 @@
 #include <cudf/io/parquet.hpp>
 
 // standard library
+#include <algorithm>
 #include <atomic>
 #include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>
 #include <span>
 #include <string>
+#include <utility>
 #include <vector>
 
 namespace sirius::scan_manager {
@@ -67,6 +69,12 @@ class parquet_ingestible_table_info : public ingestible_table_info {
  public:
   duckdb::vector<sirius::logical_type> returned_types;
   std::vector<std::string> resolved_file_paths;
+  /// Per-file byte range `[start, start+length)`, parallel to @ref resolved_file_paths.
+  /// Empty means every file is read whole; a `(0,0)` entry means that file is read whole.
+  /// A ranged file reads only the row groups whose start offset falls inside the range
+  /// (see parquet_byte_range.hpp) — the mechanism behind distributed byte-range splits,
+  /// where N ranges of one file must read every row exactly once across scan instances.
+  std::vector<std::pair<std::uint64_t, std::uint64_t>> resolved_file_ranges;
   duckdb::vector<duckdb::ColumnIndex> column_ids;
   duckdb::vector<duckdb::idx_t> projection_ids;
   duckdb::vector<std::string> names;
@@ -85,6 +93,14 @@ class parquet_ingestible_table_info : public ingestible_table_info {
 
   parquet_ingestible_table_info() = default;
 
+  /// True when any file carries a real byte range (a `(0,0)` entry is a whole file).
+  [[nodiscard]] bool has_byte_ranges() const
+  {
+    return std::any_of(resolved_file_ranges.begin(),
+                       resolved_file_ranges.end(),
+                       [](auto const& range) { return range.second != 0 || range.first != 0; });
+  }
+
   [[nodiscard]] std::span<std::string const> column_names() const override { return names; }
 
   [[nodiscard]] std::span<std::string const> file_paths() const override
@@ -329,12 +345,15 @@ class parquet_gpu_ingestible : public gpu_ingestible {
   }
 
  private:
-  /// Read one file's footer, prune its row groups against the filter, and record
-  /// per-row-group byte accounting. Returns a single @c parquet_file_scan_info.
-  /// Runs on a scan-manager dispatcher thread (the task returned by
-  /// @ref next_split_provider).
-  std::unique_ptr<scan_info> build_file_scan_info(std::string const& file_path,
-                                                  std::shared_ptr<io::sirius_ioctx> const& io_ctx);
+  /// Read one file's footer, prune its row groups against the byte range and the filter, and
+  /// record per-row-group byte accounting. Returns a single @c parquet_file_scan_info.
+  /// `byte_range` is `(0,0)` for a whole-file read; otherwise only row groups starting inside
+  /// `[start, start+length)` are read. Runs on a scan-manager dispatcher thread (the task
+  /// returned by @ref next_split_provider).
+  std::unique_ptr<scan_info> build_file_scan_info(
+    std::string const& file_path,
+    std::pair<std::uint64_t, std::uint64_t> byte_range,
+    std::shared_ptr<io::sirius_ioctx> const& io_ctx);
 
   std::unique_ptr<parquet_ingestible_table_info> _info;
 

src/include/scan_manager/sirius_scan_manager.hpp

@@ -110,7 +110,11 @@ namespace sirius::scan_manager {
 /// owns the match logic that @ref sirius_scan_manager::try_match_cached_entry consults.
 class cache_entry_info {
  public:
-  std::vector<std::string> resolved_file_paths;    ///< parquet identity (file set)
+  std::vector<std::string> resolved_file_paths;  ///< parquet identity (file set)
+  /// True when the parquet scan behind this entry read byte-range subsets of its files. A
+  /// ranged scan holds a fraction of each file's rows, so it neither serves nor is served by
+  /// the cache — matching on the file set alone would silently return extra or missing rows.
+  bool has_byte_ranges = false;
   std::string catalog_name;                        ///< duckdb identity: catalog (attach alias)
   std::string schema_name;                         ///< duckdb identity: schema
   std::string table_name;                          ///< duckdb identity: table

src/op/scan/parquet_gpu_ingestible.cpp

@@ -26,6 +26,7 @@
 #include <log/logging.hpp>
 #include <op/dynamic_filter/sirius_dynamic_filter.hpp>
 #include <op/scan/dynamic_filter_merge.hpp>
+#include <op/scan/parquet_byte_range.hpp>
 #include <op/scan/parquet_gpu_ingestible.hpp>
 #include <op/scan/parquet_metadata.hpp>
 #include <op/scan/parquet_schema_mapping.hpp>
@@ -609,6 +610,14 @@ parquet_gpu_ingestible::parquet_gpu_ingestible(std::unique_ptr<parquet_ingestibl
   }
 
   _file_paths = bind.resolved_file_paths;
+  if (!bind.resolved_file_ranges.empty() &&
+      bind.resolved_file_ranges.size() != bind.resolved_file_paths.size()) {
+    throw sirius::invalid_input_exception(
+      "parquet scan carries {} byte ranges for {} files; a partial pairing would make "
+      "row-group ownership ambiguous",
+      bind.resolved_file_ranges.size(),
+      bind.resolved_file_paths.size());
+  }
 }
 
 parquet_gpu_ingestible::~parquet_gpu_ingestible() = default;
@@ -642,18 +651,23 @@ std::function<std::unique_ptr<op::scan::scan_info>()> parquet_gpu_ingestible::ne
   // per file; row-group chunking and file bundling happen downstream in
   // parquet_batch_coalescer.
   auto const& file_path = _file_paths[idx];
+  auto const byte_range = idx < _info->resolved_file_ranges.size()
+                            ? _info->resolved_file_ranges[idx]
+                            : std::pair<std::uint64_t, std::uint64_t>{0, 0};
   // The resolver returns a valid ioctx or throws if no backend supports the path.
   auto io_ctx = resolve(file_path);
-  return [this, file_path, io_ctx = std::move(io_ctx)]() -> std::unique_ptr<scan_info> {
-    return build_file_scan_info(file_path, io_ctx);
+  return [this, file_path, byte_range, io_ctx = std::move(io_ctx)]() -> std::unique_ptr<scan_info> {
+    return build_file_scan_info(file_path, byte_range, io_ctx);
   };
 }
 
 //===----------------------------------------------------------------------===//
 // build_file_scan_info — per-file footer read + row-group pruning
 //===----------------------------------------------------------------------===//
 std::unique_ptr<scan_info> parquet_gpu_ingestible::build_file_scan_info(
-  std::string const& file_path, std::shared_ptr<io::sirius_ioctx> const& io_ctx)
+  std::string const& file_path,
+  std::pair<std::uint64_t, std::uint64_t> byte_range,
+  std::shared_ptr<io::sirius_ioctx> const& io_ctx)
 {
   auto stream = cudf::get_default_stream();
 
@@ -799,6 +813,18 @@ std::unique_ptr<scan_info> parquet_gpu_ingestible::build_file_scan_info(
   }
 
   auto row_group_indices = reader.all_row_groups(opts);
+  // Distributed byte-range split: keep only the row groups this range owns (start-offset
+  // containment, parquet_byte_range.hpp), BEFORE stats pruning. An empty selection is a valid
+  // empty split and flows through the all-pruned fallback below — never a whole-file read.
+  if (byte_range.second != 0 || byte_range.first != 0) {
+    row_group_indices =
+      detail::row_groups_in_byte_range(metadata, byte_range.first, byte_range.second);
+    SIRIUS_LOG_DEBUG("[parquet_gpu_ingestible] Byte range [{}, +{}) of {} owns {} row group(s)",
+                     byte_range.first,
+                     byte_range.second,
+                     file_path,
+                     row_group_indices.size());
+  }
   if (ast_expression && !disable_filter_pushdown) {
     auto const rgs_before = row_group_indices.size();
     row_group_indices     = reader.filter_row_groups_with_stats(row_group_indices, opts, stream);

src/scan_manager/sirius_scan_manager.cpp

@@ -1894,8 +1894,9 @@ cache_entry_info cache_entry_info::from(const op::scan::ingestible_table_info& i
     // resolved) path.
     ci.resolved_file_paths = p->resolved_file_paths;
     op::scan::canonicalize_scan_file_paths(ci.resolved_file_paths);
-    ci.column_ids = p->column_ids;
-    ci.names      = aligned_column_names(p->names, p->column_ids);
+    ci.has_byte_ranges = p->has_byte_ranges();
+    ci.column_ids      = p->column_ids;
+    ci.names           = aligned_column_names(p->names, p->column_ids);
   } else if (auto const* d =
                dynamic_cast<op::scan::duckdb_native_ingestible_table_info const*>(&info)) {
     ci.catalog_name = d->catalog_name;
@@ -1915,6 +1916,9 @@ std::vector<std::size_t> cache_entry_info::can_serve_with_columns(
   // never serves a scan of the other — the identity check below falls through (a
   // duckdb cache has empty resolved_file_paths; a parquet cache has an empty table_name).
   if (auto const* p = dynamic_cast<op::scan::parquet_ingestible_table_info const*>(&other)) {
+    // A byte-range split scan reads a subset of each file's row groups, so it can neither be
+    // served by a whole-file pin (extra rows) nor produce a pin that serves anyone else.
+    if (has_byte_ranges || p->has_byte_ranges()) { return {}; }
     if (!matches_parquet_files(p->resolved_file_paths)) { return {}; }
     return column_projection_for(p->column_ids);
   }

test/cpp/scan/test_can_serve_with_columns.cpp

@@ -147,6 +147,42 @@ TEST_CASE("cache_entry_info: parquet different file set misses", "[scan][can_ser
   REQUIRE(pinned.can_serve_with_columns(two_files).empty());
 }
 
+TEST_CASE("cache_entry_info: a byte-range split scan neither serves nor is served by a pin",
+          "[scan][can_serve]")
+{
+  // A ranged scan holds a fraction of each file's rows, so the file set alone is not
+  // its identity: a whole-file pin would hand it extra rows, and a pin built from it
+  // would hand a whole-file scan missing rows. Both directions miss even though the
+  // file set and the columns match.
+  auto whole_file_pin = parquet_cache({"a.parquet"}, {0, 1});
+  parquet_ingestible_table_info ranged_scan;
+  fill(ranged_scan, {"a.parquet"}, {0});
+  ranged_scan.resolved_file_ranges = {{0, 4096}};
+  REQUIRE(ranged_scan.has_byte_ranges());
+  REQUIRE(whole_file_pin.can_serve_with_columns(ranged_scan).empty());
+
+  // The pin side records the fact at construction (cache_entry_info::from): a pin
+  // built from the ranged scan never serves a whole-file scan, while the same pin
+  // built from a whole-file scan does — so the miss is the range, not the file set.
+  ranged_scan.names = {"a"};
+  auto ranged_pin   = cache_entry_info::from(ranged_scan);
+  REQUIRE(ranged_pin.has_byte_ranges);
+  parquet_ingestible_table_info whole_file_scan;
+  fill(whole_file_scan, {"a.parquet"}, {0});
+  whole_file_scan.names = {"a"};
+  REQUIRE(ranged_pin.can_serve_with_columns(whole_file_scan).empty());
+  auto control_pin = cache_entry_info::from(whole_file_scan);
+  REQUIRE_FALSE(control_pin.has_byte_ranges);
+  REQUIRE(control_pin.can_serve_with_columns(whole_file_scan) == std::vector<std::size_t>{0});
+
+  // A (0,0) entry is a whole-file read, not a range: it still hits.
+  parquet_ingestible_table_info unranged_scan;
+  fill(unranged_scan, {"a.parquet"}, {0});
+  unranged_scan.resolved_file_ranges = {{0, 0}};
+  REQUIRE_FALSE(unranged_scan.has_byte_ranges());
+  REQUIRE(whole_file_pin.can_serve_with_columns(unranged_scan) == std::vector<std::size_t>{0});
+}
+
 TEST_CASE("cache_entry_info: duckdb same-table subset request hits with a gather projection",
           "[scan][can_serve]")
 {

test/cpp/scan/test_parquet_byte_range.cpp

@@ -18,15 +18,23 @@
 #include <catch.hpp>
 
 // sirius
+#include <io/kvikio/kvikio_context.hpp>
 #include <op/scan/parquet_byte_range.hpp>
+#include <op/scan/parquet_gpu_ingestible.hpp>
 #include <sirius/exception.hpp>
 
 // cudf
+#include <cudf/column/column_factories.hpp>
 #include <cudf/io/datasource.hpp>
 #include <cudf/io/experimental/hybrid_scan.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
 #include <cudf/io/parquet_schema.hpp>
+#include <cudf/table/table_view.hpp>
+#include <cudf/utilities/default_stream.hpp>
+
+// rmm
+#include <rmm/device_buffer.hpp>
 
 // standard library
 #include <algorithm>
@@ -227,3 +235,136 @@ TEST_CASE("real footer: rule agrees with cudf's byte-range filter on the test li
          << cudf_left.size() << " ours=" << left.size());
   }
 }
+
+namespace {
+
+/// Writes a one-column (a BIGINT, values 0..rows-1) parquet with `rows / rows_per_group` row
+/// groups, and returns its path. Deterministic row-group layout for the split-scan tests.
+fs::path write_multi_row_group_parquet(fs::path const& dir,
+                                       std::int64_t rows,
+                                       std::int64_t rows_per_group)
+{
+  auto const path = dir / "byte_range_scan.parquet";
+  auto stream     = cudf::get_default_stream();
+  std::vector<std::int64_t> host(rows);
+  std::iota(host.begin(), host.end(), 0);
+  rmm::device_buffer data(host.data(), rows * sizeof(std::int64_t), stream);
+  auto column = std::make_unique<cudf::column>(cudf::data_type{cudf::type_id::INT64},
+                                               static_cast<cudf::size_type>(rows),
+                                               std::move(data),
+                                               rmm::device_buffer{},
+                                               0);
+  cudf::table_view table({column->view()});
+  cudf::io::table_input_metadata metadata(table);
+  metadata.column_metadata[0].set_name("a");
+  auto options =
+    cudf::io::parquet_writer_options::builder(cudf::io::sink_info(path.string()), table)
+      .metadata(std::move(metadata))
+      .row_group_size_rows(rows_per_group)
+      .build();
+  cudf::io::write_parquet(options);
+  return path;
+}
+
+/// Table info for a byte-range scan of `path` projecting the single BIGINT column.
+std::unique_ptr<sirius::op::scan::parquet_ingestible_table_info> ranged_info(fs::path const& path,
+                                                                             std::uint64_t start,
+                                                                             std::uint64_t length)
+{
+  auto info                  = std::make_unique<sirius::op::scan::parquet_ingestible_table_info>();
+  info->resolved_file_paths  = {path.string()};
+  info->resolved_file_ranges = {{start, length}};
+  info->names                = {"a"};
+  info->returned_types.push_back(sirius::logical_type::make(sirius::type_id::BIGINT));
+  info->column_ids.push_back(duckdb::ColumnIndex(0));
+  info->scan_output_arity      = 1;
+  info->approximate_batch_size = std::size_t{1} << 30;
+  return info;
+}
+
+/// Drives the ingestible for one range and returns the selected row-group indices plus the
+/// number of rows they hold (from the footer metadata — no decode needed).
+std::pair<std::vector<cudf::size_type>, std::int64_t> scan_selection(
+  std::unique_ptr<sirius::op::scan::parquet_ingestible_table_info> info)
+{
+  auto ingestible = sirius::op::scan::make_ingestible(std::move(info));
+  auto ioctx      = std::make_shared<sirius::io::kvikio_context>();
+  auto task       = ingestible->next_split_provider(
+    [ioctx](std::string_view) -> std::shared_ptr<sirius::io::sirius_ioctx> { return ioctx; });
+  REQUIRE(task);
+  auto file = task();
+  REQUIRE(file);
+
+  // Splits materialize downstream of the coalescer, exactly as in production.
+  auto coalescer = ingestible->create_batch_coalescer();
+  auto batches   = coalescer->push(std::move(file));
+  for (auto& batch : coalescer->flush()) {
+    batches.push_back(std::move(batch));
+  }
+
+  std::vector<cudf::size_type> indices;
+  std::int64_t rows = 0;
+  for (auto const& batch : batches) {
+    auto* split = dynamic_cast<sirius::op::scan::parquet_split_info*>(batch.get());
+    REQUIRE(split);
+    for (auto const& slice : split->rg_slices) {
+      for (auto const idx : slice.row_group_indices) {
+        indices.push_back(idx);
+        rows += slice.file_metadata->row_groups.at(idx).num_rows;
+      }
+    }
+  }
+  std::sort(indices.begin(), indices.end());
+  return {indices, rows};
+}
+
+}  // namespace
+
+TEST_CASE("a two-way split scan selects disjoint, complete row groups",
+          "[parquet_byte_range][scan]")
+{
+  auto const dir = fs::temp_directory_path() / "sirius_byte_range_test";
+  fs::create_directories(dir);
+  // cudf clamps row_group_size_rows to a 5000-row floor; 50k rows -> 10 real row groups of
+  // sequential int64s, large enough that both halves of the file hold data pages.
+  constexpr std::int64_t kRows = 50000;
+  auto const path              = write_multi_row_group_parquet(dir, kRows, 5000);
+  auto const file_size         = std::uint64_t(fs::file_size(path));
+  auto const half              = file_size / 2;
+
+  auto [left_rgs, left_rows]   = scan_selection(ranged_info(path, 0, half));
+  auto [right_rgs, right_rows] = scan_selection(ranged_info(path, half, file_size - half));
+
+  INFO("left row groups: " << left_rgs.size() << ", right: " << right_rgs.size());
+  REQUIRE(!left_rgs.empty());
+  REQUIRE(!right_rgs.empty());
+  for (auto const idx : left_rgs) {
+    REQUIRE(std::find(right_rgs.begin(), right_rgs.end(), idx) == right_rgs.end());
+  }
+  REQUIRE(left_rows + right_rows == kRows);
+
+  // A whole-file scan ((0,0) range) is byte-identical to no range at all.
+  auto [all_rgs, all_rows] = scan_selection(ranged_info(path, 0, 0));
+  REQUIRE(all_rows == kRows);
+  REQUIRE(std::int64_t(left_rgs.size() + right_rgs.size()) == std::int64_t(all_rgs.size()));
+
+  // A range inside one row group is a valid empty scan.
+  auto [none_rgs, none_rows] = scan_selection(ranged_info(path, 10, 5));
+  REQUIRE(none_rgs.empty());
+  REQUIRE(none_rows == 0);
+
+  fs::remove_all(dir);
+}
+
+TEST_CASE("mismatched range/path pairing is refused at construction", "[parquet_byte_range][scan]")
+{
+  auto info                  = std::make_unique<sirius::op::scan::parquet_ingestible_table_info>();
+  info->resolved_file_paths  = {"a.parquet", "b.parquet"};
+  info->resolved_file_ranges = {{0, 10}};
+  info->names                = {"a"};
+  info->returned_types.push_back(sirius::logical_type::make(sirius::type_id::BIGINT));
+  info->column_ids.push_back(duckdb::ColumnIndex(0));
+  info->scan_output_arity = 1;
+  REQUIRE_THROWS_AS(sirius::op::scan::make_ingestible(std::move(info)),
+                    sirius::invalid_input_exception);
+}