feat(conflation): use separate mapping_sessions_refs table

Author: ofr1tz

mapswipe_workers/mapswipe_workers/firebase_to_postgres/transfer_results.py

@@ -432,6 +432,8 @@ def save_results_to_postgres(
 
     query_insert_mapping_sessions = f"""
         BEGIN;
+
+        -- Create or ensure mapping_sessions exist
         INSERT INTO mapping_sessions
             SELECT
                 project_id,
@@ -445,21 +447,36 @@ def save_results_to_postgres(
                 client_type
             FROM {result_temp_table}
             GROUP BY project_id, group_id, user_id, app_version, client_type
-        ON CONFLICT (project_id,group_id,user_id)
+        ON CONFLICT (project_id, group_id, user_id)
+        DO NOTHING;
+
+        INSERT INTO {result_table} (mapping_session_id, task_id, result)
+            SELECT
+                ms.mapping_session_id,
+                r.task_id,
+                {result_sql}
+            FROM {result_temp_table} r
+            JOIN mapping_sessions ms ON
+                ms.project_id = r.project_id
+                AND ms.group_id = r.group_id
+                AND ms.user_id = r.user_id
+        ON CONFLICT (mapping_session_id, task_id)
         DO NOTHING;
-        INSERT INTO {result_table}
+
+        INSERT INTO mapping_sessions_refs (mapping_session_id, task_id, ref)
             SELECT
                 ms.mapping_session_id,
                 r.task_id,
-                {result_sql},
                 r.ref
             FROM {result_temp_table} r
             JOIN mapping_sessions ms ON
                 ms.project_id = r.project_id
                 AND ms.group_id = r.group_id
                 AND ms.user_id = r.user_id
+            WHERE r.ref IS NOT NULL
         ON CONFLICT (mapping_session_id, task_id)
         DO NOTHING;
+
         COMMIT;
     """
     p_con.query(query_insert_mapping_sessions)

mapswipe_workers/mapswipe_workers/generate_stats/project_stats.py

@@ -110,7 +110,7 @@ def get_results(
     if result_table == "mapping_sessions_results_geometry":
         result_sql = "ST_AsGeoJSON(msr.result) as result"
     else:
-        result_sql = "msr.result as result, msr.ref as ref"
+        result_sql = "msr.result as result"
 
     sql_query = sql.SQL(
         f"""
@@ -126,6 +126,7 @@ def get_results(
                 ms.app_version,
                 ms.client_type,
                 {result_sql},
+                refs.ref as ref,
                 -- the username for users which login to MapSwipe with their
                 -- OSM account is not defined or ''.
                 -- We capture this here as it will cause problems
@@ -138,7 +139,10 @@ def get_results(
             LEFT JOIN mapping_sessions ms ON
                 ms.mapping_session_id = msr.mapping_session_id
             LEFT JOIN users U USING (user_id)
-            WHERE project_id = {"{}"}
+            LEFT JOIN mapping_sessions_refs refs
+                ON msr.mapping_session_id = refs.mapping_session_id
+                AND msr.task_id = refs.task_id
+            WHERE ms.project_id = {"{}"}
         ) TO STDOUT WITH CSV HEADER
         """
     ).format(sql.Literal(project_id))
@@ -513,24 +517,23 @@ def add_ref_to_agg_results(
     results_df: pd.DataFrame, agg_results_df: pd.DataFrame
 ) -> pd.DataFrame:
     """
-    Adds a 'ref' column to agg_results_df for writing to CSV
+    Adds a 'ref' column to agg_results_df if it exists in results_df.
+    For each task_id, all unique non-empty refs are collected into a list.
+    If no refs exist for a task, the corresponding value is empty string.
+    If results_df has no 'ref' column, agg_results_df is returned unchanged.
     """
+    if "ref" not in results_df.columns:
+        return agg_results_df
 
-    refs_per_task = results_df.groupby("task_id")["ref"].apply(list)
+    refs_per_task = (
+        results_df.groupby("task_id")["ref"]
+        .apply(lambda x: list({r for r in x if pd.notna(r) and r not in ({}, "")}))
+        .apply(lambda lst: json.dumps([json.loads(r) for r in lst]) if lst else "")
+    )
 
-    ref_values = {}
-    for task_id, refs in refs_per_task.items():
-        # Filter out None or empty dicts
-        refs = [r for r in refs if r not in (None, {}, "") and not pd.isna(r)]
-        if not refs:
-            continue
-        elif all(r == refs[0] for r in refs):
-            ref_values[task_id] = refs[0]
-        else:
-            ref_values[task_id] = refs
+    if refs_per_task.apply(lambda x: len(x) > 0).any():
+        agg_results_df["ref"] = agg_results_df["task_id"].map(refs_per_task).fillna("")
 
-    if ref_values:
-        agg_results_df["ref"] = agg_results_df["task_id"].map(ref_values).fillna("")
     return agg_results_df
 
 

mapswipe_workers/tests/integration/set_up_db.sql

@@ -193,7 +193,6 @@ CREATE TABLE IF NOT EXISTS mapping_sessions_results (
     mapping_session_id int8,
     task_id varchar,
     result int2 not null,
-    ref jsonb,
     PRIMARY KEY (mapping_session_id, task_id),
     FOREIGN KEY (mapping_session_id)
     references mapping_sessions (mapping_session_id)
@@ -208,6 +207,15 @@ CREATE TABLE IF NOT EXISTS mapping_sessions_results_geometry (
     references mapping_sessions (mapping_session_id)
 );
 
+CREATE TABLE IF NOT EXISTS mapping_sessions_refs (
+    mapping_session_id int8,
+    task_id varchar,
+    ref JSONB not null,
+    PRIMARY KEY (mapping_session_id, task_id),
+    FOREIGN KEY (mapping_session_id)
+    references mapping_sessions (mapping_session_id)
+);
+
 CREATE OR REPLACE FUNCTION mapping_sessions_results_constraint() RETURNS trigger
     LANGUAGE plpgsql AS
 $$

mapswipe_workers/tests/unittests/test_project_stats.py

@@ -1,10 +1,11 @@
+import json
 import unittest
 
 import pandas as pd
 
 from mapswipe_workers.generate_stats.project_stats import (
-    add_ref_to_agg_results,
     add_missing_result_columns,
+    add_ref_to_agg_results,
     calc_agreement,
     calc_count,
     calc_parent_option_count,
@@ -175,28 +176,46 @@ def test_calc_parent_option_count(self):
 
     def test_add_ref_single_ref(self):
         # All results have the same ref
-        results_df = pd.DataFrame({
-            "task_id": ["t1", "t1"],
-            "ref": [{"osmId": 123, "osmType": "ways_poly"}, {"osmId": 123, "osmType": "ways_poly"}]
-        })
+        results_df = pd.DataFrame(
+            {
+                "task_id": ["t1", "t1"],
+                "ref": [
+                    json.dumps({"osmId": 123, "osmType": "ways_poly"}),
+                    json.dumps({"osmId": 123, "osmType": "ways_poly"}),
+                ],
+            }
+        )
         agg_results_df = pd.DataFrame({"task_id": ["t1"]})
         updated_df = add_ref_to_agg_results(results_df, agg_results_df.copy())
+
         self.assertIn("ref", updated_df.columns)
-        self.assertEqual(updated_df["ref"].iloc[0], {"osmId": 123, "osmType": "ways_poly"})
+        ref_value = json.loads(updated_df["ref"].iloc[0])
+        self.assertEqual(ref_value, [{"osmId": 123, "osmType": "ways_poly"}])
 
     def test_add_ref_multiple_refs(self):
         # Different refs for same task
-        results_df = pd.DataFrame({
-            "task_id": ["t1", "t1"],
-            "ref": [{"osmId": 123}, {"osmId": 456}]
-        })
+        results_df = pd.DataFrame(
+            {
+                "task_id": ["t1", "t1"],
+                "ref": [json.dumps({"osmId": 123}), json.dumps({"osmId": 456})],
+            }
+        )
         agg_results_df = pd.DataFrame({"task_id": ["t1"]})
         updated_df = add_ref_to_agg_results(results_df, agg_results_df.copy())
+
         self.assertIn("ref", updated_df.columns)
-        self.assertEqual(
-            updated_df["ref"].iloc[0],
-            [{"osmId": 123}, {"osmId": 456}]
-        )
+        ref_value = json.loads(updated_df["ref"].iloc[0])
+        self.assertCountEqual(ref_value, [{"osmId": 123}, {"osmId": 456}])
+
+    def test_add_ref_no_refs_column(self):
+        # results_df has no 'ref' column
+        results_df = pd.DataFrame({"task_id": ["t1", "t2"], "result": [1, 2]})
+        agg_results_df = pd.DataFrame({"task_id": ["t1", "t2"]})
+
+        updated_df = add_ref_to_agg_results(results_df, agg_results_df.copy())
+
+        self.assertNotIn("ref", updated_df.columns)
+        pd.testing.assert_frame_equal(updated_df, agg_results_df)
 
 
 if __name__ == "__main__":

postgres/initdb.sql

@@ -193,7 +193,6 @@ CREATE TABLE IF NOT EXISTS mapping_sessions_results (
     mapping_session_id int8,
     task_id varchar,
     result int2 not null,
-    ref jsonb,
     PRIMARY KEY (mapping_session_id, task_id),
     FOREIGN KEY (mapping_session_id)
     references mapping_sessions (mapping_session_id)
@@ -208,6 +207,15 @@ CREATE TABLE IF NOT EXISTS mapping_sessions_results_geometry (
     references mapping_sessions (mapping_session_id)
 );
 
+CREATE TABLE IF NOT EXISTS mapping_sessions_refs (
+    mapping_session_id int8,
+    task_id varchar,
+    ref JSONB not null,
+    PRIMARY KEY (mapping_session_id, task_id),
+    FOREIGN KEY (mapping_session_id)
+    references mapping_sessions (mapping_session_id)
+);
+
 CREATE OR REPLACE FUNCTION mapping_sessions_results_constraint() RETURNS trigger
     LANGUAGE plpgsql AS
 $$

postgres/scripts/add_ref_to_results_for_conflation.sql

@@ -1,5 +1,11 @@
-ALTER TABLE mapping_sessions_results
-ADD COLUMN ref jsonb;
-
 ALTER TABLE results_temp
 ADD COLUMN ref jsonb;
+
+CREATE TABLE IF NOT EXISTS public.mapping_sessions_refs (
+    mapping_session_id int8,
+    task_id varchar,
+    ref JSONB not null,
+    PRIMARY KEY (mapping_session_id, task_id),
+    FOREIGN KEY (mapping_session_id)
+    references mapping_sessions (mapping_session_id)
+);