feat(data-index): per-upload data tables, safe re-upload, delete clean-up

.github/workflows/django-tests.yml

@@ -16,6 +16,20 @@ jobs:
   django-tests:
     runs-on: ubuntu-latest
 
+    # The data database: tests marked data_db build real tables in it.
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_PASSWORD: postgres
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+
     steps:
       - uses: actions/checkout@v4
 
@@ -40,6 +54,10 @@ jobs:
       # tests/object_types/charts and the dataset metadata tests were left
       # behind by the chart consolidation and fail on main: see #189.
       - name: Run tests
+        env:
+          TEST_DATA_DB_URL: postgres://postgres:postgres@localhost:5432/postgres
+          # Fail the run instead of skipping the data_db tests if the database is missing.
+          REQUIRE_DATA_DB_TESTS: "1"
         run: |
           pytest tests/ \
             --ignore=tests/object_types/charts \

.github/workflows/sdk-tests.yml

@@ -8,7 +8,8 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    # Pinned: ubuntu-latest (26.04) has no Python 3.9 build, which the SDK still supports.
+    runs-on: ubuntu-24.04
     strategy:
       matrix:
         python-version: ["3.9", "3.10", "3.11"]

api/management/commands/migrate_data_table_names.py

@@ -0,0 +1,166 @@
+"""Move existing data tables to per-upload names, and list the tables nothing uses.
+
+    python manage.py migrate_data_table_names            # dry run: write a report, change nothing
+    python manage.py migrate_data_table_names --apply    # rename, rebuild, fix rows
+    python manage.py migrate_data_table_names --apply --drop-orphans --from-report <reviewed.json>
+
+    --resource-id <uuid>   act on one resource only
+    --report <path>        where to write the JSON report (default: a timestamped file here)
+
+Short old names are renamed in place; long ones (which may hold another
+resource's data) are rebuilt from each resource's own file. A long one whose
+file is missing or unusable is renamed and marked "unverified" (listed in
+every report until the resource is re-uploaded), or left alone if another row
+shares it. A dry run checks the files too, so it shows what --apply will do.
+Safe to rerun: each run picks up whatever is left ("nothing to do" once only
+rows needing a person remain), and doubles as an orphan audit.
+
+Docs (runbook, failure model): ../../utils/data_indexing_architecture.md
+"""
+
+from typing import Any, Dict, List
+
+from django.core.management.base import BaseCommand, CommandError, CommandParser
+from django.utils import timezone
+
+from api.utils.data_table_migration import (
+    access_models_touched,
+    build_report,
+    drop_reviewed_orphans,
+    find_orphans,
+    find_unverified,
+    nothing_left_to_do,
+    plan_table_migration,
+    point_row_at_its_table,
+    read_data_table_rows,
+    read_reviewed_orphans,
+    rebuild_decision,
+    rebuild_from_file,
+    record_outcome,
+    reindex_decision,
+    reindex_missing_table,
+    rename_legacy_table,
+    same_file,
+    summary_lines,
+    write_report,
+)
+from api.utils.data_tables import (
+    DataDbNotSeparate,
+    assert_data_db_is_separate,
+    list_data_db_tables,
+)
+
+
+class Command(BaseCommand):
+    help = "Move data tables to per-upload rdt_<id> names; list (and drop reviewed) orphans."
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        mode = parser.add_mutually_exclusive_group()
+        mode.add_argument("--dry-run", action="store_true", help="Report only (the default).")
+        mode.add_argument("--apply", action="store_true", help="Rename, rebuild and fix rows.")
+        parser.add_argument(
+            "--drop-orphans",
+            action="store_true",
+            help="With --apply: drop orphans listed in --from-report that are still orphans.",
+        )
+        parser.add_argument("--from-report", help="A dry-run report someone reviewed.")
+        parser.add_argument("--resource-id", help="Act on this resource only.")
+        parser.add_argument("--report", help="Where to write this run's JSON report.")
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        apply, drop_orphans, from_report = (
+            options["apply"],
+            options["drop_orphans"],
+            options["from_report"],
+        )
+        mode = "apply" if apply else "dry-run"
+        now = timezone.now()
+        report_path = options["report"] or f"data_table_migration_{now:%Y%m%dT%H%M%SZ}.json"
+
+        # Orphans are dropped only by an applying run, and only from a reviewed report.
+        if drop_orphans and not (apply and from_report):
+            raise CommandError("--drop-orphans needs --apply and --from-report <reviewed report>")
+        if from_report and same_file(report_path, from_report):
+            raise CommandError("--report would overwrite the reviewed --from-report file")
+
+        # Never touch anything if the data database might be the main database.
+        try:
+            assert_data_db_is_separate()
+        except DataDbNotSeparate as error:
+            raise CommandError(f"Refusing to run: {error}") from error
+
+        # Read the reviewed orphan list now, so a bad file stops the run before any change.
+        reviewed = read_reviewed_orphans(from_report) if drop_orphans else set()
+
+        # Write an in-progress report first: the path works, and a run that dies leaves a trace.
+        write_report(report_path, build_report(mode, now, status="in_progress"))
+
+        # Read every row and every data table, and work out what each row needs.
+        plan = plan_table_migration(
+            read_data_table_rows(), list_data_db_tables(), now, resource_id=options["resource_id"]
+        )
+        entries: List[Dict[str, Any]] = [
+            record_outcome(row, "none", "already_migrated") for row in plan.already_migrated
+        ]
+
+        if apply:
+            # Finish renames a crashed run left half done: just point the row at its table.
+            for row in plan.row_updates:
+                entries.append(record_outcome(row, "row_update", point_row_at_its_table(row)))
+
+            # Rename short old tables in place; they can't have collided.
+            for row in plan.renames:
+                entries.append(record_outcome(row, "rename", rename_legacy_table(row)))
+
+            # Rebuild long old tables from each resource's own file, oldest first.
+            for row, shared in plan.rebuilds:
+                outcome = rebuild_from_file(row, shared)
+                entries.append(record_outcome(row, "rebuild", outcome, shared=shared))
+
+            # Re-index rows whose table is gone, from their file.
+            for row in plan.missing_tables:
+                entries.append(record_outcome(row, "reindex", reindex_missing_table(row)))
+        else:
+            # A dry run changes nothing, but checks each stored file so the report shows
+            # what --apply would really do (rebuild, unverified rename, or leave for a person).
+            for row in plan.row_updates:
+                entries.append(record_outcome(row, "row_update", "planned_row_update"))
+            for row in plan.renames:
+                entries.append(record_outcome(row, "rename", "planned_rename"))
+            for row, shared in plan.rebuilds:
+                decision = rebuild_decision(row, shared)
+                entries.append(record_outcome(row, "rebuild", f"planned_{decision}", shared=shared))
+            for row in plan.missing_tables:
+                entries.append(record_outcome(row, "reindex", f"planned_{reindex_decision(row)}"))
+
+        # Rebuilt resources get new column rows, which resets access models' column choices.
+        touched = access_models_touched(
+            [row.resource_id for row, _ in plan.rebuilds]
+            + [row.resource_id for row in plan.missing_tables]
+        )
+
+        dropped: List[str] = []
+        if drop_orphans:
+            # Drop the reviewed orphans that are still orphans right now.
+            current = find_orphans(read_data_table_rows(), list_data_db_tables(), timezone.now())
+            dropped = drop_reviewed_orphans(reviewed, current)
+
+        # List every table still marked unverified, from this run or any earlier one.
+        unverified = find_unverified(read_data_table_rows(), list_data_db_tables())
+
+        # Write the full report and print the summary.
+        report = build_report(
+            mode,
+            now,
+            entries,
+            plan.orphans,
+            dropped,
+            touched,
+            unverified,
+            plan.shared_groups,
+        )
+        write_report(report_path, report)
+        if nothing_left_to_do(entries):
+            self.stdout.write("nothing to do")
+        for line in summary_lines(report, report_path):
+            self.stdout.write(line)

api/management/commands/update_dataindex.py

@@ -4,7 +4,6 @@
 
 import structlog
 from django.core.management.base import BaseCommand, CommandParser
-from django.db import transaction
 
 from api.models.Resource import Resource, ResourceDataTable
 from api.utils.data_indexing import INDEXED_FORMATS, index_resource_data
@@ -125,9 +124,9 @@ def handle(self, *args: Any, **options: Dict[str, Any]) -> None:
                     f"[{i}/{total_resources}] Processing resource {resource.id} - {resource.name}"
                 )
 
-                # Update dataindex for the resource
-                with transaction.atomic():
-                    data_table = index_resource_data(resource)
+                # Not inside a transaction: the flow commits its own switch and refuses
+                # to run inside an outer one (a failed resource keeps its old data).
+                data_table = index_resource_data(resource)
 
                 if data_table:
                     success_count += 1

api/models/Resource.py

@@ -11,6 +11,7 @@
 
 logger = structlog.getLogger(__name__)
 
+from api.utils.data_table_names import data_table_name
 from api.utils.enums import DataType
 from api.utils.storage import private_storage
 
@@ -101,11 +102,8 @@ def __str__(self):
 
     def save(self, *args, **kwargs):
         if not self.table_name:
-            # Generate a unique table name based on resource ID
-            safe_name = slugify(self.resource.name)[:50]
-            safe_title = slugify(self.resource.dataset.title)[:50]
-            resource_id = self.resource.id.hex[:8]  # Use shorter ID
-            self.table_name = f"{safe_name}_{safe_title}_{resource_id}"
+            # Named from this row's own id: unique, and never cut by Postgres's 63-byte limit.
+            self.table_name = data_table_name(self.id)
         super().save(*args, **kwargs)
 
 

api/schema/resource_schema.py

@@ -184,6 +184,12 @@ def _validate_file_details_and_update_format(resource: Resource) -> None:
 
 def _create_file_resource_schema(resource: Resource) -> None:
     """Create file resource schema."""
+    # TODO: a failed indexing (the file saved, but its rows couldn't be loaded) still
+    # reports success, so the download serves the new file while previews, charts and
+    # the data API keep the old one. Pending a product decision: enforce the database's
+    # limits at upload (e.g. column names within 255 characters) and flag problem
+    # columns where people add column descriptions. See
+    # api/utils/data_indexing_architecture.md, Limitations.
     # Try to index CSV data if applicable
     data_table = index_resource_data(resource)
 

api/signals/__init__.py

@@ -1,6 +1,7 @@
 # Import signals to register them
 from api.signals import (
     aimodel_signals,
+    data_table_signals,
     dataset_signals,
     event_signals,
     publication_signals,

api/signals/data_table_signals.py

@@ -0,0 +1,33 @@
+"""Drop a data table once the row that points at it is deleted.
+
+Re-uploads, resource deletes, dataset deletes (by cascade) and admin deletes
+all delete the resource's ResourceDataTable row, so this one receiver frees the
+storage for every one of them. The drop waits for the delete to commit, so a
+rolled-back delete keeps its data, and it leaves a legacy table alone while
+another row still shares it.
+
+Registered in api/signals/__init__.py (a test proves it is connected once the
+app loads). Docs: ../utils/data_indexing_architecture.md
+"""
+
+from functools import partial
+from typing import Any
+
+from django.db import transaction
+from django.db.models.signals import post_delete
+from django.dispatch import receiver
+
+from api.models.Resource import ResourceDataTable
+from api.utils.data_tables import drop_data_table_if_unshared
+
+
+@receiver(post_delete, sender=ResourceDataTable)
+def drop_table_after_row_delete(
+    sender: Any, instance: ResourceDataTable, using: str, **kwargs: Any
+) -> None:
+    # Once the delete commits, drop the row's table (unless another row still uses it).
+    transaction.on_commit(
+        partial(drop_data_table_if_unshared, instance.table_name, instance.pk),
+        using=using,
+        robust=True,
+    )