triggers kinda work

11 files changed, 301 insertions, 36 deletions

diff --git a/sql/londiste/expected/londiste_fkeys.out b/sql/londiste/expected/londiste_fkeys.out
index 26a4adc1..5c2a0cfb 100644
--- a/sql/londiste/expected/londiste_fkeys.out
+++ b/sql/londiste/expected/londiste_fkeys.out
@@ -35,21 +35,21 @@ select * from londiste.set_add_table('branch_set', 'public.ref_3');
 (1 row)
 
 select * from londiste.node_add_table('branch_set', 'public.ref_1');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |         ret_desc          
+----------+---------------------------
+      200 | Table added: public.ref_1
 (1 row)
 
 select * from londiste.node_add_table('branch_set', 'public.ref_2');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |         ret_desc          
+----------+---------------------------
+      200 | Table added: public.ref_2
 (1 row)
 
 select * from londiste.node_add_table('branch_set', 'public.ref_3');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |         ret_desc          
+----------+---------------------------
+      200 | Table added: public.ref_3
 (1 row)
 
 select * from londiste.find_table_fkeys('public.ref_1');
diff --git a/sql/londiste/expected/londiste_provider.out b/sql/londiste/expected/londiste_provider.out
index 727a3e04..2fe6c83b 100644
--- a/sql/londiste/expected/londiste_provider.out
+++ b/sql/londiste/expected/londiste_provider.out
@@ -30,11 +30,15 @@ select * from pgq_set.create_node('aset', 'root', 'rnode', 'londiste_root', null
 (1 row)
 
 select * from londiste.node_add_table('aset', 'public.testdata_nopk');
-ERROR:  need key column
+ ret_code |                      ret_desc                      
+----------+----------------------------------------------------
+      400 | Primary key missing on table: public.testdata_nopk
+(1 row)
+
 select * from londiste.node_add_table('aset', 'public.testdata');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |           ret_desc           
+----------+------------------------------
+      200 | Table added: public.testdata
 (1 row)
 
 select * from londiste.node_get_table_list('aset');
@@ -44,9 +48,9 @@ select * from londiste.node_get_table_list('aset');
 (1 row)
 
 select * from londiste.node_remove_table('aset', 'public.testdata');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |            ret_desc            
+----------+--------------------------------
+      200 | Table removed: public.testdata
 (1 row)
 
 select * from londiste.node_remove_table('aset', 'public.testdata');
diff --git a/sql/londiste/expected/londiste_subscriber.out b/sql/londiste/expected/londiste_subscriber.out
index b34a835e..b09399ae 100644
--- a/sql/londiste/expected/londiste_subscriber.out
+++ b/sql/londiste/expected/londiste_subscriber.out
@@ -44,9 +44,9 @@ select * from londiste.set_add_table('branch_set', 'public.slavedata');
 (1 row)
 
 select * from londiste.node_add_table('branch_set', 'public.slavedata');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |           ret_desc            
+----------+-------------------------------
+      200 | Table added: public.slavedata
 (1 row)
 
 select * from londiste.node_get_table_list('branch_set');
@@ -56,9 +56,9 @@ select * from londiste.node_get_table_list('branch_set');
 (1 row)
 
 select * from londiste.node_remove_table('branch_set', 'public.slavedata');
- ret_code | ret_desc 
-----------+----------
-      200 | OK
+ ret_code |            ret_desc             
+----------+---------------------------------
+      200 | Table removed: public.slavedata
 (1 row)
 
 select * from londiste.node_remove_table('branch_set', 'public.slavedata');
diff --git a/sql/londiste/functions/londiste.node_add_table.sql b/sql/londiste/functions/londiste.node_add_table.sql
index 09865a87..d3b3d82f 100644
--- a/sql/londiste/functions/londiste.node_add_table.sql
+++ b/sql/londiste/functions/londiste.node_add_table.sql
@@ -20,7 +20,8 @@ begin
     fq_table_name := londiste.make_fqname(i_table_name);
     col_types := londiste.find_column_types(fq_table_name);
     if position('k' in col_types) < 1 then
-        raise exception 'need key column';
+        select 400, 'Primary key missing on table: ' || fq_table_name into ret_code, ret_desc;
+        return;
     end if;
 
     perform 1 from pgq_set.set_info where set_name = i_set_name;
@@ -36,7 +37,11 @@ begin
     end if;
 
     if pgq_set.is_root(i_set_name) then
-        perform londiste.set_add_table(i_set_name, fq_table_name);
+        select * into ret_code, ret_desc
+            from londiste.set_add_table(i_set_name, fq_table_name);
+        if ret_code <> 200 then
+            return;
+        end if;
     else
         perform 1 from londiste.set_table where set_name = i_set_name and table_name = fq_table_name;
         if not found then
@@ -45,16 +50,27 @@ begin
         end if;
     end if;
 
-    if pgq_set.is_root(i_set_name) then
-        select * into ret_code, ret_desc
-            from londiste.set_add_table(i_set_name, fq_table_name);
-        if ret_code <> 200 then
+    insert into londiste.node_table (set_name, table_name)
+        values (i_set_name, fq_table_name);
+
+    for ret_code, ret_desc in
+        select f.ret_code, f.ret_desc
+        from londiste.node_prepare_triggers(i_set_name, fq_table_name) f
+    loop
+        if ret_code > 299 then
             return;
         end if;
-    end if;
+    end loop;
+
+    for ret_code, ret_desc in
+        select f.ret_code, f.ret_desc
+        from londiste.node_refresh_triggers(i_set_name, fq_table_name) f
+    loop
+        if ret_code > 299 then
+            return;
+        end if;
+    end loop;
 
-    insert into londiste.node_table (set_name, table_name)
-        values (i_set_name, fq_table_name);
     select 200, 'Table added: ' || fq_table_name into ret_code, ret_desc;
     return;
 end;
diff --git a/sql/londiste/functions/londiste.node_disable_triggers.sql b/sql/londiste/functions/londiste.node_disable_triggers.sql
new file mode 100644
index 00000000..df5a0759
--- /dev/null
+++ b/sql/londiste/functions/londiste.node_disable_triggers.sql
@@ -0,0 +1,77 @@
+
+create or replace function londiste.node_disable_triggers(
+    in i_set_name   text,
+    in i_table_name text,
+    out ret_code    int4,
+    out ret_desc    text)
+returns setof record strict as $$
+-- ----------------------------------------------------------------------
+-- Function: londiste.node_disable_triggers(2)
+--
+--      Drop all registered triggers from particular table.
+-- ----------------------------------------------------------------------
+declare
+    tbl_oid oid;
+    fq_table_name text;
+    tg record;
+    is_active int4;
+begin
+    fq_table_name := londiste.make_fqname(i_table_name);
+    perform 1 from pgq_set.set_info where set_name = i_set_name;
+    if not found then
+        select 400, 'Unknown set: ' || i_set_name;
+        return next;
+        return;
+    end if;
+    tbl_oid := londiste.find_table_oid(fq_table_name);
+    for tg in
+        select tg_name, tg_type, tg_def from londiste.node_trigger
+         where set_name = i_set_name and table_name = fq_table_name
+         order by tg_name
+    loop
+        -- check if active
+        perform 1 from pg_catalog.pg_trigger
+         where tgrelid = tbl_oid
+           and tgname = tg.tg_name;
+        if found then
+            execute 'drop trigger ' || quote_ident(tg.tg_name)
+                || ' on ' || londiste.quote_fqname(fq_table_name);
+            select 200, 'Dropped trigger ' || tg.tg_name
+                || ' from table ' || fq_table_name
+                into ret_code, ret_desc;
+                return next;
+        end if;
+    end loop;
+    return;
+end;
+$$ language plpgsql security definer;
+
+create or replace function londiste.node_disable_triggers(
+    in i_set_name   text,
+    out ret_code    int4,
+    out ret_desc    text)
+returns setof record strict as $$
+-- ----------------------------------------------------------------------
+-- Function: londiste.node_disable_triggers(1)
+--
+--      Drop all registered triggers from set tables.
+-- ----------------------------------------------------------------------
+declare
+    t record;
+begin
+    for t in
+        select table_name from londiste.node_table
+         where set_name = i_set_name
+         order by nr
+    loop
+        for ret_code, ret_desc in
+            select f.ret_code, f.ret_desc
+                from londiste.node_disable_triggers(i_set_name, t.table_name) f
+        loop
+            return next;
+        end loop;
+    end loop;
+    return;
+end;
+$$ language plpgsql security definer;
+
diff --git a/sql/londiste/functions/londiste.node_prepare_triggers.sql b/sql/londiste/functions/londiste.node_prepare_triggers.sql
new file mode 100644
index 00000000..8e658fa8
--- /dev/null
+++ b/sql/londiste/functions/londiste.node_prepare_triggers.sql
@@ -0,0 +1,50 @@
+
+create or replace function londiste.node_prepare_triggers(
+    in i_set_name   text,
+    in i_table_name text,
+    out ret_code    int4,
+    out ret_desc    text)
+returns setof record strict as $$
+-- ----------------------------------------------------------------------
+-- Function: londiste.node_prepare_triggers(2)
+--
+--      Regsiter Londiste trigger for table.
+-- ----------------------------------------------------------------------
+declare
+    t_name   text;
+    logtrg   text;
+    denytrg  text;
+    logtrg_name text;
+    denytrg_name text;
+    qname    text;
+    fq_table_name text;
+begin
+    fq_table_name := londiste.make_fqname(i_table_name);
+    select queue_name into qname from pgq_set.set_info where set_name = i_set_name;
+    if not found then
+        select 400, 'Set not found: ' || i_set_name into ret_code, ret_desc;
+        return next;
+        return;
+    end if;
+    logtrg_name := i_set_name || '_logtrigger';
+    denytrg_name := i_set_name || '_denytrigger';
+    logtrg := 'create trigger ' || quote_ident(logtrg_name)
+        || ' after insert or update or delete on ' || londiste.quote_fqname(fq_table_name)
+        || ' for each row execute procedure pgq.sqltriga(' || quote_literal(qname) || ')';
+    insert into londiste.node_trigger (set_name, table_name, tg_name, tg_type, tg_def)
+    values (i_set_name, fq_table_name, logtrg_name, 'root', logtrg);
+    select 200, logtrg into ret_code, ret_desc;
+    return next;
+
+    denytrg := 'create trigger ' || quote_ident(denytrg_name)
+        || ' before insert or update or delete on ' || londiste.quote_fqname(fq_table_name)
+        || ' for each row execute procedure pgq.denytriga(' || quote_literal(qname) || ')';
+    insert into londiste.node_trigger (set_name, table_name, tg_name, tg_type, tg_def)
+    values (i_set_name, fq_table_name, denytrg_name, 'non-root', denytrg);
+    select 200, denytrg into ret_code, ret_desc;
+    return next;
+
+    return;
+end;
+$$ language plpgsql security definer;
+
diff --git a/sql/londiste/functions/londiste.node_refresh_triggers.sql b/sql/londiste/functions/londiste.node_refresh_triggers.sql
new file mode 100644
index 00000000..76da2d22
--- /dev/null
+++ b/sql/londiste/functions/londiste.node_refresh_triggers.sql
@@ -0,0 +1,102 @@
+
+create or replace function londiste.node_refresh_triggers(
+    in i_set_name   text,
+    in i_table_name text,
+    out ret_code    int4,
+    out ret_desc    text)
+returns setof record strict as $$
+-- ----------------------------------------------------------------------
+-- Function: londiste.node_refresh_triggers(2)
+--
+--      Sync actual trigger state with registered triggers.
+-- ----------------------------------------------------------------------
+declare
+    tbl_oid oid;
+    fq_table_name text;
+    tg record;
+    is_root bool;
+    is_active int4;
+begin
+    fq_table_name := londiste.make_fqname(i_table_name);
+    perform 1 from pgq_set.set_info where set_name = i_set_name;
+    if not found then
+        select 400, 'Unknown set: ' || i_set_name;
+        return next;
+        return;
+    end if;
+    is_root := pgq_set.is_root(i_set_name);
+    tbl_oid := londiste.find_table_oid(fq_table_name);
+    for tg in
+        select tg_name, tg_type, tg_def from londiste.node_trigger
+         where set_name = i_set_name and table_name = fq_table_name
+         order by tg_name
+    loop
+        if tg.tg_type not in ('root', 'non-root') then
+            select 400, 'trigger ' || tg.tg_name
+                || ' on table ' || fq_table_name
+                || ' had unsupported type: ' || tg.tg_type
+                into ret_code, ret_desc;
+            return next;
+        else
+            -- check if active
+            select count(1) into is_active
+              from pg_catalog.pg_trigger
+             where tgrelid = tbl_oid
+               and tgname = tg.tg_name;
+
+            -- create or drop if needed
+            if (tg.tg_type = 'root') = is_root then
+                -- trigger must be active
+                if is_active = 0 then
+                    execute tg.tg_def;
+                    select 200, 'Created trigger ' || tg.tg_name
+                        || ' on table ' || fq_table_name
+                        into ret_code, ret_desc;
+                    return next;
+                end if;
+            else
+                -- trigger must be dropped
+                if is_active = 1 then
+                    execute 'drop trigger ' || quote_ident(tg.tg_name)
+                        || ' on ' || londiste.quote_fqname(fq_table_name);
+                    select 200, 'Dropped trigger ' || tg.tg_name
+                        || ' from table ' || fq_table_name
+                        into ret_code, ret_desc;
+                    return next;
+                end if;
+            end if;
+        end if;
+    end loop;
+    return;
+end;
+$$ language plpgsql security definer;
+
+create or replace function londiste.node_refresh_triggers(
+    in i_set_name   text,
+    out ret_code    int4,
+    out ret_desc    text)
+returns setof record strict as $$
+-- ----------------------------------------------------------------------
+-- Function: londiste.node_refresh_triggers(2)
+--
+--      Sync actual trigger state with registered triggers for all tables.
+-- ----------------------------------------------------------------------
+declare
+    t record;
+begin
+    for t in
+        select table_name from londiste.node_table
+         where set_name = i_set_name
+         order by nr
+    loop
+        for ret_code, ret_desc in
+            select f.ret_code, f.ret_desc
+                from londiste.node_refresh_triggers(i_set_name, t.table_name) f
+        loop
+            return next;
+        end loop;
+    end loop;
+    return;
+end;
+$$ language plpgsql security definer;
+
diff --git a/sql/londiste/functions/londiste.node_remove_table.sql b/sql/londiste/functions/londiste.node_remove_table.sql
index ae350b06..b4875b12 100644
--- a/sql/londiste/functions/londiste.node_remove_table.sql
+++ b/sql/londiste/functions/londiste.node_remove_table.sql
@@ -7,6 +7,17 @@ declare
     fq_table_name text;
 begin
     fq_table_name := londiste.make_fqname(i_table_name);
+
+    for ret_code, ret_desc in
+        select f.ret_code, f.ret_desc from londiste.node_disable_triggers(i_set_name, fq_table_name) f
+    loop
+        if ret_code > 299 then
+            return;
+        end if;
+    end loop;
+    delete from londiste.node_trigger
+        where set_name = i_set_name
+          and table_name = fq_table_name;
     delete from londiste.node_table
         where set_name = i_set_name
           and table_name = fq_table_name;
diff --git a/sql/londiste/sql/londiste_install.sql b/sql/londiste/sql/londiste_install.sql
index 9bd55963..efb4f23e 100644
--- a/sql/londiste/sql/londiste_install.sql
+++ b/sql/londiste/sql/londiste_install.sql
@@ -3,6 +3,8 @@ set log_error_verbosity = 'terse';
 \i ../txid/txid.sql
 \i ../pgq/pgq.sql
 \i ../pgq_set/pgq_set.sql
-\i londiste.sql
+--\i londiste.sql
+\i structure/tables.sql
+\i structure/functions.sql
 \set ECHO all
 
diff --git a/sql/londiste/structure/functions.sql b/sql/londiste/structure/functions.sql
index 06facfdf..bb2d8bb0 100644
--- a/sql/londiste/structure/functions.sql
+++ b/sql/londiste/structure/functions.sql
@@ -21,8 +21,9 @@
 
 -- Group: Internal functions
 \i functions/londiste.node_set_skip_truncate.sql
--- \i functions/londiste.node_create_trigger.sql
--- \i functions/londiste.node_refresh_trigger.sql
+\i functions/londiste.node_prepare_triggers.sql
+\i functions/londiste.node_refresh_triggers.sql
+\i functions/londiste.node_disable_triggers.sql
 -- \i functions/londiste.node_notify_change.sql
 
 -- Group: Utility functions
diff --git a/sql/londiste/structure/tables.sql b/sql/londiste/structure/tables.sql
index 3b2e8533..b511b58b 100644
--- a/sql/londiste/structure/tables.sql
+++ b/sql/londiste/structure/tables.sql
@@ -111,7 +111,7 @@ create table londiste.node_table (
 -- Columns:
 --      set_name        - set it belongs to
 --      table_name      - table name
---      tg_type         - any / root / non-root
+--      tg_type         - any / root / non-root / unknown?
 --      tg_name         - name for the trigger
 --      tg_def          - full statement for trigger creation
 -- ----------------------------------------------------------------------
@@ -122,7 +122,9 @@ create table londiste.node_trigger (
     tg_type             text not null,
     tg_def              text not null,
     foreign key (set_name, table_name) references londiste.node_table,
-    primary key (set_name, table_name, tg_name)
+    primary key (set_name, table_name, tg_name),
+    check (tg_type in ('root', 'non-root'))
+    -- check (tg_type in ('always', 'origin', 'replica', 'disabled'))
 );
 
 -- ----------------------------------------------------------------------