[skip ci] Add unified partition reader design RFC

[liurenjie1024]

Fixes: N/A (design RFC; no tracking issue).

Description

This PR adds an RFC for a unified partition-reader framework in cudf-spark.

The RFC:

• describes the maintenance and extensibility limitations of the current per-file,
  multi-threaded, and coalescing readers;
• proposes UnifiedReader as the format-neutral execution and resource-lifecycle coordinator;
• defines Scheduler, Combiner, and Decoder extension points;
• documents composable raw Parquet/ORC scheduling and dedicated Iceberg/Delta scheduler
  hierarchies; and
• provides an Iceberg Parquet example showing how the components are assembled.

This is a design-only change and does not alter runtime behavior.

AI assistance: This RFC was drafted and revised with Codex.

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
  (Please provide the names of the existing tests in the PR description.)
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Greptile Summary

The PR adds a design RFC for a format-neutral unified partition-reader framework.

• Defines UnifiedReader as the execution and resource-lifecycle coordinator.
• Introduces scheduler, combiner, and decoder extension points.
• Describes raw-file, Iceberg, and Delta scheduler hierarchies and an Iceberg Parquet assembly example.

Confidence Score: 4/5

The documentation-only PR is safe to merge after correcting the non-blocking inconsistency in the proposed next() cleanup path.

The RFC is otherwise internally coherent, but an implementation following its pseudocode would not release reader-owned resources when the decoded iterator throws from next().

Files Needing Attention: docs/rfc/rapids_table_scan_v2/design.md

Important Files Changed

Filename	Overview
docs/rfc/rapids_table_scan_v2/design.md	Adds the unified reader RFC, but its next() error path bypasses the cleanup behavior promised by the design.

_{Reviews (1): Last reviewed commit: "docs: add unified partition reader desig..." | Re-trigger Greptile}

[greptile-apps]

Next bypasses failure cleanup

When lazy decoding or post-processing throws from currentBatches.next(), the exception bypasses the cleanup guard in hasNext(), leaving the decoded iterator, combined input, and scheduler open despite the RFC's stated failure-handling contract.

Suggested change

	next():
	if not hasNext():
	throw NoSuchElementException
	return currentBatches.next()
	next():
	try:
	if not hasNext():
	throw NoSuchElementException
	return currentBatches.next()
	catch error:
	close()
	throw error

[revans2]

@liurenjie1024 I started to read through the document and generally it looks good, but I think we need to coordinate with @pmattione-nvidia as he is in the middle of rewriting the parquet reader to use the new hybrid reader APIs in cudf, and that can change a lot of how and when I/O can happen. That may not be something we have to think/worry about in the short term, but long term it is very much something we need to think about.

We also want to very much think about read ahead and how it would fit into a design like this. There are two cases here. The first is when we have a single task with a very large amount of input. How do we make sure that we can keep that task fed with data so that it does not have to wait for data once we start it. The second is about lots of tasks. How do we make sure that we are reading data for other tasks so that when the first task finishes we are ready with the data for the next task to start. The real issue here is around memory management (mostly host memory but in the case of the hybrid reader it may also include some device memory)

Finally I want to make sure that we concentrate on the right things. What are actually the requirements. local mode is not that important from a performance standpoint. That generally corresponds to per-file, but it may also be used in some corner cases still when we cannot support combining yet. Also because fetching data, especially from blob stores like S3 can have very high latency, even if the throughput is good. A lot of this is why we allowed for starting processing when a small amount of data is available. The requirement is not to maintain this. The requirement is to not lose performance in those cases. If we do scheduling properly and understand the latency vs throughput we may be able to adjust things like read-ahead amount to avoid these problems entirely.

[liurenjie1024]

Hi, @revans2 The design in this rfc is mainly about framework level interface, and doesn't have much implementation details. I have invited @pmattione-nvidia to discuss with us together. If you want, I can implement a draft pr for iceberg parquet reader for further discussion.

We also want to very much think about read ahead and how it would fit into a design like this. There are two cases here. The first is when we have a single task with a very large amount of input. How do we make sure that we can keep that task fed with data so that it does not have to wait for data once we start it. The second is about lots of tasks. How do we make sure that we are reading data for other tasks so that when the first task finishes we are ready with the data for the next task to start. The real issue here is around memory management (mostly host memory but in the case of the hybrid reader it may also include some device memory)

Actually, that's part of the implementation details of a scheduler. In fact, what's hidden under a scheduler is several parts:

1. The rule to determine what part could be merged. Table formats like iceberg has a different rule set compared with file formats, and that's main motivation of this rfc.
2. The scheduling of io. Our current implementation did some optimization on object store like coalescing adjacent chunks, vector io, etc. From what I learn when tuning nds workload on s3, there are more to explore: split large chunks, merge small chunks, application level control, async file io api by utilizing a standalone io pool, etc. That's what I plan to do after we have finished the refactoring of this rfc.

Your point seems to be more related to how to perform faster io with limited resources, and that's exactly what I plan to do after this refactoring.

Finally I want to make sure that we concentrate on the right things. What are actually the requirements. local mode is not that important from a performance standpoint. That generally corresponds to per-file, but it may also be used in some corner cases still when we cannot support combining yet. Also because fetching data, especially from blob stores like S3 can have very high latency, even if the throughput is good. A lot of this is why we allowed for starting processing when a small amount of data is available. The requirement is not to maintain this. The requirement is to not lose performance in those cases. If we do scheduling properly and understand the latency vs throughput we may be able to adjust things like read-ahead amount to avoid these problems entirely.

Got it. I'll make per file reader as simple as possible. I'll keep the requirements in mind to avoid things got too complicated while keeping performance.

[revans2]

@liurenjie1024 sorry I should have been more clear. Most of my comments were not intended to say your design was wrong in some way. It was just to point out what I see as the requirements. My main concern was making sure that the design could accommodate what @pmattione-nvidia has been working on. In general it looks good. I just don't know what the requirements are that Paul has. Nor do I know what would be nice to have in the new setup. I really just wanted to pull him into the conversation and if he is in it, then I am happy.

[pmattione-nvidia]

The main contention between the hybrid reader and this proposal is that the hybrid reader wants to interleave I/O (with gpu semaphore) with decoding. The hybrid reader wants to: 1) I/O read filter columns (get/release gpu semaphore) -> 2) decode filter columns 3) filter out row groups and build surviving row mask -> 4) I/O read payload columns (get/release gpu semaphore) -> 5) decode surviving payload data

So this proposal may need two separate combiner and decoder stages to accommodate this. A config parameter determines whether the parquet read is done in 2 stages (filter, payload) or not. For this config parameter, the AUTO option we would use some file/footer metadata to evaluate during the scheduler phase, which I believe we'd have at that time.

[liurenjie1024]

The main contention between the hybrid reader and this proposal is that the hybrid reader wants to interleave I/O (with gpu semaphore) with decoding. The hybrid reader wants to: 1) I/O read filter columns (get/release gpu semaphore) -> 2) decode filter columns 3) filter out row groups and build surviving row mask -> 4) I/O read payload columns (get/release gpu semaphore) -> 5) decode surviving payload data

So this proposal may need two separate combiner and decoder stages to accommodate this. A config parameter determines whether the parquet read is done in 2 stages (filter, payload) or not. For this config parameter, the AUTO option we would use some file/footer metadata to evaluate during the scheduler phase, which I believe we'd have at that time.

Hi, @pmattione-nvidia One thing I get your optimization, it's similar to late materialize. One thing I want to confirm is that do we still need to combine as before? If so, will the combination still happen on cpu or gpu?

[pmattione-nvidia]

One thing I get your optimization, it's similar to late materialize. One thing I want to confirm is that do we still need to combine as before? If so, will the combination still happen on cpu or gpu?

@liurenjie1024 For file coalescing mode, yes we will still combine files together into a large virtual file, and yes it will still be done on the CPU/host. However, in 2-stage mode we will need to be able to combine twice: first to combine only the filter columns (from the first I/O read), then later to combine payload column data from the surviving row groups (from the 2nd I/O read).

Note that in two-stage mode we may still decide to do all of the I/O up-front (e.g. if we think significant payload pruning is unlikely, or the dataset is small anyway). In this case we'd just want to do a single combine (coalescing) step.

Eventually the plan is to transition to the hybrid reader's multi-file support, and no coalescing will be necessary at all, because the reader will actually be able to handle multiple files. this is longer term though.