Add edit-journal / speed up row inserts using current revision system

api/actions.py

@@ -1316,14 +1316,21 @@ def fetchmany(request: dict, context):
     return cursor.fetchmany(int(request["size"]))
 
 
-def apply_changes(table_obj: Table, cursor: AbstractCursor | None = None):
+def apply_changes(
+    table_obj: Table,
+    cursor: AbstractCursor | None = None,
+    change_types: tuple = ("insert", "update", "delete"),
+):
     """Apply changes from the meta tables to the actual table.
 
     Meta tables are :
     * _<NAME>_insert
     * _<NAME>_update
     * _<NAME>_delete
 
+    Only the meta tables named in `change_types` are scanned, so callers that
+    know which operation they just performed (e.g. a pure insert) don't pay
+    for scanning the other meta tables.
     """
 
     def add_type(d, type):
@@ -1347,64 +1354,34 @@ def add_type(d, type):
         # so we pass user=None
         oedb_table = table_obj.get_oedb_table_proxy(user=None)
 
-        insert_sa_table = oedb_table._insert_table.get_sa_table()
-        _execute(
-            cursor,
-            'select * from "{schema}"."{table}" where _applied = FALSE;'.format(
-                schema=insert_sa_table.schema, table=insert_sa_table.name
-            ),
-        )
-        changes = [
-            add_type(
-                {
-                    c.name: v
-                    for c, v in zip(cursor.description, row)
-                    if c.name in extended_columns
-                },
-                "insert",
-            )
-            for row in cursor.fetchall()
-        ]
-
-        update_sa_table = oedb_table._edit_table.get_sa_table()
-        _execute(
-            cursor,
-            'select * from "{schema}"."{table}" where _applied = FALSE;'.format(
-                schema=update_sa_table.schema, table=update_sa_table.name
-            ),
-        )
-        changes += [
-            add_type(
-                {
-                    c.name: v
-                    for c, v in zip(cursor.description, row)
-                    if c.name in extended_columns
-                },
-                "update",
-            )
-            for row in cursor.fetchall()
-        ]
+        meta_tables = {
+            "insert": (oedb_table._insert_table, extended_columns),
+            "update": (oedb_table._edit_table, extended_columns),
+            "delete": (oedb_table._delete_table, ["_id", "id", "_submitted"]),
+        }
 
-        delete_sa_table = oedb_table._delete_table.get_sa_table()
-        _execute(
-            cursor,
-            'select * from "{schema}"."{table}" where _applied = FALSE;'.format(
-                schema=delete_sa_table.schema, table=delete_sa_table.name
-            ),
-        )
-        changes += [
-            add_type(
-                {
-                    c.name: v
-                    for c, v in zip(cursor.description, row)
-                    if c.name in ["_id", "id", "_submitted"]
-                },
-                "delete",
+        changes = []
+        for change_type in change_types:
+            meta_table, relevant_columns = meta_tables[change_type]
+            meta_sa_table = meta_table.get_sa_table()
+            _execute(
+                cursor,
+                'select * from "{schema}"."{table}" where _applied = FALSE;'.format(
+                    schema=meta_sa_table.schema, table=meta_sa_table.name
+                ),
             )
-            for row in cursor.fetchall()
-        ]
+            changes += [
+                add_type(
+                    {
+                        c.name: v
+                        for c, v in zip(cursor.description, row)
+                        if c.name in relevant_columns
+                    },
+                    change_type,
+                )
+                for row in cursor.fetchall()
+            ]
 
-        changes = list(changes)
         sa_table = oedb_table._main_table.get_sa_table()
 
         # ToDo: This may require some kind of dependency tree resolution
@@ -1415,10 +1392,9 @@ def add_type(d, type):
             if prev_type and change["_type"] != prev_type:
                 _apply_stack(cursor, sa_table, change_batch, prev_type)
                 change_batch = []
-            else:
-                change_batch.append((distilled_change, change["_id"]))
+            change_batch.append((distilled_change, change["_id"]))
             prev_type = change["_type"]
-        if prev_type:
+        if change_batch:
             _apply_stack(cursor, sa_table, change_batch, prev_type)
         if artificial_connection:
             connection.commit()
@@ -1458,15 +1434,16 @@ def set_applied(
     else:
         raise NotImplementedError
 
+    # Set-based comparison instead of an N-term OR chain; the extra
+    # `_applied = FALSE` condition lets the partial index on unapplied
+    # rows serve this update.
     update_query = (
         meta_sa_table.update()
-        .where(sql.or_(*(meta_sa_table.c._id == i for i in rids)))
+        .where(meta_sa_table.c._id.in_(list(rids)))
+        .where(meta_sa_table.c._applied == sql.false())
         .values(_applied=True)
-        .compile()
     )
-
-    query = str(update_query)
-    _execute(session, query, update_query.params)
+    execute_sqla(update_query, session)
 
 
 def apply_insert(session: AbstractCursor | Session, sa_table: "SATable", rows, rids):
@@ -1478,23 +1455,23 @@ def apply_insert(session: AbstractCursor | Session, sa_table: "SATable", rows, r
 
 def apply_update(session: AbstractCursor | Session, sa_table, rows, rids):
     logger.debug("apply updates (%d)", len(rids))
-    for row, rid in zip(rows, rids):
+    for row in rows:
         pks = [c.name for c in sa_table.columns if c.primary_key]
         query = sa_table.update(
             *[getattr(sa_table.c, pk) == row[pk] for pk in pks]
         ).values(row)
         execute_sqla(query, session)
-        set_applied(session, sa_table, [rid], __UPDATE)
+    set_applied(session, sa_table, list(rids), __UPDATE)
 
 
 def apply_deletion(session: AbstractCursor | Session, sa_table: "SATable", rows, rids):
     logger.debug("apply deletion (%d)", len(rids))
-    for row, rid in zip(rows, rids):
+    for row in rows:
         query = sa_table.delete().where(
             *[getattr(sa_table.c, col) == row[col] for col in row]
         )
         execute_sqla(query, session)
-        set_applied(session, sa_table, [rid], __DELETE)
+    set_applied(session, sa_table, list(rids), __DELETE)
 
 
 def update_meta_search(table: str) -> None:
@@ -1785,7 +1762,7 @@ def data_insert(request: dict, context: dict) -> dict:
         ]
     response["rowcount"] = cursor.rowcount
 
-    apply_changes(table_obj, cursor)
+    apply_changes(table_obj, cursor, change_types=("insert",))
 
     return response
 
@@ -1810,7 +1787,7 @@ def data_delete(request: dict, context: dict) -> dict:
     )
 
     result = __change_rows(table_obj, request, context, sa_table_delete, setter, ["id"])
-    apply_changes(table_obj, cursor)
+    apply_changes(table_obj, cursor, change_types=("delete",))
     return result
 
 
@@ -1834,7 +1811,7 @@ def data_update(request: dict, context: dict) -> dict:
         setter = dict(zip(field_names, setter))
     cursor = load_cursor_from_context(context)  # TODO:
     result = __change_rows(table_obj, request, context, sa_table_edit, setter)
-    apply_changes(table_obj, cursor)
+    apply_changes(table_obj, cursor, change_types=("update",))
     return result
 
 

api/views.py

@@ -61,7 +61,6 @@
 import login.models as login_models
 from api import sessions
 from api.actions import (
-    apply_changes,
     close_cursor,
     close_raw_connection,
     column_add,
@@ -645,7 +644,6 @@ def post(
                 status_code = status.HTTP_201_CREATED
             else:
                 response = self.__update_rows(request, table_obj, payload_query, None)
-        apply_changes(table_obj)
         return stream(response, status_code=status_code)
 
     @api_exception
@@ -691,11 +689,9 @@ def put(
         exists = table_has_row_with_id(table_obj, id=row_id) if row_id else False
         if exists:
             response = self.__update_rows(request, table_obj, payload_query, row_id)
-            apply_changes(table_obj)
             return JsonResponse(response)
         else:
             result = self.__insert_row(request, table_obj, payload_query, row_id)
-            apply_changes(table_obj)
             return JsonResponse(result, status=status.HTTP_201_CREATED)
 
     @api_exception
@@ -712,7 +708,6 @@ def delete(
             )
 
         result = self.__delete_rows(request, table_obj, row_id)
-        apply_changes(table_obj)
         return JsonResponse(result)
 
     @load_cursor()

oedb/utils.py

@@ -190,15 +190,35 @@ def __init__(
         )
 
     def _create_if_missing(self, include_indexes: bool = True) -> None:
+        # Short-circuit when the table exists: the DDL below must only run at
+        # actual creation time. Running it on every access can deadlock, e.g.
+        # the CREATE INDEX waits behind an uncommitted insert into the meta
+        # table held by the very transaction that triggered this call.
+        if self.exists():
+            return None
         query = (
             f'CREATE TABLE IF NOT EXISTS "{self.schema_name}"."{self.name}" (LIKE '
             f'"{self.main_table.schema_name}"."{self.main_table.name}"'
         )
         if include_indexes:
             query += "INCLUDING ALL EXCLUDING INDEXES, PRIMARY KEY (_id) "
         query += f") INHERITS ({EditBase.__tablename__});"
+        # Partial index so that finding/marking unapplied changes doesn't
+        # sequentially scan the meta table as it grows (issue #2362).
+        query += (
+            f' CREATE INDEX IF NOT EXISTS "{self.unapplied_index_name}" ON '
+            f'"{self.schema_name}"."{self.name}" (_id) WHERE _applied = FALSE;'
+        )
         return self._execute(query, requires_permission=ADMIN_PERM)
 
+    @property
+    def unapplied_index_name(self) -> str:
+        # postgres truncates identifiers to 63 bytes; truncate explicitly so
+        # CREATE INDEX IF NOT EXISTS matches the effective name on re-runs.
+        # Must stay in sync with _index_name in the oedb migration
+        # e3b1f6c2d9a4_index_unapplied_meta_rows.py.
+        return f"{self.name}_unapplied_idx"[:63]
+
     def get_sa_table(self) -> SATable:
         # create on demand
         self._create_if_missing()

oedb/versions/e3b1f6c2d9a4_index_unapplied_meta_rows.py

@@ -0,0 +1,57 @@
+"""Add partial index on unapplied rows to all existing meta tables
+
+Every write to a table is journaled in its meta tables
+(_<name>_insert/_edit/_delete) and applied by scanning them with
+``WHERE _applied = FALSE``. Without an index those scans are sequential
+and grow with the journal, which slows every upload as tables age
+(issue #2362). New meta tables get this index on creation; this
+migration back-fills all existing ones. Meta tables are found via
+postgres inheritance from public._edit_base.
+
+Revision ID: e3b1f6c2d9a4
+Revises: 89f049e538aa
+Create Date: 2026-07-03
+
+
+SPDX-FileCopyrightText: 2026 Jonas Huber <https://github.com/jh-RLI> © Reiner Lemoine Institut
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""  # noqa: 501
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e3b1f6c2d9a4"
+down_revision = "89f049e538aa"
+branch_labels = None
+depends_on = None
+
+META_TABLES_QUERY = """
+    SELECT n.nspname, c.relname
+    FROM pg_inherits i
+    JOIN pg_class c ON c.oid = i.inhrelid
+    JOIN pg_namespace n ON n.oid = c.relnamespace
+    JOIN pg_class p ON p.oid = i.inhparent
+    JOIN pg_namespace pn ON pn.oid = p.relnamespace
+    WHERE p.relname = '_edit_base' AND pn.nspname = 'public'
+"""
+
+
+def _index_name(table_name):
+    # postgres truncates identifiers to 63 bytes; truncate explicitly so
+    # the name matches what CREATE INDEX IF NOT EXISTS uses at runtime
+    return f"{table_name}_unapplied_idx"[:63]
+
+
+def upgrade():
+    connection = op.get_bind()
+    for schema, table in connection.execute(META_TABLES_QUERY).fetchall():
+        connection.execute(
+            f'CREATE INDEX IF NOT EXISTS "{_index_name(table)}" '
+            f'ON "{schema}"."{table}" (_id) WHERE _applied = FALSE;'
+        )
+
+
+def downgrade():
+    connection = op.get_bind()
+    for schema, table in connection.execute(META_TABLES_QUERY).fetchall():
+        connection.execute(f'DROP INDEX IF EXISTS "{schema}"."{_index_name(table)}";')

versions/changelogs/current.md

@@ -11,6 +11,14 @@ SPDX-License-Identifier: CC0-1.0
 
 ## Changes
 
+- Speed up the row upload API: index the unapplied rows of the edit-journal meta
+  tables (`_<table>_insert/_edit/_delete`, back-filled by an oedb migration),
+  mark applied rows with one set-based update instead of a per-row OR chain,
+  scan only the meta table relevant to the operation, and apply changes exactly
+  once per request instead of twice. Also fixes a bug where the first change of
+  each type was dropped when a meta table held mixed pending change types.
+  [(#2362)](https://github.com/OpenEnergyPlatform/oeplatform/issues/2362)
+
 ## Features
 
 - Redesign the OPR Summary tab as a condensed, grouped overview with per-state