Support for automatic handling of fkeys.

By Eric Jones, plus some hacking by me.

diff --git a/python/londiste/playback.py b/python/londiste/playback.py
index 82c7d0c03a070f06ac77640c56f738cbbc50ecd5..ebc8920491592c9f976dd279dd1d843776598343 100644 (file)
--- a/python/londiste/playback.py
+++ b/python/londiste/playback.py
@@ -249,6 +249,8 @@ class Replicator(pgq.SerialConsumer):
         self.load_table_state(dst_curs)
         self.sync_tables(dst_db)
 
+        self.restore_fkeys(dst_db)
+
         # now the actual event processing happens.
         # they must be done all in one tx in dst side
         # and the transaction must be kept open so that
@@ -346,7 +348,7 @@ class Replicator(pgq.SerialConsumer):
         "Copy thread sync logic."
 
         #
-        # decide what to do - order is imortant
+        # decide what to do - order is important
         #
         if cnt.do_sync:
             # main thread is waiting, catch up, then handle over
@@ -580,6 +582,18 @@ class Replicator(pgq.SerialConsumer):
             if src_enc != dst_enc:
                 dst_curs.execute("set client_encoding = %s", [src_enc])
 
+    def restore_fkeys(self, dst_db):
+        dst_curs = dst_db.cursor()
+        # restore fkeys -- one at a time
+        q = "select * from londiste.subscriber_get_queue_valid_pending_fkeys(%s)"
+        dst_curs.execute(q, [self.pgq_queue_name])
+        list = dst_curs.dictfetchall()
+        for row in list:
+            self.log.info('Creating fkey: %(fkey_name)s (%(from_table)s --> %(to_table)s)' % row)
+            q2 = "select londiste.subscriber_restore_table_fkey(%(from_table)s, %(fkey_name)s)"
+            dst_curs.execute(q2, row)
+            dst_db.commit()
+        
 if __name__ == '__main__':
     script = Replicator(sys.argv[1:])
     script.start()

diff --git a/python/londiste/setup.py b/python/londiste/setup.py
index 2e20b164ca405ca24a94a925a5cf988ae2bfc568..f42291b24d7764efb1d91e9f9c58ea2237b478f6 100644 (file)
--- a/python/londiste/setup.py
+++ b/python/londiste/setup.py
@@ -293,7 +293,7 @@ class SubscriberSetup(CommonSetup):
         elif cmd == "check":
             self.check_tables(self.get_provider_table_list())
         elif cmd == "fkeys":
-            self.collect_fkeys(self.get_provider_table_list())
+            self.collect_fkeys(self.get_provider_table_list(), self.args[3:])
         elif cmd == "seqs":
             self.subscriber_list_seqs()
         elif cmd == "add-seq":
@@ -304,40 +304,41 @@ class SubscriberSetup(CommonSetup):
             self.log.error('bad subcommand: ' + cmd)
             sys.exit(1)
 
-    def collect_fkeys(self, table_list):
+    def collect_fkeys(self, table_list, args):
+        if args == []:
+            args = ['pending', 'active']
+            
         dst_db = self.get_database('subscriber_db')
         dst_curs = dst_db.cursor()
+        
+        subscriber_tables = self.get_subscriber_table_list()
 
-        oid_list = []
-        for tbl in table_list:
-            try:
-                oid = skytools.get_table_oid(dst_curs, tbl)
-                if oid:
-                    oid_list.append(str(oid))
-            except:
-                pass
-        if len(oid_list) == 0:
-            print "No tables"
-            return
-        oid_str = ",".join(oid_list)
-
-        q = "SELECT n.nspname || '.' || t.relname as tbl, c.conname as con,"\
-            "       pg_get_constraintdef(c.oid) as def"\
-            "  FROM pg_constraint c, pg_class t, pg_namespace n"\
-            " WHERE c.contype = 'f' and (c.conrelid in (%s) or c.confrelid in (%s))"\
-            "   AND t.oid = c.conrelid AND n.oid = t.relnamespace" % (oid_str, oid_str)
-        dst_curs.execute(q)
-        res = dst_curs.dictfetchall()
-        dst_db.commit()
-
-        print "-- dropping"
-        for row in res:
-            q = "ALTER TABLE ONLY %(tbl)s DROP CONSTRAINT %(con)s;"
-            print q % row
-        print "-- creating"
-        for row in res:
-            q = "ALTER TABLE ONLY %(tbl)s ADD CONSTRAINT %(con)s %(def)s;"
-            print q % row
+        fkeys = {'active_fkeys': [], 'pending_fkeys': []}
+        for type in args:
+            q = "select * from londiste.subscriber_get_table_%s_fkeys(%%s);" % type
+            for tbl in table_list:
+                if tbl not in subscriber_tables:
+                    continue
+                dst_curs.execute(q, [tbl])
+                fkeys['%s_fkeys' % type].extend(dst_curs.dictfetchall())
+        dst_db.commit()                    
+        
+        for type in args:
+            widths = [15,15,15]
+            for row in fkeys['%s_fkeys' % type]:
+                widths[0] = widths[0] > len(row[0]) and widths[0] or len(row[0])
+                widths[1] = widths[1] > len(row[1]) and widths[1] or len(row[1])
+                widths[2] = widths[2] > len(row[2]) and widths[2] or len(row[2])
+            widths[0] += 2; widths[1] += 2; widths[2] += 2
+            
+            fmt = '%%-%ds%%-%ds%%-%ds%%s' % tuple(widths)
+            print '%s fkeys:' % type
+            print fmt % ('from_table', 'to_table', 'fkey_name', 'fkey_def')
+            print fmt % ('----------', '--------', '---------', '--------')
+                
+            for row in fkeys['%s_fkeys' % type]:
+                print fmt % row.values()
+            print '\n'
 
     def check_tables(self, table_list):
         src_db = self.get_database('provider_db')
@@ -368,7 +369,7 @@ class SubscriberSetup(CommonSetup):
         if not oid:
             self.log.error('Table %s not found' % tbl)
             return 1
-        q = "select count(1) from pg_trigger where tgrelid = %s"
+        q = "select count(1) from pg_trigger where tgrelid = %s and tgisconstraint is not true"
         dst_curs.execute(q, [oid])
         got = dst_curs.fetchone()[0]
         if got:

diff --git a/python/londiste/table_copy.py b/python/londiste/table_copy.py
index edbcbbdeeedda89caf7b1a63c6aa0d4929fbea46..ba359642a026f7ffff0d00d67b5111719a00d14d 100644 (file)
--- a/python/londiste/table_copy.py
+++ b/python/londiste/table_copy.py
@@ -50,6 +50,17 @@ class CopyTable(Replicator):
         for c in src_struct.get_column_list():
             if c not in dlist:
                 raise Exception('Column %s does not exist on dest side' % c)
+        
+        # drop all foreign keys to and from this table
+        # they need to be dropped one at a time to avoid deadlocks with user code
+        q = "select * from londiste.find_table_fkeys(%s)"
+        dst_curs.execute(q, [dst_struct.table_name])
+        list = dst_curs.dictfetchall()
+        for row in list:
+            self.log.info('Dropping fkey: %s' % row['fkey_name'])
+            q2 = "select londiste.subscriber_drop_table_fkey(%(from_table)s, %(fkey_name)s)"
+            dst_curs.execute(q2, row)
+            dst_db.commit()
 
         # drop unnecessary stuff
         objs = T_CONSTRAINT | T_INDEX | T_TRIGGER | T_RULE
@@ -69,13 +80,14 @@ class CopyTable(Replicator):
 
         # create previously dropped objects
         dst_struct.create(dst_curs, objs, log = self.log)
+        dst_db.commit()
 
         # set state
-        tbl_stat.change_snapshot(snapshot)
         if self.copy_thread:
             tbl_stat.change_state(TABLE_CATCHING_UP)
         else:
             tbl_stat.change_state(TABLE_OK)
+        tbl_stat.change_snapshot(snapshot)
         self.save_table_state(dst_curs)
         dst_db.commit()
 

diff --git a/python/skytools/dbstruct.py b/python/skytools/dbstruct.py
index ece0ea5454956e94523e0f6d9f18c98c841153f3..39e3b83817e5c6d6a039c9457336c30341a4c70b 100644 (file)
--- a/python/skytools/dbstruct.py
+++ b/python/skytools/dbstruct.py
@@ -74,7 +74,7 @@ class TConstraint(TElem):
     type = T_CONSTRAINT
     SQL = """
         SELECT conname as name, pg_get_constraintdef(oid) as def, contype
-          FROM pg_constraint WHERE conrelid = %(oid)s
+          FROM pg_constraint WHERE conrelid = %(oid)s AND contype != 'f'
     """
     def __init__(self, table_name, row):
         self.table_name = table_name

diff --git a/sql/londiste/Makefile b/sql/londiste/Makefile
index c3422c4e20a5f2b13ab997825f2e818dd519e48b..49f05beffcfccece8a00a8da5fdf281225d487c5 100644 (file)
--- a/sql/londiste/Makefile
+++ b/sql/londiste/Makefile
@@ -5,7 +5,7 @@ DOCS = README.londiste
 FUNCS = $(wildcard functions/*.sql)
 SRCS = structure/tables.sql structure/grants.sql structure/types.sql $(FUNCS)
 
-REGRESS = londiste_install londiste_provider londiste_subscriber
+REGRESS = londiste_install londiste_provider londiste_subscriber londiste_fkeys
 # londiste_denytrigger
 REGRESS_OPTS = --load-language=plpythonu --load-language=plpgsql
 

diff --git a/sql/londiste/expected/londiste_fkeys.out b/sql/londiste/expected/londiste_fkeys.out
new file mode 100644 (file)
index 0000000..76b8b3b
--- /dev/null
+++ b/sql/londiste/expected/londiste_fkeys.out
@@ -0,0 +1,203 @@
+create table ref_1 (
+    id int4 primary key,
+    val text
+);
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "ref_1_pkey" for table "ref_1"
+create table ref_2 (
+    id int4 primary key,
+    ref int4 not null references ref_1,
+    val text
+);
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "ref_2_pkey" for table "ref_2"
+create table ref_3 (
+    id int4 primary key,
+    ref2 int4 not null references ref_2,
+    val text
+);
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "ref_3_pkey" for table "ref_3"
+select * from londiste.subscriber_add_table('refqueue', 'public.ref_1');
+ subscriber_add_table 
+----------------------
+                    0
+(1 row)
+
+select * from londiste.subscriber_add_table('refqueue', 'public.ref_2');
+ subscriber_add_table 
+----------------------
+                    0
+(1 row)
+
+select * from londiste.subscriber_add_table('refqueue', 'public.ref_3');
+ subscriber_add_table 
+----------------------
+                    0
+(1 row)
+
+select * from londiste.find_table_fkeys('public.ref_1');
+  from_table  |   to_table   |   fkey_name    |                fkey_def                
+--------------+--------------+----------------+----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey | FOREIGN KEY (ref) REFERENCES ref_1(id)
+(1 row)
+
+select * from londiste.find_table_fkeys('public.ref_2');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey  | FOREIGN KEY (ref) REFERENCES ref_1(id)
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(2 rows)
+
+select * from londiste.find_table_fkeys('public.ref_3');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(1 row)
+
+select * from londiste.subscriber_get_table_pending_fkeys('public.ref_2');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+-- drop fkeys
+select * from londiste.subscriber_drop_table_fkey('public.ref_2', 'ref_2_ref_fkey');
+ subscriber_drop_table_fkey 
+----------------------------
+                          1
+(1 row)
+
+select * from londiste.find_table_fkeys('public.ref_1');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.find_table_fkeys('public.ref_2');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(1 row)
+
+select * from londiste.find_table_fkeys('public.ref_3');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(1 row)
+
+select * from londiste.subscriber_drop_table_fkey('public.ref_3', 'ref_3_ref2_fkey');
+ subscriber_drop_table_fkey 
+----------------------------
+                          1
+(1 row)
+
+-- check if dropped
+select * from londiste.find_table_fkeys('public.ref_1');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.find_table_fkeys('public.ref_2');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.find_table_fkeys('public.ref_3');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+-- look state
+select * from londiste.subscriber_get_table_pending_fkeys('public.ref_2');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey  | FOREIGN KEY (ref) REFERENCES ref_1(id)
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(2 rows)
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+-- toggle sync
+select * from londiste.subscriber_set_table_state('refqueue', 'public.ref_1', null, 'ok');
+ subscriber_set_table_state 
+----------------------------
+                          1
+(1 row)
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.subscriber_set_table_state('refqueue', 'public.ref_2', null, 'ok');
+ subscriber_set_table_state 
+----------------------------
+                          1
+(1 row)
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+  from_table  |   to_table   |   fkey_name    |                fkey_def                
+--------------+--------------+----------------+----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey | FOREIGN KEY (ref) REFERENCES ref_1(id)
+(1 row)
+
+select * from londiste.subscriber_set_table_state('refqueue', 'public.ref_3', null, 'ok');
+ subscriber_set_table_state 
+----------------------------
+                          1
+(1 row)
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey  | FOREIGN KEY (ref) REFERENCES ref_1(id)
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(2 rows)
+
+-- restore
+select * from londiste.subscriber_restore_table_fkey('public.ref_2', 'ref_2_ref_fkey');
+ subscriber_restore_table_fkey 
+-------------------------------
+                             1
+(1 row)
+
+select * from londiste.subscriber_restore_table_fkey('public.ref_3', 'ref_3_ref2_fkey');
+ subscriber_restore_table_fkey 
+-------------------------------
+                             1
+(1 row)
+
+-- look state
+select * from londiste.subscriber_get_table_pending_fkeys('public.ref_2');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+ from_table | to_table | fkey_name | fkey_def 
+------------+----------+-----------+----------
+(0 rows)
+
+select * from londiste.find_table_fkeys('public.ref_1');
+  from_table  |   to_table   |   fkey_name    |                fkey_def                
+--------------+--------------+----------------+----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey | FOREIGN KEY (ref) REFERENCES ref_1(id)
+(1 row)
+
+select * from londiste.find_table_fkeys('public.ref_2');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_2 | public.ref_1 | ref_2_ref_fkey  | FOREIGN KEY (ref) REFERENCES ref_1(id)
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(2 rows)
+
+select * from londiste.find_table_fkeys('public.ref_3');
+  from_table  |   to_table   |    fkey_name    |                fkey_def                 
+--------------+--------------+-----------------+-----------------------------------------
+ public.ref_3 | public.ref_2 | ref_3_ref2_fkey | FOREIGN KEY (ref2) REFERENCES ref_2(id)
+(1 row)
+

diff --git a/sql/londiste/expected/londiste_provider.out b/sql/londiste/expected/londiste_provider.out
index 1c081936534abeb2222508ce50d2d16907e51370..24533352295d5c258b11b0bae359175117524c6b 100644 (file)
--- a/sql/londiste/expected/londiste_provider.out
+++ b/sql/londiste/expected/londiste_provider.out
@@ -12,10 +12,10 @@ create table testdata_nopk (
 );
 select londiste.provider_add_table('pqueue', 'public.testdata_nopk');
 ERROR:  need key column
-CONTEXT:  PL/pgSQL function "provider_add_table" line 2 at return
+CONTEXT:  PL/pgSQL function "provider_add_table" line 2 at RETURN
 select londiste.provider_add_table('pqueue', 'public.testdata');
 ERROR:  no such event queue
-CONTEXT:  PL/pgSQL function "provider_add_table" line 2 at return
+CONTEXT:  PL/pgSQL function "provider_add_table" line 2 at RETURN
 select pgq.create_queue('pqueue');
  create_queue 
 --------------
@@ -29,10 +29,10 @@ select londiste.provider_add_table('pqueue', 'public.testdata');
 (1 row)
 
 select londiste.provider_add_table('pqueue', 'public.testdata');
-ERROR:  duplicate key violates unique constraint "provider_table_pkey"
+ERROR:  duplicate key value violates unique constraint "provider_table_pkey"
 CONTEXT:  SQL statement "INSERT INTO londiste.provider_table (queue_name, table_name, trigger_name) values ( $1 ,  $2 ,  $3 )"
 PL/pgSQL function "provider_add_table" line 23 at SQL statement
-PL/pgSQL function "provider_add_table" line 2 at return
+PL/pgSQL function "provider_add_table" line 2 at RETURN
 select londiste.provider_refresh_trigger('pqueue', 'public.testdata');
  provider_refresh_trigger 
 --------------------------
@@ -68,9 +68,9 @@ select * from londiste.provider_get_seq_list('pqueue');
 
 select londiste.provider_add_seq('pqueue', 'public.no_seq');
 ERROR:  seq not found
-CONTEXT:  PL/pgSQL function "find_seq_oid" line 2 at return
+CONTEXT:  PL/pgSQL function "find_seq_oid" line 2 at RETURN
 SQL statement "SELECT  1 from pg_class where oid = londiste.find_seq_oid( $1 )"
-PL/pgSQL function "provider_add_seq" line 10 at perform
+PL/pgSQL function "provider_add_seq" line 10 at PERFORM
 select londiste.provider_add_seq('pqueue', 'public.testdata_id_seq');
  provider_add_seq 
 ------------------
@@ -78,7 +78,7 @@ select londiste.provider_add_seq('pqueue', 'public.testdata_id_seq');
 (1 row)
 
 select londiste.provider_add_seq('pqueue', 'public.testdata_id_seq');
-ERROR:  duplicate key violates unique constraint "provider_seq_pkey"
+ERROR:  duplicate key value violates unique constraint "provider_seq_pkey"
 CONTEXT:  SQL statement "INSERT INTO londiste.provider_seq (queue_name, seq_name) values ( $1 ,  $2 )"
 PL/pgSQL function "provider_add_seq" line 16 at SQL statement
 select * from londiste.provider_get_seq_list('pqueue');
@@ -112,7 +112,7 @@ select londiste.provider_add_table('pqueue', 'public.testdata');
 insert into londiste.link (source, dest) values ('mqueue', 'pqueue');
 select londiste.provider_add_table('pqueue', 'public.testdata');
 ERROR:  Linked queue, manipulation not allowed
-CONTEXT:  PL/pgSQL function "provider_add_table" line 2 at return
+CONTEXT:  PL/pgSQL function "provider_add_table" line 2 at RETURN
 select londiste.provider_remove_table('pqueue', 'public.testdata');
 ERROR:  Linked queue, manipulation not allowed
 select londiste.provider_add_seq('pqueue', 'public.testdata_id_seq');

diff --git a/sql/londiste/expected/londiste_subscriber.out b/sql/londiste/expected/londiste_subscriber.out
index aa7d130f5b1a88c59de6f44582618056ec5c8e0a..a5dffac8a4702cfe9dfbdc74b40dc1247d03ca6f 100644 (file)
--- a/sql/londiste/expected/londiste_subscriber.out
+++ b/sql/londiste/expected/londiste_subscriber.out
@@ -25,11 +25,11 @@ select pgq.create_queue('pqueue');
 (1 row)
 
 select londiste.subscriber_add_table('pqueue', 'public.testdata');
-ERROR:  duplicate key violates unique constraint "subscriber_table_pkey"
+ERROR:  duplicate key value violates unique constraint "subscriber_table_pkey"
 CONTEXT:  SQL statement "INSERT INTO londiste.subscriber_table (queue_name, table_name) values ( $1 ,  $2 )"
 PL/pgSQL function "subscriber_add_table" line 2 at SQL statement
 select londiste.subscriber_add_table('pqueue', 'public.testdata');
-ERROR:  duplicate key violates unique constraint "subscriber_table_pkey"
+ERROR:  duplicate key value violates unique constraint "subscriber_table_pkey"
 CONTEXT:  SQL statement "INSERT INTO londiste.subscriber_table (queue_name, table_name) values ( $1 ,  $2 )"
 PL/pgSQL function "subscriber_add_table" line 2 at SQL statement
 select * from londiste.subscriber_get_table_list('pqueue');
@@ -74,7 +74,7 @@ select londiste.subscriber_add_seq('pqueue', 'public.testdata_id_seq');
 (1 row)
 
 select londiste.subscriber_add_seq('pqueue', 'public.testdata_id_seq');
-ERROR:  duplicate key violates unique constraint "subscriber_seq_pkey"
+ERROR:  duplicate key value violates unique constraint "subscriber_seq_pkey"
 CONTEXT:  SQL statement "INSERT INTO londiste.subscriber_seq (queue_name, seq_name) values ( $1 ,  $2 )"
 PL/pgSQL function "subscriber_add_seq" line 4 at SQL statement
 select * from londiste.subscriber_get_seq_list('pqueue');
@@ -109,7 +109,7 @@ select londiste.subscriber_add_table('pqueue', 'public.testdata');
 
 insert into londiste.link (source, dest) values ('mqueue', 'pqueue');
 select londiste.subscriber_add_table('pqueue', 'public.testdata');
-ERROR:  duplicate key violates unique constraint "subscriber_table_pkey"
+ERROR:  duplicate key value violates unique constraint "subscriber_table_pkey"
 CONTEXT:  SQL statement "INSERT INTO londiste.subscriber_table (queue_name, table_name) values ( $1 ,  $2 )"
 PL/pgSQL function "subscriber_add_table" line 2 at SQL statement
 select londiste.subscriber_remove_table('pqueue', 'public.testdata');

diff --git a/sql/londiste/functions/londiste.find_column_types.sql b/sql/londiste/functions/londiste.find_column_types.sql
index 52f8864acd0ed73f29a1f9552ba712e981ca7ea7..594bfdfecda593c4e7cb2a92bd3052f90e0dec00 100644 (file)
--- a/sql/londiste/functions/londiste.find_column_types.sql
+++ b/sql/londiste/functions/londiste.find_column_types.sql
@@ -22,5 +22,5 @@ begin
 
     return res;
 end;
-$$ language plpgsql;
+$$ language plpgsql strict stable;
 

diff --git a/sql/londiste/functions/londiste.find_table_fkeys.sql b/sql/londiste/functions/londiste.find_table_fkeys.sql
new file mode 100644 (file)
index 0000000..ce02ce7
--- /dev/null
+++ b/sql/londiste/functions/londiste.find_table_fkeys.sql
@@ -0,0 +1,26 @@
+
+create or replace function londiste.find_table_fkeys(i_table_name text)
+returns setof londiste.subscriber_pending_fkeys as $$
+declare
+    fkey      record;
+    tbl_oid   oid;
+begin
+    select londiste.find_table_oid(i_table_name) into tbl_oid;
+        
+    for fkey in
+        select n1.nspname || '.' || t1.relname as from_table, n2.nspname || '.' || t2.relname as to_table,
+            conname::text as fkey_name, pg_get_constraintdef(c.oid) as fkey_def
+        from pg_constraint c, pg_namespace n1, pg_class t1, pg_namespace n2, pg_class t2
+        where c.contype = 'f' and (c.conrelid = tbl_oid or c.confrelid = tbl_oid)
+            and t1.oid = c.conrelid and n1.oid = t1.relnamespace
+            and t2.oid = c.confrelid and n2.oid = t2.relnamespace
+        order by 1,2,3
+    loop
+        return next fkey;
+    end loop;
+    
+    return;
+end;
+$$ language plpgsql strict stable;
+
+

diff --git a/sql/londiste/functions/londiste.find_table_oid.sql b/sql/londiste/functions/londiste.find_table_oid.sql
index 907b71bab10cce6205e4436d58b9f2fd36cfc5c9..4d7da888d9c0c51f099dc923e017991f2bd7b4f6 100644 (file)
--- a/sql/londiste/functions/londiste.find_table_oid.sql
+++ b/sql/londiste/functions/londiste.find_table_oid.sql
@@ -31,19 +31,19 @@ begin
 
     return res;
 end;
-$$ language plpgsql;
+$$ language plpgsql strict stable;
 
 create or replace function londiste.find_table_oid(tbl text)
 returns oid as $$
 begin
     return londiste.find_rel_oid(tbl, 'r');
 end;
-$$ language plpgsql;
+$$ language plpgsql strict stable;
 
 create or replace function londiste.find_seq_oid(tbl text)
 returns oid as $$
 begin
     return londiste.find_rel_oid(tbl, 'S');
 end;
-$$ language plpgsql;
+$$ language plpgsql strict stable;
 

diff --git a/sql/londiste/functions/londiste.get_last_tick.sql b/sql/londiste/functions/londiste.get_last_tick.sql
index e50d6e9906d6f8f5ad8efe40e59de58d46a42ab8..8cfd9a809f57c6c2d7a8919f703c7dade6063e47 100644 (file)
--- a/sql/londiste/functions/londiste.get_last_tick.sql
+++ b/sql/londiste/functions/londiste.get_last_tick.sql
@@ -9,5 +9,5 @@ begin
      where consumer_id = i_consumer;
     return res;
 end;
-$$ language plpgsql security definer;
+$$ language plpgsql security definer strict stable;
 

diff --git a/sql/londiste/functions/londiste.subscriber_fkeys_funcs.sql b/sql/londiste/functions/londiste.subscriber_fkeys_funcs.sql
new file mode 100644 (file)
index 0000000..cad1eba
--- /dev/null
+++ b/sql/londiste/functions/londiste.subscriber_fkeys_funcs.sql
@@ -0,0 +1,88 @@
+
+
+create or replace function londiste.subscriber_get_table_pending_fkeys(i_table_name text) 
+returns setof londiste.subscriber_pending_fkeys as $$
+declare
+    fkeys   record;
+begin
+    for fkeys in
+        select *
+        from londiste.subscriber_pending_fkeys
+        where from_table=i_table_name or to_table=i_table_name
+        order by 1,2,3
+    loop
+        return next fkeys;
+    end loop;
+    
+    return;
+end;
+$$ language plpgsql;
+
+
+create or replace function londiste.subscriber_get_queue_valid_pending_fkeys(i_queue_name text)
+returns setof londiste.subscriber_pending_fkeys as $$
+declare
+    fkeys   record;
+begin
+    for fkeys in
+        select pf.*
+        from londiste.subscriber_pending_fkeys pf
+             left join londiste.subscriber_table st_from on (st_from.table_name = pf.from_table)
+             left join londiste.subscriber_table st_to on (st_to.table_name = pf.to_table)
+        where (st_from.table_name is null or (st_from.merge_state = 'ok' and st_from.snapshot is null))
+          and (st_to.table_name is null or (st_to.merge_state = 'ok' and st_to.snapshot is null))
+          and (coalesce(st_from.queue_name = i_queue_name, false)
+               or coalesce(st_to.queue_name = i_queue_name, false))
+        order by 1, 2, 3
+    loop
+        return next fkeys;
+    end loop;
+    
+    return;
+end;
+$$ language plpgsql;
+
+
+create or replace function londiste.subscriber_drop_table_fkey(i_from_table text, i_fkey_name text)
+returns integer as $$
+declare
+    fkey       record;
+begin        
+    select * into fkey
+    from londiste.find_table_fkeys(i_from_table) 
+    where fkey_name = i_fkey_name and from_table = i_from_table;
+    
+    if not found then
+        return 1;
+    end if;
+            
+    insert into londiste.subscriber_pending_fkeys values (fkey.from_table, fkey.to_table, i_fkey_name, fkey.fkey_def);
+        
+    execute 'alter table only ' || fkey.from_table || ' drop constraint ' || i_fkey_name || ';';
+    
+    return 1;
+end;
+$$ language plpgsql;
+
+
+create or replace function londiste.subscriber_restore_table_fkey(i_from_table text, i_fkey_name text)
+returns integer as $$
+declare
+    fkey    record;
+begin
+    select * into fkey
+    from londiste.subscriber_pending_fkeys 
+    where fkey_name = i_fkey_name and from_table = i_from_table;
+    
+    if not found then
+        return 1;
+    end if;
+    
+    delete from londiste.subscriber_pending_fkeys where fkey_name = fkey.fkey_name;
+        
+    execute 'alter table only ' || fkey.from_table || ' add constraint '
+        || fkey.fkey_name || ' ' || fkey.fkey_def || ';';
+        
+    return 1;
+end;
+$$ language plpgsql;

diff --git a/sql/londiste/sql/londiste_fkeys.sql b/sql/londiste/sql/londiste_fkeys.sql
new file mode 100644 (file)
index 0000000..3b9d8b2
--- /dev/null
+++ b/sql/londiste/sql/londiste_fkeys.sql
@@ -0,0 +1,72 @@
+
+set log_error_verbosity = 'terse';
+
+create table ref_1 (
+    id int4 primary key,
+    val text
+);
+
+create table ref_2 (
+    id int4 primary key,
+    ref int4 not null references ref_1,
+    val text
+);
+
+create table ref_3 (
+    id int4 primary key,
+    ref2 int4 not null references ref_2,
+    val text
+);
+
+select * from londiste.subscriber_add_table('refqueue', 'public.ref_1');
+select * from londiste.subscriber_add_table('refqueue', 'public.ref_2');
+select * from londiste.subscriber_add_table('refqueue', 'public.ref_3');
+
+select * from londiste.find_table_fkeys('public.ref_1');
+select * from londiste.find_table_fkeys('public.ref_2');
+select * from londiste.find_table_fkeys('public.ref_3');
+
+select * from londiste.subscriber_get_table_pending_fkeys('public.ref_2');
+
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+
+-- drop fkeys
+
+select * from londiste.subscriber_drop_table_fkey('public.ref_2', 'ref_2_ref_fkey');
+
+select * from londiste.find_table_fkeys('public.ref_1');
+select * from londiste.find_table_fkeys('public.ref_2');
+select * from londiste.find_table_fkeys('public.ref_3');
+
+select * from londiste.subscriber_drop_table_fkey('public.ref_3', 'ref_3_ref2_fkey');
+
+-- check if dropped
+
+select * from londiste.find_table_fkeys('public.ref_1');
+select * from londiste.find_table_fkeys('public.ref_2');
+select * from londiste.find_table_fkeys('public.ref_3');
+
+-- look state
+select * from londiste.subscriber_get_table_pending_fkeys('public.ref_2');
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+
+-- toggle sync
+select * from londiste.subscriber_set_table_state('refqueue', 'public.ref_1', null, 'ok');
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+select * from londiste.subscriber_set_table_state('refqueue', 'public.ref_2', null, 'ok');
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+select * from londiste.subscriber_set_table_state('refqueue', 'public.ref_3', null, 'ok');
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+
+-- restore
+select * from londiste.subscriber_restore_table_fkey('public.ref_2', 'ref_2_ref_fkey');
+select * from londiste.subscriber_restore_table_fkey('public.ref_3', 'ref_3_ref2_fkey');
+
+-- look state
+select * from londiste.subscriber_get_table_pending_fkeys('public.ref_2');
+select * from londiste.subscriber_get_queue_valid_pending_fkeys('refqueue');
+select * from londiste.find_table_fkeys('public.ref_1');
+select * from londiste.find_table_fkeys('public.ref_2');
+select * from londiste.find_table_fkeys('public.ref_3');
+
+

diff --git a/sql/londiste/structure/tables.sql b/sql/londiste/structure/tables.sql
index 6e7839836a220408c5c5b6dc0ec8aa2578a375f0..3b1d31e61fe29d71f8fafd3763056f6241f87459 100644 (file)
--- a/sql/londiste/structure/tables.sql
+++ b/sql/londiste/structure/tables.sql
@@ -58,7 +58,6 @@ create table londiste.subscriber_pending_fkeys(
     fkey_name           text not null,
     fkey_def            text not null,
     
-    primary key (from_table, to_table, fkey_name),
-    unique (to_table, from_table, fkey_name)
+    primary key (from_table, fkey_name)
 );
 

diff --git a/tests/londiste/data.sql b/tests/londiste/data.sql
index 5fe0ae9af33ece01dd317f6c51f3fe2cf658f36e..d1373e0c24c30093c502528162b6b4dd295fa215 100644 (file)
--- a/tests/londiste/data.sql
+++ b/tests/londiste/data.sql
@@ -13,7 +13,7 @@ create index idx_data1_rand on data1 (id, data);
 create table data2 (
     id serial primary key,
     data text,
-    -- old_id integer references data1,
+    ref1 integer references data1,
     constraint uq_data2 unique (data)
 );