DM-46631: fix replace=True on dimensions (etc) with only primary key columns

doc/changes/DM-46631.bugfix.md

@@ -0,0 +1 @@
+Fix inserts with `replace=True` on dimensions with only primary key columns.

python/lsst/daf/butler/registry/databases/postgresql.py

@@ -351,6 +351,9 @@ def replace(self, table: sqlalchemy.schema.Table, *rows: dict) -> None:
             for column in table.columns
             if column.name not in table.primary_key
         }
+        if not data:
+            self.ensure(table, *rows)
+            return
         query = query.on_conflict_do_update(constraint=table.primary_key, set_=data)
         with self._transaction() as (_, connection):
             connection.execute(query, rows)

python/lsst/daf/butler/registry/databases/sqlite.py

@@ -355,6 +355,9 @@ def replace(self, table: sqlalchemy.schema.Table, *rows: dict) -> None:
             for column in table.columns
             if column.name not in table.primary_key
         }
+        if not data:
+            self.ensure(table, *rows)
+            return
         query = query.on_conflict_do_update(index_elements=table.primary_key, set_=data)
         with self._transaction() as (_, connection):
             connection.execute(query, rows)

python/lsst/daf/butler/registry/tests/_database.py

@@ -685,6 +685,24 @@ def testReplace(self):
             [r._asdict() for r in self.query_list(db, tables.a.select())], [row1, row2a, row3]
         )
 
+    def test_replace_pkey_only(self):
+        """Test `Database.replace` on a table that only has primary key"""
+        spec = ddl.TableSpec(
+            [
+                ddl.FieldSpec("a1", dtype=sqlalchemy.BigInteger, primaryKey=True),
+                ddl.FieldSpec("a2", dtype=sqlalchemy.BigInteger, primaryKey=True),
+            ]
+        )
+        db = self.makeEmptyDatabase(origin=1)
+        with db.declareStaticTables(create=True) as context:
+            table = context.addTable("a", spec)
+        row1 = {"a1": 1, "a2": 2}
+        row2 = {"a1": 1, "a2": 3}
+        db.replace(table, row1)
+        db.replace(table, row2)
+        db.replace(table, row1)
+        self.assertCountEqual([r._asdict() for r in self.query_list(db, table.select())], [row1, row2])
+
     def testEnsure(self):
         """Tests for `Database.ensure`."""
         db = self.makeEmptyDatabase(origin=1)