Fix sso duplication issue and allow for bulk reading of sso file

pyproject.toml

@@ -22,11 +22,13 @@ dependencies = [
     "pydantic_settings",
     "asyncer",
     "astropy",
+    "numpy",
     "scipy",
     "astroquery",
     "astropydantic",
     "pandas",
     "pyarrow",
+    "tqdm",
     "uuid7-standard",
 ]
 
@@ -48,6 +50,7 @@ socat-act-fits = "socat.ingest.actfits:main"
 socat-websky-csv = "socat.ingest.webskycsv:main"
 socat-migrate = "socat.migrate:migrate"
 socat-jpl-parqet = "socat.ingest.jplparquet:main"
+socat-build-db = "socat.ingest.bulk_sqlite:main"
 
 [tool.ruff.lint]
 preview = true

socat/alembic/versions/1bfc5561403e_add_sso_id_time_composite_index_to_.py

@@ -0,0 +1,38 @@
+"""Add sso_id/time composite index to moving_sources
+
+Revision ID: 1bfc5561403e
+Revises: c3d4e5f6a7b8
+Create Date: 2026-07-17 00:00:00.000000
+
+get_ephem_points()/get_source_generator() filter moving_sources by a known
+sso_id first, then narrow by time. moving_sources.time already has its own
+index (added by 35a6a33e0a34's initial op.create_table), but sso_id was
+never indexed -- at real ephemeris-table scale (tens of millions of rows,
+see simonsobs/sotrplib's scripts/solar_system/build_socat_db.py), that made
+every per-object ephemeris lookup a full table scan. This mirrors the
+RegisteredMovingSourceTable.__table_args__ index added to the ORM model in
+the same change, so a database provisioned via `alembic upgrade head` ends
+up with the same indexes as one provisioned via SQLModel.metadata.create_all().
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "1bfc5561403e"
+down_revision: str | None = "c3d4e5f6a7b8"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "idx_moving_sources_sso_time",
+        "moving_sources",
+        ["sso_id", "time"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_moving_sources_sso_time", table_name="moving_sources")

socat/database/sources.py

@@ -6,6 +6,7 @@
 from astropy.time import Time
 from astropydantic import AstroPydanticICRS, AstroPydanticQuantity, AstroPydanticTime
 from pydantic import BaseModel
+from sqlalchemy import Index
 from sqlmodel import Field, SQLModel
 
 
@@ -190,6 +191,13 @@ class RegisteredMovingSourceTable(SQLModel, table=True):
     """
 
     __tablename__ = "moving_sources"
+    __table_args__ = (
+        # get_ephem_points()/get_source_generator() filter by a known
+        # sso_id first, then narrow by time -- this composite index
+        # serves both that and plain sso_id-only lookups (leftmost
+        # prefix), so no separate single-column sso_id index is needed.
+        Index("idx_moving_sources_sso_time", "sso_id", "time"),
+    )
 
     ephem_id: uuid.UUID = Field(primary_key=True, default_factory=uuid.create)
     sso_id: uuid.UUID = Field(
@@ -207,7 +215,11 @@ class RegisteredMovingSourceTable(SQLModel, table=True):
         nullable=False,
         ondelete="CASCADE",
     )
-    time: datetime
+    # get_box_sso()/get_monitored_ssos()/get_pointing_ssos() have no
+    # sso_id to filter on (that's what they're looking up) -- they filter
+    # by time directly, so it needs its own leading index too, not just
+    # as the second column of the composite above.
+    time: datetime = Field(index=True, nullable=False)
     ra_deg: float = Field(nullable=False)
     dec_deg: float = Field(nullable=False)
     flux_mJy: float | None = Field(nullable=True)

socat/database/statements.py

@@ -11,7 +11,7 @@
 from astropy.time import Time
 from astropy.units import Quantity
 from astroquery.query import BaseVOQuery
-from sqlmodel import select, union_all, update
+from sqlmodel import select, union, union_all, update
 
 from socat.database.services import AstroqueryServiceTable
 from socat.database.sources import (
@@ -146,6 +146,15 @@ def get_box_sso(
         Database statement.
     """
     if lower_left.ra > upper_right.ra:
+        # Each branch joins against every matching ephemeris point, so
+        # without its own .distinct() a single object with N matching
+        # ephem points produces N duplicate rows here -- and the
+        # .distinct() on the outer `select(...).from_statement(union_stmt)`
+        # below does NOT fix that: from_statement() replaces the executed
+        # SQL wholesale, so that outer .distinct() is a no-op. union_all
+        # (rather than union) is also insufficient on its own: an object
+        # can legitimately have ephem points on both sides of RA=0 within
+        # the same time window, and union_all wouldn't merge those.
         right_box = (
             select(SolarSystemObjectTable)
             .outerjoin(
@@ -163,6 +172,7 @@ def get_box_sso(
                 RegisteredMovingSourceTable.dec_deg
                 <= float(upper_right.dec.to_value("deg")),
             )
+            .distinct()
         )
         left_box = (
             select(SolarSystemObjectTable)
@@ -181,9 +191,10 @@ def get_box_sso(
                 RegisteredMovingSourceTable.dec_deg
                 <= float(upper_right.dec.to_value("deg")),
             )
+            .distinct()
         )
-        union_stmt = union_all(right_box, left_box)
-        return select(SolarSystemObjectTable).distinct().from_statement(union_stmt)
+        union_stmt = union(right_box, left_box)
+        return select(SolarSystemObjectTable).from_statement(union_stmt)
     else:
         return (
             select(SolarSystemObjectTable)