Supply stream and mr to column synthesizers

[mhaseeb123]

Description

Follow up #23077

Supply stream and mr to column synthesizers which are used to produce the output columns

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.

[coderabbitai]

📝 Walkthrough

Summary by CodeRabbit

• Bug Fixes
  • Improved Parquet reads that include optional row index or source index columns.
  • These generated columns are now built using the active execution context, which helps ensure more reliable output and consistent behavior across read operations.
  • Fixed handling for both single-source and multi-source cases when adding these metadata-style columns to results.

Walkthrough

Parquet output assembly now forwards explicit CUDA stream and memory resource parameters to row-index and source-index synthesis helpers. The helpers use those parameters for device allocations, asynchronous copies, labeling, transforms, synchronization, and returned columns.

Changes

Parquet index column context

Layer / File(s)	Summary
Propagate stream and memory resource cpp/src/io/parquet/reader_impl.hpp, cpp/src/io/parquet/reader_impl.cpp	Synthesis declarations and finalize_output call sites now pass explicit CUDA stream and memory resource parameters.
Allocate and execute with supplied context cpp/src/io/parquet/reader_impl_preprocess.cu	Row-index and source-index synthesis use the supplied context for allocations, transforms, labeling, synchronization, and returned columns.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Possibly related PRs

• rapidsai/cudf#22879: Updates Parquet source-index construction to use explicit stream and memory-resource context.

Suggested labels: cuIO, improvement, non-breaking, 4 - Needs Review

Suggested reviewers: qbacpey, davidwendt, bdice

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly matches the main change: passing stream and memory resource details to column synthesizers.
Description check	✅ Passed	The description is directly related to the change and accurately summarizes the updated column synthesizers.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

cpp/src/io/parquet/reader_impl_preprocess.cu (1)

1226-1240: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Stream mismatch: host_row_offsets still allocated on _stream instead of the new stream parameter.

synthesize_source_index_column now takes an explicit stream/mr, and every other allocation/op in this block correctly uses stream (lines 1235-1239), but host_row_offsets on line 1230 is still allocated against the member _stream. The subsequent async H2D copy (make_device_uvector_async(host_row_offsets, stream, ...)) and label_segments run on stream, while the pinned staging buffer's lifetime/deallocation is tracked against _stream. If a future caller passes a stream different from _stream, the pinned buffer could be reclaimed/reused once _stream completes, before the copy on stream finishes — a stream-ordering race on the staging buffer. Today it happens to be safe only because the sole caller (finalize_output) passes _stream for both parameters, but that coincidence defeats the purpose of this refactor.

🐛 Proposed fix

     auto host_row_offsets =
-      cudf::detail::make_empty_pinned_vector<cudf::size_type>(num_sources + 1, _stream);
+      cudf::detail::make_empty_pinned_vector<cudf::size_type>(num_sources + 1, stream);

As per coding guidelines, "Propagate stream and memory-resource parameters through internal APIs consistently" and "Ensure async memcpy staging buffers outlive the copy they are staging."

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@cpp/src/io/parquet/reader_impl_preprocess.cu` around lines 1226 - 1240, In
synthesize_source_index_column, the pinned staging buffer allocation for
host_row_offsets is still tied to _stream while the rest of the work uses the
explicit stream parameter. Update that allocation to use the same stream passed
into the function so the pinned buffer lifetime and async H2D copy in
make_device_uvector_async remain ordered on the correct stream. Keep the
existing label_segments and synchronization flow, but make the stream usage
consistent throughout this block.

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@cpp/src/io/parquet/reader_impl_preprocess.cu`:
- Around line 1226-1240: In synthesize_source_index_column, the pinned staging
buffer allocation for host_row_offsets is still tied to _stream while the rest
of the work uses the explicit stream parameter. Update that allocation to use
the same stream passed into the function so the pinned buffer lifetime and async
H2D copy in make_device_uvector_async remain ordered on the correct stream. Keep
the existing label_segments and synchronization flow, but make the stream usage
consistent throughout this block.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Enterprise

Run ID: 34aa6af4-35af-4b03-b9de-6fd7bde95b4a

📥 Commits

Reviewing files that changed from the base of the PR and between 713e7dd and c2ce174.

📒 Files selected for processing (3)

• cpp/src/io/parquet/reader_impl.cpp
• cpp/src/io/parquet/reader_impl.hpp
• cpp/src/io/parquet/reader_impl_preprocess.cu

[mhaseeb123]

/merge