diff --git a/delphi/polismath/database/postgres.py b/delphi/polismath/database/postgres.py index 0d47379a1..d4744ed0e 100644 --- a/delphi/polismath/database/postgres.py +++ b/delphi/polismath/database/postgres.py @@ -25,6 +25,7 @@ import pandas as pd from polismath.utils.general import postgres_vote_to_delphi +from polismath.utils.serialization import convert_numpy_types # Set up logging logger = logging.getLogger(__name__) @@ -214,6 +215,23 @@ def __repr__(self): return f"" +class MathBidToPid(Base): + """Stores the base-cluster bid -> participant-id mapping (server consumes it + via server/src/utils/participants.ts). Mirrors the Clojure math_bidtopid + table written by upload-math-bidtopid (postgres.clj:369-380).""" + + __tablename__ = "math_bidtopid" + + zid = sa.Column(sa.Integer, primary_key=True) + math_env = sa.Column(sa.String, primary_key=True) + math_tick = sa.Column(sa.BigInteger, nullable=False, default=-1) + data = sa.Column(JSONB, nullable=False) + modified = sa.Column(sa.BigInteger, server_default=text("now_as_millis()")) + + def __repr__(self): + return f"" + + class MathReportCorrelationMatrix(Base): """Stores correlation matrices for reports.""" @@ -392,10 +410,31 @@ def execute(self, sql: str, params: Optional[Dict[str, Any]] = None) -> int: if not self._initialized: self.initialize() - with self.engine.connect() as conn: + with self.engine.begin() as conn: result = conn.execute(text(sql), params or {}) return result.rowcount + def _write_returning( + self, sql: str, params: Optional[Dict[str, Any]] = None + ) -> List[Dict[str, Any]]: + """Execute a writing statement inside a COMMITTED transaction and return + any RETURNING rows. + + ``query()`` uses ``engine.connect()`` (SQLAlchemy 2.0 "commit as you go"), + which rolls back on close — fine for SELECTs but it silently discards + INSERT/UPDATEs. The upsert writers (math_main / math_ticks / math_bidtopid + / math_ptptstats) MUST persist, so they route through here: + ``engine.begin()`` commits on successful exit. + """ + if not self._initialized: + self.initialize() + + with self.engine.begin() as conn: + result = conn.execute(text(sql), params or {}) + if result.returns_rows: + return [dict(row) for row in result.mappings().all()] + return [] + def get_zinvite_from_zid(self, zid: int) -> Optional[str]: """ Get the zinvite (conversation code) for a conversation ID. @@ -466,10 +505,16 @@ def poll_votes( """ # Add timestamp filter if provided - if since: + if since is not None: sql += " AND created > :since" params["since"] = since + # Row order matters for parity: Clojure conv-poll orders by + # [:zid :tid :pid :created] (postgres.clj:197-212). update_votes assigns + # base-cluster IDs by first-appearance order of participants, which seeds + # k-means; a different row order changes k. So we must ORDER identically. + sql += " ORDER BY zid, tid, pid, created" + # Execute query votes = self.query(sql, params) @@ -484,6 +529,78 @@ def poll_votes( for v in votes ] + def poll_votes_since(self, since: int) -> List[Dict[str, Any]]: + """ + Global vote poll across ALL conversations since a watermark. + + Mirrors the Clojure vote poller query (postgres.clj:132-145): + SELECT * FROM votes WHERE created > watermark + ORDER BY zid, tid, pid, created + Signs are flipped to the Delphi convention at this ingress boundary. + + Args: + since: Watermark (millis since epoch); returns rows with created > since + + Returns: + List of votes {zid, pid, tid, vote, created}, sign-flipped, ordered. + """ + rows = self.query( + """ + SELECT zid, tid, pid, vote, created + FROM votes + WHERE created > :since + ORDER BY zid, tid, pid, created + """, + {"since": since}, + ) + return [ + { + "zid": int(v["zid"]), + "pid": str(v["pid"]), + "tid": str(v["tid"]), + "vote": postgres_vote_to_delphi(int(v["vote"])), + "created": v["created"], + } + for v in rows + ] + + def poll_moderation_since(self, since: int) -> List[Dict[str, Any]]: + """ + Global moderation poll across ALL conversations since a watermark. + + Mirrors the Clojure mod poller query (postgres.clj:148-161): + SELECT * FROM comments WHERE modified > watermark + ORDER BY zid, tid, modified + Returns the raw changed-comment rows so the caller can group by zid and + advance the watermark to max(modified). The per-zid worker then + re-derives the FULL current moderation state via poll_moderation(zid). + + Args: + since: Watermark (millis since epoch); rows with modified > since + + Returns: + List of {zid, tid, modified, mod, is_meta}. + """ + rows = self.query( + """ + SELECT zid, tid, modified, mod, is_meta + FROM comments + WHERE modified > :since + ORDER BY zid, tid, modified + """, + {"since": since}, + ) + return [ + { + "zid": int(m["zid"]), + "tid": int(m["tid"]), + "modified": m["modified"], + "mod": m["mod"], + "is_meta": m["is_meta"], + } + for m in rows + ] + def get_report_comment_selections( self, zid: int, rid: Optional[int] = None ) -> List[Dict[str, Any]]: @@ -643,69 +760,120 @@ def write_math_main( math_tick: Optional[int] = None, ) -> None: """ - Write math results for a conversation. + Write math results for a conversation (Clojure upload-math-main parity). + + caching_tick is NEVER taken from the caller: it is derived in-SQL exactly + as Clojure does (postgres.clj:323-338): + + caching_tick = COALESCE( + (SELECT max(caching_tick) + 1 FROM math_main WHERE math_env = ?), + 1) + + so the TS server's prefetch (pca.ts:84-151 polls caching_tick > last) sees + a strictly increasing, per-math_env cursor. The `caching_tick` parameter + is accepted for signature compatibility but ignored. Args: zid: Conversation ID - data: Math data + data: Math data (JSON blob stored verbatim) last_vote_timestamp: Timestamp of last processed vote - caching_tick: Current caching tick - math_tick: Current math tick + caching_tick: Ignored (derived in SQL); kept for back-compat + math_tick: Current math tick (shared with the other writes this cycle) """ - with self.session() as session: - # Check if record exists - math_main = ( - session.query(MathMain) - .filter_by(zid=zid, math_env=self.config.math_env) - .first() - ) - - if math_main: - # Update existing record - math_main.data = data - if last_vote_timestamp is not None: - math_main.last_vote_timestamp = last_vote_timestamp - if caching_tick is not None: - math_main.caching_tick = caching_tick - if math_tick is not None: - math_main.math_tick = math_tick - else: - # Create new record - math_main = MathMain( - zid=zid, - math_env=self.config.math_env, - data=data, - last_vote_timestamp=last_vote_timestamp or int(time.time() * 1000), - caching_tick=caching_tick or 0, - math_tick=math_tick or -1, - ) - session.add(math_main) + last_vote_timestamp = ( + last_vote_timestamp + if last_vote_timestamp is not None + else int(time.time() * 1000) + ) + # NOTE: math_env appears twice in the params — once for the row value and + # once inside the caching_tick subquery (mirrors Clojure's duplicated ?). + self._write_returning( + """ + insert into math_main + (zid, math_env, last_vote_timestamp, math_tick, data, caching_tick) + values + (:zid, :math_env, :last_vote_timestamp, :math_tick, + cast(:data as jsonb), + COALESCE((select max(caching_tick) + 1 from math_main + where math_env = :math_env), 1)) + on conflict (zid, math_env) + do update set modified = now_as_millis(), + data = excluded.data, + last_vote_timestamp = excluded.last_vote_timestamp, + math_tick = excluded.math_tick, + caching_tick = excluded.caching_tick + returning zid; + """, + { + "zid": zid, + "math_env": self.config.math_env, + "last_vote_timestamp": last_vote_timestamp, + "math_tick": math_tick if math_tick is not None else -1, + "data": json.dumps(data, default=convert_numpy_types), + }, + ) - def write_participant_stats(self, zid: int, data: Dict[str, Any]) -> None: + def write_math_bidtopid( + self, zid: int, data: Dict[str, Any], math_tick: Optional[int] = None + ) -> None: """ - Write participant statistics for a conversation. + Write the bid -> participant-id mapping (Clojure upload-math-bidtopid, + postgres.clj:369-380). Net-new writer: the TS server's + getBidIndexToPidMapping / getPidsForGid (participants.ts) depend on it. Args: zid: Conversation ID - data: Participant statistics data + data: prep-bidToPid blob {"zid", "bidToPid", "lastVoteTimestamp"} + math_tick: Current math tick (shared with the other writes this cycle) """ - with self.session() as session: - # Check if record exists - ptpt_stats = ( - session.query(MathPtptStats) - .filter_by(zid=zid, math_env=self.config.math_env) - .first() - ) + self._write_returning( + """ + insert into math_bidtopid (zid, math_env, math_tick, data) + values (:zid, :math_env, :math_tick, cast(:data as jsonb)) + on conflict (zid, math_env) + do update set modified = now_as_millis(), + data = excluded.data, + math_tick = excluded.math_tick + returning zid; + """, + { + "zid": zid, + "math_env": self.config.math_env, + "math_tick": math_tick if math_tick is not None else -1, + "data": json.dumps(data, default=convert_numpy_types), + }, + ) - if ptpt_stats: - # Update existing record - ptpt_stats.data = data - else: - # Create new record - ptpt_stats = MathPtptStats( - zid=zid, math_env=self.config.math_env, data=data - ) - session.add(ptpt_stats) + def write_participant_stats( + self, zid: int, data: Dict[str, Any], math_tick: Optional[int] = None + ) -> None: + """ + Write participant statistics (Clojure upload-math-ptptstats parity, + postgres.clj:350-361). Writes math_tick so the three data tables share + the single tick minted for the cycle (conv_man.clj:158-169). + + Args: + zid: Conversation ID + data: Participant statistics data (prep-ptpt-stats blob) + math_tick: Current math tick (shared with the other writes this cycle) + """ + self._write_returning( + """ + insert into math_ptptstats (zid, math_env, math_tick, data) + values (:zid, :math_env, :math_tick, cast(:data as jsonb)) + on conflict (zid, math_env) + do update set modified = now_as_millis(), + data = excluded.data, + math_tick = excluded.math_tick + returning zid; + """, + { + "zid": zid, + "math_env": self.config.math_env, + "math_tick": math_tick if math_tick is not None else -1, + "data": json.dumps(data, default=convert_numpy_types), + }, + ) def write_correlation_matrix(self, rid: int, data: Dict[str, Any]) -> None: """ @@ -738,7 +906,16 @@ def write_correlation_matrix(self, rid: int, data: Dict[str, Any]) -> None: def increment_math_tick(self, zid: int) -> int: """ - Increment the math tick counter for a conversation. + Atomically increment the math tick counter for a conversation. + + Clojure inc-math-tick (postgres.clj:292-295) does this in a SINGLE + statement so concurrent writers never race a read-modify-write: + + insert into math_ticks (zid, math_env) values (?, ?) + on conflict (zid, math_env) + do update set modified = now_as_millis(), + math_tick = (math_ticks.math_tick + 1) + returning math_tick; Args: zid: Conversation ID @@ -746,29 +923,17 @@ def increment_math_tick(self, zid: int) -> int: Returns: New tick value """ - with self.session() as session: - # Check if record exists - math_ticks = ( - session.query(MathTicks) - .filter_by(zid=zid, math_env=self.config.math_env) - .first() - ) - - if math_ticks: - # Update existing record - math_ticks.math_tick += 1 - new_math_tick = math_ticks.math_tick - else: - # Create new record - math_ticks = MathTicks( - zid=zid, math_env=self.config.math_env, math_tick=1 - ) - session.add(math_ticks) - new_math_tick = 1 - - # Commit and return new math tick - session.commit() - return new_math_tick + rows = self._write_returning( + """ + insert into math_ticks (zid, math_env) values (:zid, :math_env) + on conflict (zid, math_env) + do update set modified = now_as_millis(), + math_tick = (math_ticks.math_tick + 1) + returning math_tick; + """, + {"zid": zid, "math_env": self.config.math_env}, + ) + return rows[0]["math_tick"] def poll_tasks( self, task_type: str, last_timestamp: int = 0, limit: int = 10 diff --git a/delphi/polismath/regression/utils.py b/delphi/polismath/regression/utils.py index f65273ab9..320cfafa0 100644 --- a/delphi/polismath/regression/utils.py +++ b/delphi/polismath/regression/utils.py @@ -16,6 +16,10 @@ import pandas as pd from polismath.conversation.conversation import Conversation +# Backward-compatible re-export: convert_numpy_types was defined here (nested in +# save_golden_snapshot); it now lives in the shared serialization util so the +# Postgres math writers can share it. Import keeps existing references working. +from polismath.utils.serialization import convert_numpy_types # noqa: F401 # Set up logger logger = logging.getLogger(__name__) @@ -344,17 +348,7 @@ def save_golden_snapshot(snapshot: Dict, golden_path: Path) -> None: # Ensure parent directory exists golden_path.parent.mkdir(parents=True, exist_ok=True) - # Custom JSON encoder that converts numpy types to Python native types - def convert_numpy_types(obj): - """Convert numpy types to Python native types for JSON serialization.""" - import numpy as np - if isinstance(obj, np.integer): - return int(obj) - elif isinstance(obj, np.floating): - return float(obj) - elif isinstance(obj, np.ndarray): - return obj.tolist() - raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") - + # convert_numpy_types is hoisted to polismath.utils.serialization (and + # re-exported below for backward compatibility). with open(golden_path, 'w') as f: json.dump(snapshot, f, indent=2, default=convert_numpy_types) \ No newline at end of file diff --git a/delphi/polismath/utils/serialization.py b/delphi/polismath/utils/serialization.py new file mode 100644 index 000000000..6a035005e --- /dev/null +++ b/delphi/polismath/utils/serialization.py @@ -0,0 +1,29 @@ +"""JSON serialization helpers shared across the math pipeline. + +``convert_numpy_types`` is the canonical ``default=`` for ``json.dumps`` when a +blob may carry numpy scalar/array types. It lives here (rather than nested inside +``regression.utils.save_golden_snapshot``) so the Postgres math writers +(``write_math_main`` / ``write_math_bidtopid`` / ``write_participant_stats``) can +share the exact same coercion. +""" + +import numpy as np + + +def convert_numpy_types(obj): + """Convert numpy scalar/array types to JSON-native Python types. + + Use as the ``default=`` callback for ``json.dumps``. Without it, a blob that + carries a numpy integer — e.g. the repness ``gid`` (repness.py:847 + ``astype(int)`` produces a numpy ``int64``) or the na/nd/ns counts + (repness.py:672-675) — raises ``TypeError: Object of type int64 is not JSON + serializable``. Note ``json`` already handles ``np.float64`` (a subclass of + Python ``float``) but NOT ``np.int64``, so integral fields are the trap. + """ + if isinstance(obj, np.integer): + return int(obj) + elif isinstance(obj, np.floating): + return float(obj) + elif isinstance(obj, np.ndarray): + return obj.tolist() + raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") diff --git a/delphi/tests/test_math_writer_numpy_serialization.py b/delphi/tests/test_math_writer_numpy_serialization.py new file mode 100644 index 000000000..ed1ed506f --- /dev/null +++ b/delphi/tests/test_math_writer_numpy_serialization.py @@ -0,0 +1,112 @@ +"""T3: the three Postgres math writers must serialize numpy scalar types. + +`write_math_main` / `write_math_bidtopid` / `write_participant_stats` +(postgres.py) serialize their blob with `json.dumps`. A real math blob can carry +numpy scalars — the repness `gid` is `astype(int)` => numpy `int64` +(repness.py:847), and the na/nd/ns counts are `astype(int)` sums (:672-675). +`json.dumps` handles `np.float64` (a subclass of `float`) but NOT `np.int64`, so +any such blob raised `TypeError: Object of type int64 is not JSON serializable` +and rolled the WHOLE write cycle back. The writers now pass +`default=convert_numpy_types`. + +Note: pandas 2.x's `DataFrame.to_dict('records')` down-converts numpy scalars to +Python natives, which masks the raw reproduction on the committed test datasets. +But numpy scalars DO reach the serialization boundary in production — see the +boto3 "Float types are not supported" note at repness.py:843. So this test takes +a REAL, computed, 2-group `to_dict()` blob (not a toy dict, which is what the +earlier writer unit tests used) and faithfully reintroduces the numpy integer +type at every integral leaf, which is environment-independent. +""" + +import json + +import numpy as np +import pytest + +from polismath.conversation.conversation import Conversation +from polismath.database.postgres import PostgresClient, PostgresConfig +from polismath.poller.math_writer import derive_bidtopid, derive_ptptstats + + +def _two_group_conv(): + """A synthetic conversation with two opposing camps -> 2 groups + repness.""" + votes = [] + n_per, n_cmts = 8, 8 + for p in range(n_per * 2): + camp = 0 if p < n_per else 1 + for t in range(n_cmts): + v = (1.0 if t % 2 == 0 else -1.0) if camp == 0 else (-1.0 if t % 2 == 0 else 1.0) + votes.append({"pid": f"p{p}", "tid": f"c{t}", "vote": v}) + return Conversation("t3").update_votes({"votes": votes}) + + +def _numpy_ints(o): + """Deep-copy a JSON-ish structure, casting every integral leaf to np.int64 + (leaving bools/floats/strings alone). Faithfully simulates the numpy scalars + that repness.py:847/:672-675 emit and that older pandas / boto3 paths keep.""" + if isinstance(o, bool): + return o + if isinstance(o, (int, np.integer)): + return np.int64(o) + if isinstance(o, dict): + return {k: _numpy_ints(v) for k, v in o.items()} + if isinstance(o, list): + return [_numpy_ints(v) for v in o] + return o + + +def _client_capturing(): + """A PostgresClient whose _write_returning is stubbed to capture the params + (so we exercise the real json.dumps in the writer without a live DB).""" + client = PostgresClient(PostgresConfig(url="postgresql://ignored/db", math_env="t3")) + captured = {} + + def _fake_write_returning(sql, params=None): + captured["params"] = params + return [] + + client._write_returning = _fake_write_returning + return client, captured + + +class TestWritersSerializeNumpy: + def test_math_main_round_trips_real_blob_with_numpy(self): + conv = _two_group_conv() + blob = conv.to_dict() + # Real repness records are present (writer's fidelity-critical input). + recs = blob["repness"]["comment_repness"] + assert recs and "gid" in recs[0] + + blob_np = _numpy_ints(blob) + # The gid is now the numpy int64 that repness.py:847 astype(int) produces. + assert any(isinstance(r["gid"], np.integer) + for r in blob_np["repness"]["comment_repness"]) + + # RED precondition (environment-independent): bare json.dumps rejects it. + with pytest.raises(TypeError, match="int64 is not JSON serializable"): + json.dumps(blob_np) + + # The writer serializes and the blob round-trips (gid back as JSON int). + client, cap = _client_capturing() + client.write_math_main(1, blob_np, last_vote_timestamp=123, math_tick=0) + restored = json.loads(cap["params"]["data"]) + got = restored["repness"]["comment_repness"][0]["gid"] + assert isinstance(got, int) and not isinstance(got, bool) + + def test_bidtopid_round_trips_real_blob_with_numpy(self): + conv = _two_group_conv() + blob = _numpy_ints(derive_bidtopid(conv, 1)) + with pytest.raises(TypeError, match="int64 is not JSON serializable"): + json.dumps(blob) + client, cap = _client_capturing() + client.write_math_bidtopid(1, blob, math_tick=0) + assert json.loads(cap["params"]["data"])["zid"] == 1 + + def test_ptptstats_round_trips_real_blob_with_numpy(self): + conv = _two_group_conv() + blob = _numpy_ints(derive_ptptstats(conv, 1)) + with pytest.raises(TypeError, match="int64 is not JSON serializable"): + json.dumps(blob) + client, cap = _client_capturing() + client.write_participant_stats(1, blob, math_tick=0) + assert json.loads(cap["params"]["data"])["zid"] == 1