2 Commits

Author SHA1 Message Date
53460242d8 Release version 3.1.1 2026-07-17 17:28:54 +02:00
01a00dd791 fix(restore): drop overloaded functions by identity signature in the --empty pre-clean
A signature-less DROP FUNCTION aborts with 'function name is not unique' as soon as a schema overloads a name, killing the whole --empty replay under ON_ERROR_STOP. The function branch now emits name(identity args) via pg_get_function_identity_arguments; because that compound must not be identifier-quoted as a whole, quoting moves from the outer format into each branch, guarded by a unit test that pins the per-branch %I quoting.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-17 17:27:55 +02:00
4 changed files with 35 additions and 8 deletions

View File

@@ -1,5 +1,16 @@
# Changelog # Changelog
## [3.1.1] - 2026-07-17
- Restore: the postgres *--empty* pre-clean drops functions and procedures
by their identity signature (*pg_get_function_identity_arguments*), so a
schema that overloads a function name (e.g. discourse) no longer aborts
the replay with "function name is not unique" under ON_ERROR_STOP.
Identifier quoting moves from the outer DROP format into each object
branch, since the *name(args)* compound must not be quoted as a whole; a
unit test pins the per-branch *%I* quoting so future branches cannot
regress unquoted.
## [3.1.0] - 2026-07-15 ## [3.1.0] - 2026-07-15
- Restore: the postgres *--empty* pre-clean emits one DROP per object and - Restore: the postgres *--empty* pre-clean emits one DROP per object and

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "backup-docker-to-local" name = "backup-docker-to-local"
version = "3.1.0" version = "3.1.1"
description = "Backup Docker volumes to local with rsync and optional DB dumps." description = "Backup Docker volumes to local with rsync and optional DB dumps."
readme = "README.md" readme = "README.md"
requires-python = ">=3.9" requires-python = ">=3.9"

View File

@@ -60,9 +60,9 @@ def restore_postgres_sql(
# discourse's discourse_functions) does not fail on an already-existing schema. # discourse's discourse_functions) does not fail on an already-existing schema.
# Extension members (pg_trgm's set_limit) are superuser-owned; IF EXISTS absorbs CASCADE fallout. # Extension members (pg_trgm's set_limit) are superuser-owned; IF EXISTS absorbs CASCADE fallout.
drop_sql = r""" drop_sql = r"""
SELECT format('DROP %s IF EXISTS public.%I CASCADE', obj.type, obj.name) SELECT format('DROP %s IF EXISTS public.%s CASCADE', obj.type, obj.name)
FROM ( FROM (
SELECT c.relname AS name, SELECT format('%I', c.relname) AS name,
CASE c.relkind CASE c.relkind
WHEN 'v' THEN 'VIEW' WHEN 'v' THEN 'VIEW'
WHEN 'm' THEN 'MATERIALIZED VIEW' WHEN 'm' THEN 'MATERIALIZED VIEW'
@@ -73,18 +73,20 @@ SELECT format('DROP %s IF EXISTS public.%I CASCADE', obj.type, obj.name)
WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'v', 'm', 'f') WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'v', 'm', 'f')
AND pg_get_userbyid(c.relowner) = current_user AND pg_get_userbyid(c.relowner) = current_user
UNION ALL UNION ALL
SELECT p.proname AS name, -- Overloaded functions share a proname; DROP needs the identity
-- signature or psql aborts with "function name is not unique".
SELECT format('%I(%s)', p.proname, pg_get_function_identity_arguments(p.oid)) AS name,
CASE p.prokind WHEN 'p' THEN 'PROCEDURE' ELSE 'FUNCTION' END AS type CASE p.prokind WHEN 'p' THEN 'PROCEDURE' ELSE 'FUNCTION' END AS type
FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
WHERE n.nspname = 'public' AND p.prokind IN ('f', 'p', 'w') WHERE n.nspname = 'public' AND p.prokind IN ('f', 'p', 'w')
AND pg_get_userbyid(p.proowner) = current_user AND pg_get_userbyid(p.proowner) = current_user
UNION ALL UNION ALL
SELECT c.relname AS name, 'SEQUENCE' AS type SELECT format('%I', c.relname) AS name, 'SEQUENCE' AS type
FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = 'public' AND c.relkind = 'S' WHERE n.nspname = 'public' AND c.relkind = 'S'
AND pg_get_userbyid(c.relowner) = current_user AND pg_get_userbyid(c.relowner) = current_user
UNION ALL UNION ALL
SELECT t.typname AS name, 'TYPE' AS type SELECT format('%I', t.typname) AS name, 'TYPE' AS type
FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace
WHERE n.nspname = 'public' WHERE n.nspname = 'public'
AND pg_get_userbyid(t.typowner) = current_user AND pg_get_userbyid(t.typowner) = current_user
@@ -93,7 +95,7 @@ SELECT format('DROP %s IF EXISTS public.%I CASCADE', obj.type, obj.name)
SELECT 1 FROM pg_class c2 SELECT 1 FROM pg_class c2
WHERE c2.oid = t.typrelid AND c2.relkind = 'c'))) WHERE c2.oid = t.typrelid AND c2.relkind = 'c')))
UNION ALL UNION ALL
SELECT col.collname AS name, 'COLLATION' AS type SELECT format('%I', col.collname) AS name, 'COLLATION' AS type
FROM pg_collation col JOIN pg_namespace n ON n.oid = col.collnamespace FROM pg_collation col JOIN pg_namespace n ON n.oid = col.collnamespace
WHERE n.nspname = 'public' WHERE n.nspname = 'public'
AND pg_get_userbyid(col.collowner) = current_user AND pg_get_userbyid(col.collowner) = current_user

View File

@@ -36,7 +36,21 @@ class TestPostgresEmptyDrop(unittest.TestCase):
sql = self._drop_sql() sql = self._drop_sql()
for marker in ("pg_class", "pg_proc", "'SEQUENCE' AS type", "'TYPE' AS type"): for marker in ("pg_class", "pg_proc", "'SEQUENCE' AS type", "'TYPE' AS type"):
self.assertIn(marker, sql) self.assertIn(marker, sql)
self.assertIn("DROP %s IF EXISTS public.%I CASCADE", sql) self.assertIn("DROP %s IF EXISTS public.%s CASCADE", sql)
def test_functions_drop_with_identity_signature(self) -> None:
"""Overloaded functions share a proname; a signature-less DROP aborts
with "function name is not unique", so the function branch must emit
name(identity args) while every other branch stays %I-quoted."""
sql = self._drop_sql()
self.assertIn("pg_get_function_identity_arguments(p.oid)", sql)
self.assertIn("format('%I(%s)', p.proname", sql)
for branch in (
"format('%I', c.relname)",
"format('%I', t.typname)",
"format('%I', col.collname)",
):
self.assertIn(branch, sql)
if __name__ == "__main__": if __name__ == "__main__":