[Checkins] SVN: relstorage/branches/postgres_blob_oid/relstorage/adapters/ Update PostgreSQL schema to switch ZODB BLOBS from BYTEA to OID.

[Martijn Pieters]

Log message for revision 121922:
  Update PostgreSQL schema to switch ZODB BLOBS from BYTEA to OID.
  
  Included is a AFTER DELETE trigger that unlinks the OID data when the blob_chunk row is deleted.

Changed:
  U   relstorage/branches/postgres_blob_oid/relstorage/adapters/mover.py
  U   relstorage/branches/postgres_blob_oid/relstorage/adapters/schema.py

-=-
Modified: relstorage/branches/postgres_blob_oid/relstorage/adapters/mover.py
===================================================================
--- relstorage/branches/postgres_blob_oid/relstorage/adapters/mover.py	2011-06-12 12:02:37 UTC (rev 121921)
+++ relstorage/branches/postgres_blob_oid/relstorage/adapters/mover.py	2011-06-12 13:18:54 UTC (rev 121922)
@@ -392,7 +392,7 @@
         CREATE TEMPORARY TABLE temp_blob_chunk (
             zoid        BIGINT NOT NULL,
             chunk_num   BIGINT NOT NULL,
-            chunk       BYTEA
+            chunk       OID
         ) ON COMMIT DROP;
         CREATE UNIQUE INDEX temp_blob_chunk_key
             ON temp_blob_chunk (zoid, chunk_num);

Modified: relstorage/branches/postgres_blob_oid/relstorage/adapters/schema.py
===================================================================
--- relstorage/branches/postgres_blob_oid/relstorage/adapters/schema.py	2011-06-12 12:02:37 UTC (rev 121921)
+++ relstorage/branches/postgres_blob_oid/relstorage/adapters/schema.py	2011-06-12 13:18:54 UTC (rev 121922)
@@ -104,7 +104,7 @@
             tid         BIGINT NOT NULL,
             chunk_num   BIGINT NOT NULL,
                         PRIMARY KEY (zoid, tid, chunk_num),
-            chunk       BYTEA NOT NULL
+            chunk       OID NOT NULL
         );
         CREATE INDEX blob_chunk_lookup ON blob_chunk (zoid, tid);
         ALTER TABLE blob_chunk ADD CONSTRAINT blob_chunk_fk
@@ -394,6 +394,27 @@
         CREATE SEQUENCE zoid_seq;
 """
 
+postgresql_history_preserving_plpgsql = """
+CREATE OR REPLACE FUNCTION blob_chunk_delete_trigger() RETURNS TRIGGER 
+AS $blob_chunk_delete_trigger$
+    -- Version: %s
+    -- Unlink large object data file after blob_chunck row deletion
+    BEGIN
+        PERFORM lo_unlink(OLD.chunk);
+        RETURN NULL;
+    END;
+$blob_chunk_delete_trigger$ LANGUAGE plpgsql;
+/
+
+DROP TRIGGER IF EXISTS blob_chunk_delete ON blob_chunk;
+/
+CREATE TRIGGER blob_chunk_delete 
+    AFTER DELETE ON blob_chunk
+    FOR EACH ROW
+    EXECUTE PROCEDURE blob_chunk_delete_trigger();
+/
+""" % relstorage_op_version
+
 oracle_history_preserving_plsql = """
 CREATE OR REPLACE PACKAGE relstorage_op AS
     TYPE numlist IS TABLE OF NUMBER(20) INDEX BY BINARY_INTEGER;
@@ -494,7 +515,7 @@
             chunk_num   BIGINT NOT NULL,
                         PRIMARY KEY (zoid, chunk_num),
             tid         BIGINT NOT NULL,
-            chunk       BYTEA NOT NULL
+            chunk       OID NOT NULL
         );
         CREATE INDEX blob_chunk_lookup ON blob_chunk (zoid);
         ALTER TABLE blob_chunk ADD CONSTRAINT blob_chunk_fk
@@ -674,6 +695,8 @@
         CREATE SEQUENCE zoid_seq;
 """
 
+postgresql_history_free_plpgsql = postgresql_history_preserving_plpgsql
+
 oracle_history_free_plsql = """
 CREATE OR REPLACE PACKAGE relstorage_op AS
     TYPE numlist IS TABLE OF NUMBER(20) INDEX BY BINARY_INTEGER;
@@ -885,6 +908,41 @@
             connmanager, runner, keep_history)
         self.locker = locker
 
+    def prepare(self):
+        """Create the database schema if it does not already exist."""
+        def callback(conn, cursor):
+            tables = self.list_tables(cursor)
+            if not 'object_state' in tables:
+                self.create(cursor)
+            else:
+                self.check_compatibility(cursor, tables)
+                self.update_schema(cursor, tables)
+            triggers = self.list_triggers(cursor)
+            if triggers.get('blob_chunk_delete_trigger') != relstorage_op_version:
+                self.install_triggers(cursor)
+                triggers = self.list_triggers(cursor)
+                if triggers.get('blob_chunk_delete_trigger') != relstorage_op_version:
+                    raise AssertionError(
+                        "Could not get version information after "
+                        "installing the blob_chunk_delete_trigger trigger.")
+        self.connmanager.open_and_call(callback)
+
+    def install_triggers(self, cursor):
+        """Install the PL/pgSQL triggers"""
+        if self.keep_history:
+            plpgsql = postgresql_history_preserving_plpgsql
+        else:
+            plpgsql = postgresql_history_free_plpgsql
+
+        lines = []
+        for line in plpgsql.splitlines():
+            if line.strip() == '/':
+                # end of a statement
+                cursor.execute('\n'.join(lines))
+                lines = []
+            elif line.strip():
+                lines.append(line)
+
     def create(self, cursor):
         """Create the database tables."""
         super(PostgreSQLSchemaInstaller, self).create(cursor)
@@ -899,7 +957,25 @@
         cursor.execute("SELECT relname FROM pg_class WHERE relkind = 'S'")
         return [name for (name,) in cursor]
 
+    def list_triggers(self, cursor):
+        """Returns {trigger name: version}.  version may be None."""
+        stmt = """
+        SELECT proname, prosrc
+        FROM pg_catalog.pg_namespace n
+        JOIN pg_catalog.pg_proc p ON pronamespace = n.oid
+        JOIN pg_catalog.pg_type t ON prorettype = t.oid
+        WHERE nspname = 'public' AND typname = 'trigger'
+        """
+        cursor.execute(stmt)
+        res = {}
+        for (name, text) in cursor:
+            version = None
+            match = re.search(r'Version:\s*([0-9a-zA-Z.]+)', text)
+            if match is not None:
+                version = match.group(1)
+            res[name.lower()] = version
 
+
 class MySQLSchemaInstaller(AbstractSchemaInstaller):
     implements(ISchemaInstaller)