Remove unused config dataclasses from download functions

Removes NemotronIngressConfig, DCLMHQDownloadConfig, and TransferConfig.
The underlying functions (download_nemotron_cc, extract_dclm_hq_dump,
transfer_files) now take plain parameters directly. Updates tests and
nemotron.py experiment to use the flat-param API or *_step() functions.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ravwojdyla

experiments/pretraining_datasets/nemotron.py

@@ -8,20 +8,14 @@
 
 from experiments.defaults import DEFAULT_NEW_RUN_DATA_SHUFFLE
 from experiments.pretraining_datasets.dclm import dclm_components_llama3
-from marin.datakit.download.nemotron_cc import NemotronIngressConfig, download_nemotron_cc
+from marin.datakit.download.nemotron_cc import nemotron_cc_step
 from marin.execution.executor import ExecutorStep, output_path_of, this_output_path, versioned
 from marin.processing.tokenize import TokenizeConfig, lm_mixture_data_config, tokenize
 from marin.processing.tokenize.data_configs import TokenizerStep
 
 # Raw dataset download step
 downloads = {
-    "nemotron_cc": ExecutorStep(
-        name="raw/nemotro-cc",
-        fn=download_nemotron_cc,
-        config=NemotronIngressConfig(
-            output_path=this_output_path(),
-        ),
-    )
+    "nemotron_cc": nemotron_cc_step("raw/nemotro-cc").as_executor_step(),
 }
 
 _nemotron_cc_path = output_path_of(downloads["nemotron_cc"], "contrib/Nemotron/Nemotron-CC/data-jsonl/")

lib/marin/src/marin/datakit/download/dclm_hq.py

@@ -34,12 +34,6 @@
 logger = logging.getLogger(__name__)
 
 
-@dataclass
-class DCLMHQDownloadConfig:
-    input_path: str
-    output_path: str
-
-
 @dataclass
 class FileTask:
     """Represents a single file processing task."""
@@ -172,21 +166,8 @@ def process_file(task: FileTask) -> None:
         raise
 
 
-def extract_dclm_hq_dump(input_path_or_cfg: str | DCLMHQDownloadConfig, output_path: str | None = None) -> None:
-    """Process the DCLM HQ dump and enrich with HTML from Common Crawl.
-
-    Args:
-        input_path_or_cfg: Input directory path, or a DCLMHQDownloadConfig for backward compat.
-        output_path: Output directory path. Required when input_path_or_cfg is a string.
-    """
-    if isinstance(input_path_or_cfg, DCLMHQDownloadConfig):
-        input_path = input_path_or_cfg.input_path
-        output_path = input_path_or_cfg.output_path
-    else:
-        input_path = input_path_or_cfg
-        if output_path is None:
-            raise ValueError("output_path is required when input_path_or_cfg is a string")
-
+def extract_dclm_hq_dump(input_path: str, output_path: str) -> None:
+    """Process the DCLM HQ dump and enrich with HTML from Common Crawl."""
     logger.info(f"Starting processing of DCLM HQ dump in {input_path}")
 
     all_files = []

lib/marin/src/marin/datakit/download/filesystem.py

@@ -5,7 +5,6 @@
 import os
 import random
 import time
-from dataclasses import dataclass
 
 from iris.marin_fs import url_to_fs
 from marin.execution.step_spec import StepSpec
@@ -16,16 +15,6 @@
 logger = logging.getLogger(__name__)
 
 
-@dataclass
-class TransferConfig:
-    """Kept for backward compatibility. Prefer ``transfer_files()`` with flat params."""
-
-    input_path: str
-    output_path: str
-    num_random_files: int | None = None
-    filetype: str = "jsonl.zst"
-
-
 def transfer_files(
     input_path: str,
     output_path: str,

lib/marin/src/marin/datakit/download/nemotron_cc.py

@@ -14,12 +14,10 @@
 import logging
 import os
 from collections.abc import Iterator
-from dataclasses import dataclass
 
 import requests
 import zstandard
 from iris.marin_fs import open_url
-from marin.execution import THIS_OUTPUT_PATH
 from marin.execution.step_spec import StepSpec
 from marin.utils import fsspec_exists
 from requests.adapters import HTTPAdapter
@@ -85,22 +83,8 @@ def download_single_nemotron_path(input_file_path: str, output_file_path: str) -
     return {"input_file": input_file_path, "output_file": output_file_path, "num_records": num_records}
 
 
-@dataclass
-class NemotronIngressConfig:
-    """Kept for backward compatibility with ExecutorStep callers."""
-
-    output_path: str = THIS_OUTPUT_PATH
-
-
-def download_nemotron_cc(output_path_or_cfg: str | NemotronIngressConfig) -> None:
-    """Download and process Nemotron-CC dataset from Common Crawl.
-
-    Args:
-        output_path_or_cfg: Output directory path, or a NemotronIngressConfig for backward compat.
-    """
-    output_path = (
-        output_path_or_cfg.output_path if isinstance(output_path_or_cfg, NemotronIngressConfig) else output_path_or_cfg
-    )
+def download_nemotron_cc(output_path: str) -> None:
+    """Download and process Nemotron-CC dataset from Common Crawl."""
 
     paths_file_path = os.path.join(output_path, "data-jsonl.paths")
     logger.info(f"Downloading Nemotron CC path file {paths_file_path}")

lib/marin/src/marin/download/dclm_hq/download_dclm_hq_html.py

@@ -2,7 +2,6 @@
 # SPDX-License-Identifier: Apache-2.0
 # Backward-compat shim. Canonical location: marin.datakit.download.dclm_hq
 
-from marin.datakit.download.dclm_hq import DCLMHQDownloadConfig as DCLMHQDownloadConfig
 from marin.datakit.download.dclm_hq import FileTask as FileTask
 from marin.datakit.download.dclm_hq import extract_dclm_hq_dump as extract_dclm_hq_dump
 from marin.datakit.download.dclm_hq import fetch_warc_from_cc as fetch_warc_from_cc

lib/marin/src/marin/download/filesystem/transfer.py

@@ -2,5 +2,4 @@
 # SPDX-License-Identifier: Apache-2.0
 # Backward-compat shim. Canonical location: marin.datakit.download.filesystem
 
-from marin.datakit.download.filesystem import TransferConfig as TransferConfig
 from marin.datakit.download.filesystem import transfer_files as transfer_files

lib/marin/src/marin/download/nemotron_cc/download_nemotron_cc.py

@@ -2,6 +2,5 @@
 # SPDX-License-Identifier: Apache-2.0
 # Backward-compat shim. Canonical location: marin.datakit.download.nemotron_cc
 
-from marin.datakit.download.nemotron_cc import NemotronIngressConfig as NemotronIngressConfig
 from marin.datakit.download.nemotron_cc import download_nemotron_cc as download_nemotron_cc
 from marin.datakit.download.nemotron_cc import download_single_nemotron_path as download_single_nemotron_path

tests/download/test_dclm_hq.py

@@ -7,7 +7,7 @@
 from unittest.mock import patch
 
 import zstandard as zstd
-from marin.datakit.download.dclm_hq import DCLMHQDownloadConfig, extract_dclm_hq_dump
+from marin.datakit.download.dclm_hq import extract_dclm_hq_dump
 
 SAMPLE_DCLM_RECORDS = [
     {
@@ -171,8 +171,7 @@ def mock_requests_get(url, **kwargs):
         raise ValueError(f"Unexpected URL: {url}")
 
     with patch("marin.datakit.download.dclm_hq.requests.get", side_effect=mock_requests_get):
-        cfg = DCLMHQDownloadConfig(input_path=str(tmp_path / "input"), output_path=str(output_dir))
-        extract_dclm_hq_dump(cfg)
+        extract_dclm_hq_dump(str(tmp_path / "input"), str(output_dir))
 
     # Verify output files were created in nested structure
     shard1_output = output_dir / "shard1"

tests/download/test_nemotron_cc.py

@@ -9,7 +9,7 @@
 import pytest
 import zstandard as zstd
 from iris.marin_fs import open_url as _real_open_url
-from marin.datakit.download.nemotron_cc import NemotronIngressConfig, download_nemotron_cc
+from marin.datakit.download.nemotron_cc import download_nemotron_cc
 
 _OPEN_URL_TARGET = "marin.datakit.download.nemotron_cc.open_url"
 _REQUESTS_SESSION_TARGET = "marin.datakit.download.nemotron_cc.requests.Session"
@@ -114,8 +114,7 @@ def test_download_nemotron_cc_pipeline(tmp_path, mock_paths_open):
         patch(_OPEN_URL_TARGET, side_effect=mock_paths_open(paths)),
         patch(_REQUESTS_SESSION_TARGET, _mock_session_for({"file1": file1_data, "file2": file2_data})),
     ):
-        cfg = NemotronIngressConfig(output_path=str(output_dir))
-        download_nemotron_cc(cfg)
+        download_nemotron_cc(str(output_dir))
 
     all_records = read_all_jsonl_zst(output_dir / "contrib" / "Nemotron")
 
@@ -152,8 +151,7 @@ def test_download_nemotron_cc_dolma_format(tmp_path, mock_paths_open):
         patch(_OPEN_URL_TARGET, side_effect=mock_paths_open(paths)),
         patch(_REQUESTS_SESSION_TARGET, _mock_session_for({"test": compressed_data})),
     ):
-        cfg = NemotronIngressConfig(output_path=str(output_dir))
-        download_nemotron_cc(cfg)
+        download_nemotron_cc(str(output_dir))
 
     records = read_all_jsonl_zst(output_dir / "contrib" / "Nemotron")
     assert len(records) == 1
@@ -188,8 +186,7 @@ def test_download_nemotron_cc_skips_existing(tmp_path, mock_paths_open):
         patch(_OPEN_URL_TARGET, side_effect=mock_paths_open(paths)),
         patch(_REQUESTS_SESSION_TARGET) as mock_session,
     ):
-        cfg = NemotronIngressConfig(output_path=str(output_dir))
-        download_nemotron_cc(cfg)
+        download_nemotron_cc(str(output_dir))
 
     mock_session.return_value.get.assert_not_called()
     assert existing_output.read_text() == "existing"