Fix filter_* FK violation and add flux/filter consistency check

docs/tbls/public.fluxstd.md

@@ -102,8 +102,8 @@
 | fluxstd_pkey | CREATE UNIQUE INDEX fluxstd_pkey ON public.fluxstd USING btree (fluxstd_id) |
 | uq_obj_id_input_catalog_id_version | CREATE UNIQUE INDEX uq_obj_id_input_catalog_id_version ON public.fluxstd USING btree (obj_id, input_catalog_id, version) |
 | fluxstd_q3c_ang2ipix_idx | CREATE INDEX fluxstd_q3c_ang2ipix_idx ON public.fluxstd USING btree (q3c_ang2ipix(ra, "dec")) |
-| ix_fluxstd_version | CREATE INDEX ix_fluxstd_version ON public.fluxstd USING btree (version) |
 | ix_fluxstd_input_catalog_fluxstdid | CREATE INDEX ix_fluxstd_input_catalog_fluxstdid ON public.fluxstd USING btree (input_catalog_id, fluxstd_id) |
+| ix_fluxstd_version | CREATE INDEX ix_fluxstd_version ON public.fluxstd USING btree (version) |
 | ix_fluxstd_version_fluxstdid | CREATE INDEX ix_fluxstd_version_fluxstdid ON public.fluxstd USING btree (version, fluxstd_id) |
 
 ## Relations

docs/tbls/public.sky.md

@@ -34,9 +34,9 @@
 | ---- | ---------- |
 | sky_pkey | CREATE UNIQUE INDEX sky_pkey ON public.sky USING btree (sky_id) |
 | sky_obj_id_input_catalog_id_version_key | CREATE UNIQUE INDEX sky_obj_id_input_catalog_id_version_key ON public.sky USING btree (obj_id, input_catalog_id, version) |
+| ix_sky_input_catalog_id | CREATE INDEX ix_sky_input_catalog_id ON public.sky USING btree (input_catalog_id) |
 | sky_q3c_ang2ipix_idx | CREATE INDEX sky_q3c_ang2ipix_idx ON public.sky USING btree (q3c_ang2ipix(ra, "dec")) |
 | ix_sky_version | CREATE INDEX ix_sky_version ON public.sky USING btree (version) |
-| ix_sky_input_catalog_id | CREATE INDEX ix_sky_input_catalog_id ON public.sky USING btree (input_catalog_id) |
 
 ## Relations
 

docs/tbls/public.target.md

@@ -121,11 +121,11 @@
 | target_pkey | CREATE UNIQUE INDEX target_pkey ON public.target USING btree (target_id) |
 | target_propid_obcode_catid_objid_resolution_key | CREATE UNIQUE INDEX target_propid_obcode_catid_objid_resolution_key ON public.target USING btree (proposal_id, ob_code, input_catalog_id, obj_id, is_medium_resolution) |
 | target_propid_obcode_key | CREATE UNIQUE INDEX target_propid_obcode_key ON public.target USING btree (proposal_id, ob_code) |
+| target_proposal_id_idx | CREATE INDEX target_proposal_id_idx ON public.target USING btree (proposal_id) |
+| target_obj_id_input_catalog_id_idx | CREATE INDEX target_obj_id_input_catalog_id_idx ON public.target USING btree (obj_id, input_catalog_id) |
 | target_input_catalog_id_idx | CREATE INDEX target_input_catalog_id_idx ON public.target USING btree (input_catalog_id) |
 | target_proposal_id_obj_id_idx | CREATE INDEX target_proposal_id_obj_id_idx ON public.target USING btree (proposal_id, obj_id) |
 | target_q3c_ang2ipix_idx | CREATE INDEX target_q3c_ang2ipix_idx ON public.target USING btree (q3c_ang2ipix(ra, "dec")) |
-| target_proposal_id_idx | CREATE INDEX target_proposal_id_idx ON public.target USING btree (proposal_id) |
-| target_obj_id_input_catalog_id_idx | CREATE INDEX target_obj_id_input_catalog_id_idx ON public.target USING btree (obj_id, input_catalog_id) |
 
 ## Relations
 

src/targetdb/utils.py

@@ -371,6 +371,101 @@ def draw_diagram(
         raise ValueError(f"Unsupported generator: {generator}")
 
 
+FILTER_BANDS = ["u", "v", "g", "r", "i", "z", "y", "j"]
+
+
+def normalize_filter_columns(df):
+    """
+    Normalize filter_* columns (e.g., filter_g, filter_i) in place so that
+    missing values are stored as None instead of being passed through to the
+    database as-is.
+
+    filter_* columns are ForeignKey columns referencing filter_name.filter_name.
+    Depending on how the input was loaded, a missing value may show up as a
+    float NaN (e.g., from a masked ecsv column) or as an empty string (e.g.,
+    from a CSV column read with keep_default_na=False). Either one fails the
+    foreign key constraint if inserted as-is (NaN is even coerced to the
+    literal string "NaN" by the database driver), so both are normalized to
+    None here.
+
+    Parameters
+    ----------
+    df : pandas.DataFrame
+        The DataFrame to normalize. Modified in place.
+
+    Returns
+    -------
+    df : pandas.DataFrame
+        The same DataFrame, for convenience.
+    """
+    for band in FILTER_BANDS:
+        col = f"filter_{band}"
+        if col not in df.columns:
+            continue
+        df.loc[df[col].isna() | (df[col] == ""), col] = None
+    return df
+
+
+FLUX_COLUMN_PREFIXES = ["", "psf_", "total_"]
+
+
+def check_filter_flux_consistency(df):
+    """
+    Raise an error if a band has a flux value but no filter.
+
+    A flux (or flux error) measurement is meaningless without knowing which
+    filter it was measured through, so filter_{band} being None while
+    flux_{band} (or psf_flux_{band}, total_flux_{band}, and their
+    *_error_* counterparts) is set indicates a problem upstream in the input
+    data rather than something to silently accept.
+
+    Parameters
+    ----------
+    df : pandas.DataFrame
+        The DataFrame to check. filter_* columns are expected to already be
+        normalized (see `normalize_filter_columns`), i.e., missing filters
+        are represented as None rather than NaN or "".
+
+    Returns
+    -------
+    df : pandas.DataFrame
+        The same DataFrame, unmodified, for convenience.
+
+    Raises
+    ------
+    ValueError
+        If any row has a flux value for a band whose filter is None.
+    """
+    for band in FILTER_BANDS:
+        filter_col = f"filter_{band}"
+        if filter_col not in df.columns:
+            continue
+
+        missing_filter = df[filter_col].isna()
+        if not missing_filter.any():
+            continue
+
+        for prefix in FLUX_COLUMN_PREFIXES:
+            for kind in ["flux", "flux_error"]:
+                flux_col = f"{prefix}{kind}_{band}"
+                if flux_col not in df.columns:
+                    continue
+
+                inconsistent = missing_filter & df[flux_col].notna()
+                if inconsistent.any():
+                    bad_indices = df.index[inconsistent].tolist()
+                    logger.error(
+                        f"{flux_col} has values but {filter_col} is missing "
+                        f"for rows {bad_indices}."
+                    )
+                    raise ValueError(
+                        f"{flux_col} has values but {filter_col} is missing "
+                        f"for rows {bad_indices}. A flux value without a "
+                        "filter is not allowed."
+                    )
+    return df
+
+
 def join_backref_values(df, db=None, table=None, key=None, check_key=None):
     """
     Joins a DataFrame with a table from a database on a specified key and checks for non-existing keys.
@@ -445,6 +540,8 @@ def add_backref_values(df, db=None, table=None, upload_id=None):
     """
 
     df_tmp = df.copy()
+    normalize_filter_columns(df_tmp)
+    check_filter_flux_consistency(df_tmp)
     backref_tables, backref_keys, backref_check_keys = [], [], []
 
     if table == "target":
@@ -649,15 +746,10 @@ def make_target_df_from_uploader(
         logger.error(f"flux_type must be 'total' or 'psf'. {flux_type=}")
         raise ValueError(f"flux_type must be 'total' or 'psf'. {flux_type=}")
 
-    # fill missing values with None or NaN for filters and fluxes
+    normalize_filter_columns(df)
+
+    # rename flux columns to match the flux_type
     for band in ["g", "r", "i", "z", "y", "j"]:
-        if f"filter_{band}" in df.columns:
-            # if the table is a masked table, fill the masked values with None for filters
-            # if the table is not a masked table, just pass
-            try:
-                df.loc[df[f"filter_{band}"].isna(), f"filter_{band}"] = None
-            except AttributeError:
-                pass
         if f"flux_{band}" in df.columns:
             logger.info(f"flux_{band} is renamed to {flux_type}_flux_{band}")
             df.rename(

tests/test_utils.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from targetdb.utils import add_backref_values, check_filter_flux_consistency
+
+
+def test_add_backref_values_normalizes_missing_filter_values_for_fluxstd():
+    # filter_g/filter_i are ForeignKey columns to filter_name.filter_name.
+    # Empty entries in the input (NaN from a masked ecsv column, or an empty
+    # string from a CSV column read with keep_default_na=False) must be
+    # normalized to None, otherwise the literal "NaN"/"" is sent to the
+    # database and violates the foreign key constraint. This must hold for
+    # any table that goes through add_backref_values (fluxstd, sky,
+    # user_pointing, and target when not coming from the uploader), not just
+    # for the uploader-specific target code path.
+    df = pd.DataFrame(
+        {
+            "input_catalog_id": [1, 1, 1],
+            "filter_g": ["g_sdss", np.nan, ""],
+            "filter_i": ["i_sdss", "i_sdss", "i_sdss"],
+        }
+    )
+
+    df_out = add_backref_values(df, db=None, table="fluxstd")
+
+    assert df_out["filter_g"].tolist() == ["g_sdss", None, None]
+    assert df_out["filter_i"].tolist() == ["i_sdss", "i_sdss", "i_sdss"]
+
+    # None must round-trip as None (not float NaN) through to_dict, since
+    # that is what actually gets handed to bulk_insert_mappings.
+    records = df_out.to_dict(orient="records")
+    assert records[1]["filter_g"] is None
+    assert records[2]["filter_g"] is None
+
+
+def test_add_backref_values_raises_when_flux_present_without_filter():
+    # A flux value is meaningless without knowing which filter it was
+    # measured through, so filter_g missing while psf_flux_g is set must be
+    # rejected rather than silently inserted.
+    df = pd.DataFrame(
+        {
+            "input_catalog_id": [1, 1],
+            "filter_g": ["g_sdss", np.nan],
+            "psf_flux_g": [123.4, 56.7],
+        }
+    )
+
+    with pytest.raises(ValueError, match="psf_flux_g"):
+        add_backref_values(df, db=None, table="fluxstd")
+
+
+@pytest.mark.parametrize(
+    "flux_col", ["flux_g", "psf_flux_g", "total_flux_g", "psf_flux_error_g"]
+)
+def test_check_filter_flux_consistency_raises_for_each_flux_column_kind(flux_col):
+    df = pd.DataFrame({"filter_g": [None], flux_col: [1.0]})
+
+    with pytest.raises(ValueError, match=flux_col):
+        check_filter_flux_consistency(df)
+
+
+def test_check_filter_flux_consistency_allows_missing_filter_and_flux():
+    df = pd.DataFrame({"filter_g": [None, "g_sdss"], "psf_flux_g": [np.nan, 1.0]})
+
+    # should not raise
+    check_filter_flux_consistency(df)