Phase 2: streaming writer and chunked text parser

writers/txt.py:
- Stream each confidence-estimate DataFrame directly to CSV instead of
  materialising the full concatenated result first.  Uses mode='a' +
  header=False for all but the first chunk, eliminating the peak-memory
  spike from pd.concat() before the write.

parsers/txt.py:
- Add chunk_size parameter to read_txt().  When set, returns a generator
  that yields one pandas DataFrame per chunk_size rows, with the target
  column already converted to bool.  The non-chunked path is unchanged.
  Chunked mode is intended as the ingestion layer for the Phase 3 DuckDB
  backend.

utils.py:
- Add chunk_size parameter to parse_psms_txt().  When set, returns a
  pandas TextFileReader iterator instead of a full DataFrame, allowing
  callers (e.g. Tide/Comet/MSGF+ parsers) to read large files row-batch
  by row-batch.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: wsnoble

crema/parsers/txt.py

@@ -20,6 +20,7 @@ def read_txt(
     sep="\t",
     pairing_file_name=None,
     copy_data=False,
+    chunk_size=None,
 ):
     """Read peptide-spectrum matches (PSMs) from delimited text files.
 
@@ -57,12 +58,23 @@ def read_txt(
         is safer because it prevents accidental modification of the underlying
         data. This argument only has an effect when `pin_files` is a
         :py:class:`pandas.DataFrame`
+    chunk_size : int or None, optional
+        When set, the parser operates in streaming mode: instead of loading all
+        files into memory at once, it returns a generator that yields one
+        :py:class:`pandas.DataFrame` per chunk of ``chunk_size`` rows. This is
+        useful for feeding data into a DuckDB relation or a Parquet writer
+        without ever holding the full dataset in RAM. When ``None`` (default),
+        the existing behaviour is preserved and a
+        :py:class:`~crema.dataset.PsmDataset` is returned.
 
     Returns
     -------
-    PsmDataset
-        A :py:class:`~crema.dataset.PsmDataset` object containing the parsed
-        PSMs.
+    PsmDataset or generator of pandas.DataFrame
+        When ``chunk_size`` is ``None``, a
+        :py:class:`~crema.dataset.PsmDataset` object containing the parsed
+        PSMs.  When ``chunk_size`` is set, a generator of
+        :py:class:`pandas.DataFrame` chunks (each with the target column
+        already converted to bool).
     """
     # Store column names in a list to be used by read_csv function
     fields = [target_column, peptide_column, protein_column]
@@ -72,6 +84,12 @@ def read_txt(
     score_columns = utils.listify(score_columns)
     fields += spectrum_columns + score_columns
 
+    # Streaming mode: yield chunks without constructing a PsmDataset
+    if chunk_size is not None and not isinstance(txt_files, pd.DataFrame):
+        return _read_txt_chunked(
+            utils.listify(txt_files), sep, fields, target_column, chunk_size
+        )
+
     # Parse the data
     if isinstance(txt_files, pd.DataFrame):
         data = txt_files.copy(deep=copy_data).loc[:, fields]
@@ -100,6 +118,38 @@ def read_txt(
     return psms
 
 
+def _read_txt_chunked(txt_files, sep, cols, target_column, chunk_size):
+    """Yield DataFrames of ``chunk_size`` rows from one or more text files.
+
+    Parameters
+    ----------
+    txt_files : list of str
+        The files to read.
+    sep : str
+        The delimiter.
+    cols : list of str
+        The columns to retain.
+    target_column : str
+        The column containing target/decoy labels (converted to bool per chunk).
+    chunk_size : int
+        Number of rows per chunk.
+
+    Yields
+    ------
+    pandas.DataFrame
+        A chunk with the target column already converted to bool.
+    """
+    for txt_file in txt_files:
+        LOGGER.info(
+            "Reading PSMs from %s in chunks of %d...", txt_file, chunk_size
+        )
+        for chunk in pd.read_csv(
+            txt_file, sep=sep, usecols=cols, chunksize=chunk_size
+        ):
+            chunk[target_column] = _convert_target_col(chunk[target_column])
+            yield chunk
+
+
 def _parse_psms(txt_file, sep, cols):
     """Parse a single delimited txt file.
 

crema/utils.py

@@ -65,7 +65,7 @@ def create_pairing_from_file(pairing_file_name):
     return dict(zip(pairing_file[target_field], pairing_file[decoy_field]))
 
 
-def parse_psms_txt(txt_file, cols, skip_line):
+def parse_psms_txt(txt_file, cols, skip_line, chunk_size=None):
     """Parse a single tab-delimited file
 
     Parameters
@@ -76,11 +76,17 @@ def parse_psms_txt(txt_file, cols, skip_line):
         The columns to parse.
     skip_line : bool
         If true, skip reading the first line.
+    chunk_size : int or None, optional
+        When set, return a :py:class:`pandas.io.parsers.TextFileReader`
+        iterator that yields ``chunk_size``-row DataFrames instead of loading
+        the whole file.  When ``None`` (default), the full DataFrame is
+        returned.
 
     Returns
     -------
-    pandas.DataFrame
-        A :py:class:`pandas.DataFrame` containing the parsed PSMs
+    pandas.DataFrame or pandas.io.parsers.TextFileReader
+        A :py:class:`pandas.DataFrame` containing the parsed PSMs, or an
+        iterator of chunks when ``chunk_size`` is set.
     """
     LOGGER.info("Reading PSMs from %s...", txt_file)
 
@@ -102,15 +108,20 @@ def parse_psms_txt(txt_file, cols, skip_line):
             header = fh.readline().rstrip("\n").split("\t")
         explicit_cols = [c for c in header if c in cols]
     except UnicodeDecodeError:
+        if chunk_size is not None:
+            fallback_kwargs["chunksize"] = chunk_size
         return pd.read_csv(txt_file, **fallback_kwargs)
 
+    read_kwargs = dict(
+        sep="\t",
+        skiprows=int(skip_line),
+        usecols=explicit_cols,
+    )
+    if chunk_size is not None:
+        read_kwargs["chunksize"] = chunk_size
+        return pd.read_csv(txt_file, **read_kwargs)
+
     try:
-        return pd.read_csv(
-            txt_file,
-            sep="\t",
-            skiprows=int(skip_line),
-            usecols=explicit_cols,
-            engine="pyarrow",
-        )
+        return pd.read_csv(txt_file, engine="pyarrow", **read_kwargs)
     except (ImportError, ValueError):
         return pd.read_csv(txt_file, **fallback_kwargs)

crema/writers/txt.py

@@ -3,8 +3,6 @@
 from pathlib import Path
 from collections import defaultdict
 
-import pandas as pd
-
 
 def to_txt(
     conf, output_dir=None, file_root=None, sep="\t", decoys=False, precision=6
@@ -64,9 +62,17 @@ def to_txt(
     out_files = []
     for level, qval_list in results.items():
         out_file = str(file_base) + f".{level}.txt"
-        pd.concat(qval_list).to_csv(
-            out_file, sep=sep, index=False, float_format=f"%.{precision}f"
-        )
+        first = True
+        for df in qval_list:
+            df.to_csv(
+                out_file,
+                sep=sep,
+                index=False,
+                float_format=f"%.{precision}f",
+                header=first,
+                mode="w" if first else "a",
+            )
+            first = False
         out_files.append(out_file)
 
     return out_files