Add ability to set the source_info of parquet_reader_options (rapidsai#20253)

This is useful for multi-task chunked reading since the user can pass a "template" set of options in and the tasking framework can reconstruct the options with individual source locations without laboriously having to transfer all of the options over (and remember to update when new options appear).

Authors:
  - Lawrence Mitchell (https://github.com/wence-)

Approvers:
  - Muhammad Haseeb (https://github.com/mhaseeb123)
  - Matthew Roeschke (https://github.com/mroeschke)
  - Paul Mattione (https://github.com/pmattione-nvidia)

URL: rapidsai#20253

Author: wence-

cpp/include/cudf/io/avro.hpp

@@ -95,6 +95,13 @@ class avro_reader_options {
    */
   [[nodiscard]] size_type get_num_rows() const { return _num_rows; }
 
+  /**
+   * @brief Sets source info.
+   *
+   * @param src The source info.
+   */
+  void set_source(source_info src) { _source = std::move(src); }
+
   /**
    * @brief Set names of the column to be read.
    *

cpp/include/cudf/io/csv.hpp

@@ -486,6 +486,13 @@ class csv_reader_options {
    */
   [[nodiscard]] data_type get_timestamp_type() const { return _timestamp_type; }
 
+  /**
+   * @brief Sets source info.
+   *
+   * @param src The source info.
+   */
+  void set_source(source_info src) { _source = std::move(src); }
+
   /**
    * @brief Sets compression format of the source.
    *

cpp/include/cudf/io/json.hpp

@@ -379,6 +379,13 @@ class json_reader_options {
    */
   [[nodiscard]] std::vector<std::string> const& get_na_values() const { return _na_values; }
 
+  /**
+   * @brief Sets source info.
+   *
+   * @param src The source info.
+   */
+  void set_source(source_info src) { _source = std::move(src); }
+
   /**
    * @brief Set data types for columns to be read.
    *

cpp/include/cudf/io/orc.hpp

@@ -203,6 +203,13 @@ class orc_reader_options {
 
   // Setters
 
+  /**
+   * @brief Sets source info.
+   *
+   * @param src The source info.
+   */
+  void set_source(source_info src) { _source = std::move(src); }
+
   /**
    * @brief Sets names of the column to read.
    *

cpp/include/cudf/io/parquet.hpp

@@ -257,6 +257,13 @@ class parquet_reader_options {
    */
   [[nodiscard]] bool is_enabled_use_jit_filter() const { return _use_jit_filter; }
 
+  /**
+   * @brief Set a new source location
+   *
+   * @param src New `source_info`.
+   */
+  void set_source(source_info src) { _source = std::move(src); }
+
   /**
    * @brief Sets the names of columns to be read from all input sources.
    *

cpp/tests/io/parquet_reader_test.cpp

@@ -3681,3 +3681,25 @@ TEST_F(ParquetReaderTest, ByteBoundsAndFilters)
     CUDF_TEST_EXPECT_TABLES_EQUAL(read->view(), expected->view());
   }
 }
+
+TEST_F(ParquetReaderTest, LateBindSourceInfo)
+{
+  srand(31337);
+  auto expected = create_random_fixed_table<int>(4, 4, false);
+
+  auto filepath = temp_env->get_temp_filepath("LateBindSourceInfo.parquet");
+  cudf::io::parquet_writer_options args =
+    cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, *expected);
+  cudf::io::write_parquet(args);
+
+  cudf::io::parquet_reader_options read_opts =
+    cudf::io::parquet_reader_options::builder(cudf::io::source_info{});
+
+  EXPECT_THROW(cudf::io::read_parquet(read_opts), cudf::logic_error);
+
+  read_opts.set_source(cudf::io::source_info{filepath});
+
+  auto result = cudf::io::read_parquet(read_opts);
+
+  CUDF_TEST_EXPECT_TABLES_EQUAL(result.tbl->view(), expected->view());
+}

python/pylibcudf/pylibcudf/io/avro.pxd

@@ -16,6 +16,7 @@ cdef class AvroReaderOptions:
     cdef avro_reader_options c_obj
     cdef SourceInfo source
     cpdef void set_columns(self, list col_names)
+    cpdef void set_source(self, SourceInfo src)
 
 
 cdef class AvroReaderOptionsBuilder:

python/pylibcudf/pylibcudf/io/avro.pyi

@@ -7,6 +7,8 @@ from pylibcudf.io.types import SourceInfo, TableWithMetadata
 __all__ = ["AvroReaderOptions", "AvroReaderOptionsBuilder", "read_avro"]
 
 class AvroReaderOptions:
+    def set_columns(self, col_names: list[str]) -> None: ...
+    def set_source(self, src: SourceInfo) -> None: ...
     @staticmethod
     def builder(source: SourceInfo) -> AvroReaderOptionsBuilder: ...
 

python/pylibcudf/pylibcudf/io/avro.pyx

@@ -70,6 +70,21 @@ cdef class AvroReaderOptions:
             vec.push_back(str(name).encode())
         self.c_obj.set_columns(vec)
 
+    cpdef void set_source(self, SourceInfo src):
+        """
+        Set a new source info location.
+
+        Parameters
+        ----------
+        src : SourceInfo
+            New source information, replacing existing information.
+
+        Returns
+        -------
+        None
+        """
+        self.c_obj.set_source(src.c_obj)
+
 
 cdef class AvroReaderOptionsBuilder:
     cpdef AvroReaderOptionsBuilder columns(self, list col_names):

python/pylibcudf/pylibcudf/io/csv.pxd

@@ -45,6 +45,7 @@ cdef class CsvReaderOptions:
     cpdef void set_true_values(self, list true_values)
     cpdef void set_false_values(self, list false_values)
     cpdef void set_na_values(self, list na_values)
+    cpdef void set_source(self, SourceInfo src)
 
 
 cdef class CsvReaderOptionsBuilder: