fix: read remote parquet removal directories

[nightcityblade]

Description

Closes #1217.

Normalize remote removal-dataset URLs to the path expected by their filesystem before passing them to PyArrow. This keeps directory discovery on GCS and other fsspec backends within the same base path, while preserving caller-provided filesystems and storage options. An in-memory remote-filesystem regression test covers the directory-read path.

Usage

TextDuplicatesRemovalStage(
    ids_to_remove_path="gs://bucket/removal-ids",
    read_kwargs={"storage_options": {"token": "cloud"}},
)

Testing

• tests/stages/text/deduplication/test_removal.py — 6 passed.
• ruff check nemo_curator/stages/text/deduplication/removal.py tests/stages/text/deduplication/test_removal.py — passed.
• ruff format --check nemo_curator/stages/text/deduplication/removal.py tests/stages/text/deduplication/test_removal.py — passed.

Checklist

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

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[greptile-apps]

Greptile Summary

This PR fixes reading of remote (GCS, S3, etc.) Parquet directories in TextDuplicatesRemovalStage by resolving the filesystem and stripping the protocol prefix in __post_init__ rather than on every process() call. The resolved path and filesystem are stored as instance attributes and reused across batches.

• removal.py: Protocol detection is moved to __post_init__; url_to_fs builds the filesystem once and stores it in read_kwargs, while the stripped path is stored in _read_path for all subsequent process() calls.
• test_removal.py: A new integration test using fsspec's in-memory filesystem covers both the auto-detected filesystem path and the caller-supplied filesystem path, verifying that the stripped path and filesystem object are passed consistently to pd.read_parquet.

Confidence Score: 5/5

Safe to merge; the logic change is well-scoped, moves filesystem resolution to init rather than each batch call, and the new test exercises both code branches end-to-end.

The fix is confined to a single stage's init method with no side effects on other components. The test isolation gap (MemoryFileSystem not cleaned up) is a hygiene concern that does not affect correctness.

tests/stages/text/deduplication/test_removal.py — the new test writes to the global MemoryFileSystem store without cleanup.

Important Files Changed

Filename	Overview
nemo_curator/stages/text/deduplication/removal.py	Moves filesystem resolution from process() to post_init; correctly detects protocol, resolves or accepts a filesystem, strips the protocol prefix, and stores both for reuse across batches.
tests/stages/text/deduplication/test_removal.py	New integration test covers both the auto-detected and caller-supplied filesystem branches, but writes to the global MemoryFileSystem store without cleanup, which can pollute other tests using the same memory:// paths.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A["TextDuplicatesRemovalStage.__post_init__()"] --> B{"protocol in\n(None, 'file')?"}
    B -- Yes --> C["_read_path = ids_to_remove_path\n(no filesystem resolution)"]
    B -- No --> D{"'filesystem' in\nread_kwargs?"}
    D -- No --> E["url_to_fs(ids_to_remove_path,\n**storage_options)"]
    E --> F["read_kwargs['filesystem'] = fs\n_read_path = stripped path"]
    D -- Yes --> G["pop storage_options (ignored)"]
    G --> H{"fs._strip_protocol\nexists?"}
    H -- Yes --> I["_read_path = fs._strip_protocol(url)"]
    H -- No --> J["_read_path = split_protocol(url)[1]\n⚠ may include // prefix"]
    C --> K["process(): pd.read_parquet(_read_path, **read_kwargs)"]
    F --> K
    I --> K
    J --> K

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
    A["TextDuplicatesRemovalStage.__post_init__()"] --> B{"protocol in\n(None, 'file')?"}
    B -- Yes --> C["_read_path = ids_to_remove_path\n(no filesystem resolution)"]
    B -- No --> D{"'filesystem' in\nread_kwargs?"}
    D -- No --> E["url_to_fs(ids_to_remove_path,\n**storage_options)"]
    E --> F["read_kwargs['filesystem'] = fs\n_read_path = stripped path"]
    D -- Yes --> G["pop storage_options (ignored)"]
    G --> H{"fs._strip_protocol\nexists?"}
    H -- Yes --> I["_read_path = fs._strip_protocol(url)"]
    H -- No --> J["_read_path = split_protocol(url)[1]\n⚠ may include // prefix"]
    C --> K["process(): pd.read_parquet(_read_path, **read_kwargs)"]
    F --> K
    I --> K
    J --> K

Loading

_{Reviews (2): Last reviewed commit: "fix: reuse remote removal filesystem" | Re-trigger Greptile}