diff --git a/docs/index.rst b/docs/index.rst
index 83b607d..0857696 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -4,6 +4,9 @@ science_catalogs
 ``science_catalogs`` is a reusable toolkit for preparing science-ready catalogs
 with Dask, LSDB, and HATS.
 
+``prepare_catalog`` accepts a single input path and detects whether it points
+to a regular file collection or an existing HATS catalog.
+
 Beta public API
 ----------------------------------------------------------------------------------------
 
@@ -55,6 +58,7 @@ A runnable example with fictitious parquet inputs lives in
 
 - sample parquet files
 - YAML configs for parquet and HATS output
+- documentation for using an existing HATS catalog as input
 - an interactive walkthrough
 - a smoke test script
 
diff --git a/docs/quickstart.rst b/docs/quickstart.rst
index 211c009..61081b3 100644
--- a/docs/quickstart.rst
+++ b/docs/quickstart.rst
@@ -8,6 +8,40 @@ This package is designed around two layers:
 - ``materialize_catalog`` and ``materialize_lsdb_catalog`` return ``data`` plus
   the written output ``path``.
 
+Use ``materialize_catalog`` only when the final dataframe is expected to fit in
+memory. For larger catalogs, prefer ``build_catalog``, ``write_catalog``, or
+``materialize_lsdb_catalog`` so the workflow stays distributed and disk-backed.
+
+Input modes
+----------------------------------------------------------------------------------------
+
+``prepare_catalog`` can start from two kinds of inputs:
+
+- a regular file or directory
+- an existing HATS catalog opened through LSDB
+
+For file-based inputs, use a single path:
+
+.. code-block:: yaml
+
+   input:
+     catalog_path: /path/to/files_or_file
+     catalog_pattern: "*.parquet"
+
+If ``catalog_path`` is a single file, it is used directly. If it is a regular
+directory, ``catalog_pattern`` is used to find the files to process.
+
+For HATS-based inputs, point ``catalog_path`` to the existing catalog
+directory:
+
+.. code-block:: yaml
+
+   input:
+     catalog_path: /path/to/existing_hats_catalog
+
+In both cases, ``user_selected_cols`` still defines the columns projected into
+the processing step.
+
 Minimal interactive flow
 ----------------------------------------------------------------------------------------
 
diff --git a/examples/notebooks/demo_notebook.ipynb b/examples/notebooks/demo_notebook.ipynb
index 29441cd..63e6642 100644
--- a/examples/notebooks/demo_notebook.ipynb
+++ b/examples/notebooks/demo_notebook.ipynb
@@ -15,12 +15,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "from glob import glob\n",
+    "\n",
     "import dask.dataframe as dd\n",
     "import lsdb\n",
     "import pandas as pd\n",
-    "from glob import glob\n",
     "from dask.distributed import Client, LocalCluster\n",
-    "from science_catalogs import prepare_catalog, write_catalog, open_lsdb_catalog"
+    "\n",
+    "from science_catalogs import open_lsdb_catalog, prepare_catalog, write_catalog"
    ]
   },
   {
diff --git a/pyproject.toml b/pyproject.toml
index cab526e..e579537 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -40,9 +40,9 @@ dependencies = [
     "jinja2==3.1.*",
     "h5py",
     "cdshealpix>=0.8.1,<0.9.0",
-    "lsdb>=0.9.0,<0.11.0",
-    "hats>=0.9.0,<0.11.0",
-    "hats-import>=0.9.0,<0.11.0",
+    "lsdb>=0.9.2,<0.11.0",
+    "hats>=0.9.2,<0.11.0",
+    "hats-import>=0.9.2,<0.11.0",
 ]
 
 [project.optional-dependencies]
@@ -121,5 +121,28 @@ ignore = [
     "UP028",
 ]
 
+[tool.ruff.lint.per-file-ignores]
+"*.ipynb" = [
+    "E",
+    "W",
+    "F",
+    "N",
+    "UP",
+    "B",
+    "SIM",
+    "I",
+    "D101",
+    "D102",
+    "D103",
+    "D106",
+    "D206",
+    "D207",
+    "D208",
+    "D300",
+    "D417",
+    "D419",
+    "NPY201",
+]
+
 [tool.coverage.run]
 omit = ["src/science_catalogs/_version.py"]
diff --git a/src/science_catalogs/catalog.py b/src/science_catalogs/catalog.py
index a59a3e6..2e60f15 100644
--- a/src/science_catalogs/catalog.py
+++ b/src/science_catalogs/catalog.py
@@ -13,7 +13,7 @@
 from dask.distributed import Client, wait
 
 from science_catalogs.executor import get_executor
-from science_catalogs.processing import process_file_df
+from science_catalogs.processing import process_dataframe, process_file_df
 from science_catalogs.utils.config import decide_suffix_and_flags
 from science_catalogs.utils.dust import configure_dustmaps_path
 from science_catalogs.utils.partitioning import reorder_and_rechunk
@@ -41,8 +41,80 @@ def load_catalog_config(config_path: str) -> dict[str, Any]:
         return yaml.safe_load(_file) or {}
 
 
-def prepare_catalog(config_path: str, config: dict[str, Any] | None = None) -> PreparedCatalog:
-    """Build the lazy processed catalog without creating workflow-specific artifacts."""
+def _is_hats_catalog_path(path: Path) -> bool:
+    """Check whether a directory already contains a valid HATS catalog."""
+    if not path.is_dir():
+        return False
+
+    from hats.io.validation import is_valid_catalog
+
+    return bool(is_valid_catalog(path))
+
+
+def _glob_input_files(base_path: Path, pattern: str) -> list[str]:
+    """Collect matching input files below a directory."""
+    return sorted(glob.glob((base_path / pattern).as_posix()))
+
+
+def _resolve_input_source(inputs: dict[str, Any]) -> dict[str, Any]:
+    """Resolve a single input path as HATS, a single file, or a directory of files."""
+    raw_catalog_path = inputs.get("catalog_path")
+    raw_catalog_folder = inputs.get("catalog_folder")
+    pattern = inputs.get("catalog_pattern", "*.parquet")
+
+    if raw_catalog_path:
+        catalog_path = Path(raw_catalog_path).expanduser()
+    elif raw_catalog_folder:
+        catalog_path = Path(raw_catalog_folder).expanduser()
+    else:
+        raise ValueError("input.catalog_path is required")
+
+    if not catalog_path.exists():
+        raise FileNotFoundError(f"Input path does not exist: {catalog_path}")
+
+    if _is_hats_catalog_path(catalog_path):
+        return {"source": "hats", "catalog_path": str(catalog_path)}
+
+    if catalog_path.is_file():
+        return {"source": "files", "input_files": [str(catalog_path)]}
+
+    if catalog_path.is_dir():
+        input_files = _glob_input_files(catalog_path, pattern)
+        if not input_files:
+            raise FileNotFoundError(f"No input files found under {catalog_path} matching pattern {pattern!r}")
+        return {"source": "files", "input_files": input_files}
+
+    raise ValueError(f"Unsupported input path type: {catalog_path}")
+
+
+def _build_processed_meta(
+    ddf: dd.DataFrame,
+    cfg: dict[str, Any],
+    *,
+    will_mag: bool,
+    will_dered_flux: bool,
+    will_dered_mag: bool,
+):
+    """Infer the processed partition schema for Dask map_partitions."""
+    meta_input = ddf._meta.copy()
+    meta_output = process_dataframe(
+        meta_input,
+        cfg,
+        will_mag=will_mag,
+        will_dered_flux=will_dered_flux,
+        will_dered_mag=will_dered_mag,
+        source_name="<meta>",
+    )
+    return meta_output.iloc[:0]
+
+
+def prepare_catalog(
+    config_path: str,
+    config: dict[str, Any] | None = None,
+    *,
+    client=None,
+) -> PreparedCatalog:
+    """Build the lazy processed catalog from file inputs or an existing HATS catalog."""
     cfg = config if config is not None else load_catalog_config(config_path)
 
     inputs = cfg.get("input", {})
@@ -57,28 +129,45 @@ def prepare_catalog(config_path: str, config: dict[str, Any] | None = None) -> P
         inputs.get("compute_dereddening", True),
     )
 
-    input_files = [
-        f
-        for f in glob.glob(
-            Path(inputs.get("catalog_folder", "")).expanduser().as_posix()
-            + "/"
-            + inputs.get("catalog_pattern", "*.parquet")
+    input_source = _resolve_input_source(inputs)
+
+    if input_source["source"] == "files":
+        input_files = input_source["input_files"]
+        delayed_dfs = [
+            delayed(process_file_df)(
+                p,
+                cfg_path=config_path,
+                will_mag=will_mag,
+                will_dered_flux=will_dered_flux,
+                will_dered_mag=will_dered_mag,
+            )
+            for p in input_files
+        ]
+        ddf = dd.from_delayed(delayed_dfs)
+    else:
+        input_files = [input_source["catalog_path"]]
+        selected_columns = list(inputs.get("user_selected_cols", []) or []) or None
+        hats_catalog = open_lsdb_catalog(
+            input_source["catalog_path"],
+            client=client,
+            columns=selected_columns,
         )
-    ]
-    if not input_files:
-        raise FileNotFoundError("No input files found for catalog_pattern")
-
-    delayed_dfs = [
-        delayed(process_file_df)(
-            p,
-            cfg_path=config_path,
+        ddf = hats_catalog.to_dask_dataframe()
+        ddf = ddf.map_partitions(
+            process_dataframe,
+            cfg,
             will_mag=will_mag,
             will_dered_flux=will_dered_flux,
             will_dered_mag=will_dered_mag,
+            source_name=input_source["catalog_path"],
+            meta=_build_processed_meta(
+                ddf,
+                cfg,
+                will_mag=will_mag,
+                will_dered_flux=will_dered_flux,
+                will_dered_mag=will_dered_mag,
+            ),
         )
-        for p in input_files
-    ]
-    ddf = dd.from_delayed(delayed_dfs)
     ddf_out = reorder_and_rechunk(ddf, output_cfg)
 
     return PreparedCatalog(
@@ -237,7 +326,7 @@ def build_catalog(
             wait(cluster_comm)
         client.run(lambda: gc.collect())
 
-        prepared = prepare_catalog(config_path, config=cfg)
+        prepared = prepare_catalog(config_path, config=cfg, client=client)
         resolved_output_dir = _resolve_output_dir(prepared, output_dir)
         written_paths = write_catalog(
             prepared,
diff --git a/src/science_catalogs/processing.py b/src/science_catalogs/processing.py
index e52de0f..f95dde0 100644
--- a/src/science_catalogs/processing.py
+++ b/src/science_catalogs/processing.py
@@ -1,4 +1,4 @@
-"""Per-file processing logic."""
+"""Catalog processing logic shared by file and HATS inputs."""
 
 import astropy.units as u
 import numpy as np
@@ -20,25 +20,30 @@ def _keep_input_columns(input_cfg):
     return bool(input_cfg.get(LEGACY_KEEP_INPUT_COLUMNS_KEY, False))
 
 
-def process_file_df(
-    path: str,
-    cfg_path: str,
+def _load_processing_config(cfg_path: str) -> dict:
+    """Load the processing configuration from disk without importing catalog helpers."""
+    import yaml
+
+    with open(cfg_path, "r", encoding="utf-8") as _file:
+        return yaml.safe_load(_file) or {}
+
+
+def process_dataframe(
+    df,
+    cfg: dict,
+    *,
     will_mag: bool,
     will_dered_flux: bool,
     will_dered_mag: bool,
+    source_name: str = "<dataframe>",
 ):
-    """Read, filter, transform, and return a single catalog file as a dataframe."""
-    import yaml
+    """Filter and transform a dataframe according to the catalog configuration."""
+    df = df.copy()
+    input_cfg = cfg.get("input", {})
+    dust_cfg = cfg.get("dust", {})
+    output_cfg = cfg.get("output", {})
+    invalid = cfg.get("invalid_handling", {})
 
-    with open(cfg_path, "r", encoding="utf-8") as _f:
-        cfgw = yaml.safe_load(_f) or {}
-
-    input_cfg = cfgw.get("input", {})
-    dust_cfg = cfgw.get("dust", {})
-    output_cfg = cfgw.get("output", {})
-    invalid = cfgw.get("invalid_handling", {})
-
-    input_user_selected_cols = list(input_cfg.get("user_selected_cols", []) or [])
     is_id_index = bool(input_cfg.get("is_id_in_index", False))
 
     col_pattern = input_cfg.get("col_pattern")
@@ -50,14 +55,12 @@ def process_file_df(
     mag_offset = as_float_or_none(output_cfg.get("mag_offset"))
     a_ebv = dict(output_cfg.get("A_EBV", {}))
 
-    df = detect_and_read(path, input_user_selected_cols)
-
     filt = input_cfg.get("filter", {})
     if filt.get("enabled"):
         col = filt.get("column")
         val = filt.get("value")
         if col not in df.columns:
-            raise ValueError(f"Boolean column '{col}' not found in {path}")
+            raise ValueError(f"Boolean column '{col}' not found in {source_name}")
         df = df[df[col] == val]
         if filt.get("drop_column_after_filter") and col in df.columns:
             df = df.drop(columns=[col])
@@ -66,7 +69,7 @@ def process_file_df(
     if init.get("enabled"):
         cut_col = init.get("column")
         if cut_col not in df.columns:
-            raise ValueError(f"Initial-cut column '{cut_col}' not found in {path}")
+            raise ValueError(f"Initial-cut column '{cut_col}' not found in {source_name}")
         col_type = str(init.get("column_type", "flux")).strip().lower()
         mag_val = as_float_or_none(init.get("mag_value"))
         flux_val = as_float_or_none(init.get("flux_value"))
@@ -179,7 +182,7 @@ def apply_replacement(arr, mask, replacement_value):
         final_output_cols.update([final_col, final_err_col])
 
         if col_in not in df.columns or err_in not in df.columns:
-            raise ValueError(f"Missing column(s) {[col_in, err_in]} in file {path}")
+            raise ValueError(f"Missing column(s) {[col_in, err_in]} in source {source_name}")
 
         values = df[col_in].astype(float, copy=False).values
         errors = df[err_in].astype(float, copy=False).values
@@ -239,4 +242,26 @@ def apply_replacement(arr, mask, replacement_value):
     return df
 
 
-__all__ = ["process_file_df", "MAG_CONV"]
+def process_file_df(
+    path: str,
+    cfg_path: str,
+    will_mag: bool,
+    will_dered_flux: bool,
+    will_dered_mag: bool,
+):
+    """Read, filter, transform, and return a single catalog file as a dataframe."""
+    cfgw = _load_processing_config(cfg_path)
+    input_cfg = cfgw.get("input", {})
+    input_user_selected_cols = list(input_cfg.get("user_selected_cols", []) or [])
+    df = detect_and_read(path, input_user_selected_cols)
+    return process_dataframe(
+        df,
+        cfgw,
+        will_mag=will_mag,
+        will_dered_flux=will_dered_flux,
+        will_dered_mag=will_dered_mag,
+        source_name=path,
+    )
+
+
+__all__ = ["process_dataframe", "process_file_df", "MAG_CONV"]
diff --git a/tests/test_build_catalog.py b/tests/test_build_catalog.py
index 17d41cd..aa6f43f 100644
--- a/tests/test_build_catalog.py
+++ b/tests/test_build_catalog.py
@@ -43,7 +43,12 @@ def test_build_catalog_writes_parquet(monkeypatch):
     calls = {}
 
     monkeypatch.setattr("science_catalogs.catalog.load_catalog_config", lambda path: {"cluster": {}})
-    monkeypatch.setattr("science_catalogs.catalog.prepare_catalog", lambda path, config=None: prepared)
+
+    def fake_prepare_catalog(path, config=None, client=None):
+        calls["prepare_client"] = client
+        return prepared
+
+    monkeypatch.setattr("science_catalogs.catalog.prepare_catalog", fake_prepare_catalog)
 
     def fake_write_catalog(prepared, output_dir, client=None, output_format=None):
         calls["output_dir"] = output_dir
@@ -55,6 +60,7 @@ def fake_write_catalog(prepared, output_dir, client=None, output_format=None):
     result = build_catalog("config.yml", output_dir="/tmp/out", output_format="parquet")
 
     assert result == "/tmp/out/part0.parquet"
+    assert calls["prepare_client"] is not None
     assert calls["output_dir"] == "/tmp/out"
     assert calls["output_format"] == "parquet"
 
@@ -66,7 +72,12 @@ def test_build_catalog_defaults_to_cwd_data(monkeypatch, tmp_path):
     captured = {}
 
     monkeypatch.setattr("science_catalogs.catalog.load_catalog_config", lambda path: {"cluster": {}})
-    monkeypatch.setattr("science_catalogs.catalog.prepare_catalog", lambda path, config=None: prepared)
+
+    def fake_prepare_catalog(path, config=None, client=None):
+        captured["prepare_client"] = client
+        return prepared
+
+    monkeypatch.setattr("science_catalogs.catalog.prepare_catalog", fake_prepare_catalog)
 
     def fake_write_catalog(prepared, output_dir, client=None, output_format=None):
         captured["output_dir"] = output_dir
@@ -80,6 +91,7 @@ def fake_write_catalog(prepared, output_dir, client=None, output_format=None):
 
     expected = str(Path(tmp_path) / "data")
     assert result == (f"{expected}/part0.parquet", f"{expected}/part1.parquet")
+    assert captured["prepare_client"] is not None
     assert captured["output_dir"] == expected
     assert captured["output_format"] is None
 
@@ -92,7 +104,12 @@ def test_build_catalog_writes_hats(monkeypatch):
     calls = {}
 
     monkeypatch.setattr("science_catalogs.catalog.load_catalog_config", lambda path: {"cluster": {}})
-    monkeypatch.setattr("science_catalogs.catalog.prepare_catalog", lambda path, config=None: prepared)
+
+    def fake_prepare_catalog(path, config=None, client=None):
+        calls["prepare_client"] = client
+        return prepared
+
+    monkeypatch.setattr("science_catalogs.catalog.prepare_catalog", fake_prepare_catalog)
 
     def fake_write_catalog(prepared, output_dir, client=None, output_format=None):
         calls["output_dir"] = output_dir
@@ -104,5 +121,6 @@ def fake_write_catalog(prepared, output_dir, client=None, output_format=None):
     result = build_catalog("config.yml", output_dir="/tmp/out", output_format="hats")
 
     assert result == "/tmp/out/demo_collection"
+    assert calls["prepare_client"] is not None
     assert calls["output_dir"] == "/tmp/out"
     assert calls["output_format"] == "hats"
diff --git a/tests/test_prepare_catalog.py b/tests/test_prepare_catalog.py
new file mode 100644
index 0000000..ae8b3e9
--- /dev/null
+++ b/tests/test_prepare_catalog.py
@@ -0,0 +1,167 @@
+"""Tests for prepare_catalog input resolution."""
+
+from pathlib import Path
+
+import dask.dataframe as dd
+import pandas as pd
+import pytest
+from science_catalogs.catalog import _resolve_input_source, prepare_catalog
+
+
+def _base_cfg():
+    return {
+        "input": {
+            "ra_col": "ra",
+            "dec_col": "dec",
+            "user_selected_cols": ["object_id", "ra", "dec", "MAG_G_DERED", "MAGERR_G"],
+            "col_pattern": "MAG_BAND_DERED",
+            "err_pattern": "MAGERR_BAND",
+            "selected_bands": ["G"],
+            "band_case": "lower_case",
+            "keep_input_columns_after_filters_or_transformations": False,
+        },
+        "output": {
+            "col_final_pattern": "mag_BAND",
+            "err_final_pattern": "magerr_BAND",
+            "band_case": "lower_case",
+        },
+    }
+
+
+def test_resolve_input_source_defaults_to_files(tmp_path):
+    """Treat a regular directory catalog_path as a file collection."""
+    first = tmp_path / "part1.csv"
+    second = tmp_path / "part2.csv"
+    first.write_text("id\n1\n", encoding="utf-8")
+    second.write_text("id\n2\n", encoding="utf-8")
+
+    resolved = _resolve_input_source(
+        {
+            "catalog_path": str(tmp_path),
+            "catalog_pattern": "*.csv",
+        }
+    )
+
+    assert resolved["source"] == "files"
+    assert set(resolved["input_files"]) == {str(first), str(second)}
+
+
+def test_resolve_input_source_accepts_single_file(tmp_path):
+    """Treat a single file catalog_path as a one-file catalog input."""
+    input_file = tmp_path / "part1.parquet"
+    input_file.write_text("placeholder", encoding="utf-8")
+
+    resolved = _resolve_input_source({"catalog_path": str(input_file)})
+
+    assert resolved["source"] == "files"
+    assert resolved["input_files"] == [str(input_file)]
+
+
+def test_resolve_input_source_detects_hats(monkeypatch, tmp_path):
+    """Treat a valid HATS directory as an LSDB-opened input."""
+    monkeypatch.setattr("science_catalogs.catalog._is_hats_catalog_path", lambda path: True)
+
+    resolved = _resolve_input_source({"catalog_path": str(tmp_path)})
+
+    assert resolved["source"] == "hats"
+    assert resolved["catalog_path"] == str(tmp_path)
+
+
+def test_resolve_input_source_requires_catalog_path():
+    """Reject configs without the unified input path."""
+    with pytest.raises(ValueError, match="input.catalog_path is required"):
+        _resolve_input_source({})
+
+
+def test_prepare_catalog_keeps_file_mode_behavior(monkeypatch, tmp_path):
+    """Prepare file-glob inputs through the existing per-file path."""
+    first = tmp_path / "part1.csv"
+    second = tmp_path / "part2.csv"
+    first.write_text("", encoding="utf-8")
+    second.write_text("", encoding="utf-8")
+
+    cfg = _base_cfg()
+    cfg["input"].update(
+        {
+            "catalog_path": str(tmp_path),
+            "catalog_pattern": "*.csv",
+        }
+    )
+
+    seen = []
+
+    monkeypatch.setattr("science_catalogs.catalog.configure_dustmaps_path", lambda dust: None)
+    monkeypatch.setattr(
+        "science_catalogs.catalog.decide_suffix_and_flags",
+        lambda *args, **kwargs: ("_demo", False, False, False),
+    )
+    monkeypatch.setattr("science_catalogs.catalog.reorder_and_rechunk", lambda ddf, output_cfg: ddf)
+
+    def fake_process_file_df(path, cfg_path, will_mag, will_dered_flux, will_dered_mag):
+        seen.append(Path(path).name)
+        return pd.DataFrame({"ra": [1.0], "dec": [2.0], "mag_g": [22.5], "magerr_g": [0.1]})
+
+    monkeypatch.setattr("science_catalogs.catalog.process_file_df", fake_process_file_df)
+
+    prepared = prepare_catalog("unused.yml", config=cfg)
+    result = prepared.ddf.compute()
+
+    assert set(prepared.input_files) == {str(first), str(second)}
+    assert {"part1.csv", "part2.csv"}.issubset(set(seen))
+    assert list(result.columns) == ["ra", "dec", "mag_g", "magerr_g"]
+    assert len(result) == 2
+
+
+def test_prepare_catalog_reads_hats_input(monkeypatch, tmp_path):
+    """Open an existing HATS catalog and process it lazily per partition."""
+    hats_path = tmp_path / "demo_hats_catalog"
+    hats_path.mkdir()
+
+    cfg = _base_cfg()
+    cfg["input"].update(
+        {
+            "catalog_path": str(hats_path),
+        }
+    )
+
+    source_df = pd.DataFrame(
+        {
+            "object_id": [1, 2],
+            "ra": [10.0, 11.0],
+            "dec": [-20.0, -21.0],
+            "MAG_G_DERED": [22.5, 23.0],
+            "MAGERR_G": [0.1, 0.2],
+        }
+    )
+    calls = {}
+    fake_client = object()
+
+    class _FakeCatalog:
+        def to_dask_dataframe(self):
+            return dd.from_pandas(source_df, npartitions=2)
+
+    monkeypatch.setattr("science_catalogs.catalog.configure_dustmaps_path", lambda dust: None)
+    monkeypatch.setattr(
+        "science_catalogs.catalog.decide_suffix_and_flags",
+        lambda *args, **kwargs: ("_demo", False, False, False),
+    )
+    monkeypatch.setattr("science_catalogs.catalog._is_hats_catalog_path", lambda path: True)
+    monkeypatch.setattr("science_catalogs.catalog.reorder_and_rechunk", lambda ddf, output_cfg: ddf)
+
+    def fake_open_lsdb_catalog(path, client=None, **kwargs):
+        calls["path"] = path
+        calls["client"] = client
+        calls["columns"] = kwargs.get("columns")
+        return _FakeCatalog()
+
+    monkeypatch.setattr("science_catalogs.catalog.open_lsdb_catalog", fake_open_lsdb_catalog)
+
+    prepared = prepare_catalog("unused.yml", config=cfg, client=fake_client)
+    result = prepared.ddf.compute()
+
+    assert prepared.input_files == [str(hats_path)]
+    assert calls["path"] == str(hats_path)
+    assert calls["client"] is fake_client
+    assert calls["columns"] == ["object_id", "ra", "dec", "MAG_G_DERED", "MAGERR_G"]
+    assert list(result.columns) == ["object_id", "ra", "dec", "mag_g", "magerr_g"]
+    assert len(result) == 2
diff --git a/tests/test_processing.py b/tests/test_processing.py
index 7c94fbc..34a8d5b 100644
--- a/tests/test_processing.py
+++ b/tests/test_processing.py
@@ -1,7 +1,8 @@
 """Tests for per-file catalog processing."""
 
+import pandas as pd
 import yaml
-from science_catalogs.processing import process_file_df
+from science_catalogs.processing import process_dataframe, process_file_df
 
 
 def _write_mag_dered_inputs(tmp_path, keep_key, keep_value):
@@ -73,3 +74,27 @@ def test_legacy_keep_input_columns_key_still_works(tmp_path):
     assert "magerr_g" in df.columns
     assert "MAG_G_DERED" in df.columns
     assert "MAGERR_G" in df.columns
+
+
+def test_process_dataframe_matches_file_wrapper(tmp_path):
+    """Apply the same science logic when the input is already in memory."""
+    input_path, cfg_path = _write_mag_dered_inputs(
+        tmp_path,
+        "keep_input_columns_after_filters_or_transformations",
+        False,
+    )
+
+    config = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
+    dataframe = pd.read_csv(input_path)
+
+    from_file = process_file_df(str(input_path), str(cfg_path), False, False, False)
+    from_dataframe = process_dataframe(
+        dataframe,
+        config,
+        will_mag=False,
+        will_dered_flux=False,
+        will_dered_mag=False,
+        source_name="memory.csv",
+    )
+
+    assert from_dataframe.equals(from_file)