Added more examples and pgsql reader

examples/pgsql_meta_upgrade/migration_run.sql

@@ -0,0 +1,199 @@
+select * from dropall('migrations_run','meta');
+CREATE OR REPLACE FUNCTION meta.migration_run(
+   p_id_migration_model integer default null
+  ,p_priority_start integer default 0
+  ,p_priority_end integer default 10000
+  ,INOUT p_info jsonb default null
+  ,OUT p_retval integer
+  ,OUT p_errmsg text
+)
+LANGUAGE plpgsql VOLATILE
+SECURITY DEFINER 
+AS
+$$
+DECLARE
+  --Error Handling--
+  m_funcname text = 'meta.migration_run';
+  m_errmsg text;
+  m_errcontext text;
+  m_errdetail text;
+  m_errhint text;
+  m_errstate text;
+  m_retval integer;
+  --Error Handling--
+  m_dblink_conn TEXT;
+  m_dblink_login TEXT;
+  m_gotemplate text;
+  m_pid integer;
+  r_lp record;
+  m_errcnt integer;
+  G_BREAK_ERR_CNT constant integer = 500;
+BEGIN
+  m_dblink_conn = 'dblink_migration';
+  p_retval = 0;
+  p_errmsg = '';
+  m_pid = pg_backend_pid();
+  p_info = jsonb_build_array();
+
+  m_gotemplate = $CQl$do $OUTERDO$
+declare
+  m_lock_timeout text = '1500ms';
+  m_max_attempts int = 2;
+  m_ddl_completed boolean = false;
+begin
+  set local lock_timeout = '1500ms';
+  perform set_config('lock_timeout', m_lock_timeout, false);
+
+  for i in 1..m_max_attempts loop
+    begin
+      execute $EXECMIGRATION$[proc]$EXECMIGRATION$;
+      m_ddl_completed = true;
+      exit;
+    exception when lock_not_available then
+      raise warning 'attempt %/% to lock table "test" failed', i, m_max_attempts;
+
+      perform pg_sleep(0.1);
+    end;
+--     if i > 1
+--     then
+--       SET statement_timeout = 20000;
+--     end if;
+  end loop;
+end
+$OUTERDO$;
+ $CQl$;
+
+
+  select 'host=127.0.0.1 dbname='||r.dbname||' port=' || r.port || ' user='|| r.loginuser ||' password=' || r.password
+  from f_get_local_conn('migration') r
+  into m_dblink_login;
+
+  if not exists (SELECT T.text from unnest(dblink_get_connections()) T where T.text = m_dblink_conn)
+  then
+    PERFORM dblink_connect(m_dblink_conn, m_dblink_login);
+  end if;
+
+  perform log_event(m_funcname,format('migrations_run init [%s to %s]  with connection %s',p_priority_start,p_priority_end,m_dblink_conn),bt_enum('eventlog','upgrade'));
+
+  PERFORM dblink_exec(m_dblink_conn, format($S$set application_name to 'MIGRATION:%s'; $S$, m_pid), false);
+
+  for r_lp in (
+    select *
+      ,''::text as result
+      ,clock_timestamp() as startat
+
+    from meta.migration_script s
+    where coalesce(s.status,0) = 0
+      and s.priority >= p_priority_start
+      and s.priority <= p_priority_end
+      and octet_length(s.body) > 1
+    order by s.priority, s.sequence, s.id_migration_script
+  )
+  loop
+      if m_errcnt > G_BREAK_ERR_CNT
+      then
+        perform log_event(m_funcname,format('migrations_run Max Error Count [%s to %s] Errors %s/s%',p_priority_start,p_priority_end,m_errcnt,G_BREAK_ERR_CNT),bt_enum('eventlog','upgrade'));
+
+        exit;
+      end if;
+
+      perform dblink_exec(m_dblink_conn, format($Q$ DO $OUTERDO$ BEGIN
+
+        if not meta.f_can_migrate()
+        then
+          raise exception 'Cannot migrate. Check the f_can_migrate function.';
+        end if;
+
+        update meta.migration_script u
+        set status = 1
+          ,changedat = now()
+        where u.id_migration_script = %1$s;
+
+        END; $OUTERDO$; $Q$, r_lp.id_migration_script), true)
+      ;
+
+      begin
+
+        --raise notice 'Lp %: %',r_lp.objectname, r_lp.result;
+
+        --raise notice 'Executing % %  [%,%]', r_lp.objectname , r_lp.objecttype, r_lp.priority, r_lp.sequence;
+
+        select dblink_exec(m_dblink_conn, replace(replace(replace(m_gotemplate,'[proc]', r_lp.body),'[objectname]',r_lp.objectname),'[priority]',r_lp.priority::text), true)
+        into r_lp.result;
+
+        r_lp.duration = clock_timestamp() - r_lp.startat;
+
+        perform log_event(m_funcname,format('Ran Script %s @%s ',r_lp.objectname,r_lp.priority),bt_enum('eventlog','upgrade'));
+
+
+        perform dblink_exec(m_dblink_conn, format($Q$
+          update meta.migration_script u
+          set status = 2
+            ,changedat = now()
+            ,duration = %2$s::interval
+          where u.id_migration_script = %1$s;
+        $Q$, r_lp.id_migration_script,  quote_literal(  r_lp.duration)), true)
+        ;
+
+        p_info = jsonb_concat(p_info, jsonb_build_object('objectname',r_lp.objectname,'objecttype', r_lp.objecttype,'status','success'));
+
+
+      EXCEPTION
+        WHEN others THEN
+          GET STACKED DIAGNOSTICS
+            m_errmsg = MESSAGE_TEXT;
+
+          raise notice 'Exception % -> %',r_lp.objectname, m_errmsg;
+          r_lp.duration = clock_timestamp() - r_lp.startat;
+          perform log_event(m_funcname,format('migrations_run error %s -> %s >> %s',r_lp.objecttype,r_lp.objectname,m_errmsg),bt_enum('eventlog','local error'));
+
+          perform dblink_exec(m_dblink_conn, format($Q$
+            update meta.migration_script u
+            set status = 3
+              ,error =  %2$s || E' \nResult: ' ||  %3$s
+              ,changedat = now()
+              ,duration = %4$s::interval
+            where u.id_migration_script =  %1$s;
+            $Q$, r_lp.id_migration_script, quote_literal(m_errmsg), quote_literal(coalesce(r_lp.result,'')), quote_literal(  r_lp.duration)  ), false);
+
+          p_info = jsonb_concat(p_info, jsonb_build_object('objectname',r_lp.objectname,'objecttype', r_lp.objecttype,'status','error', 'error',m_errmsg));
+
+          m_errcnt = coalesce(m_errcnt,0) + 1;
+
+      end;
+
+      perform dblink_exec(m_dblink_conn, format($Q$DO $DDD$
+          declare
+            m_text text;
+          begin
+            select pg_notify('upgrade.events', json_build_object('type','upgrade','status',m.status,'objecttype', m.objecttype,'objectname',m.objectname)::text)
+            from meta.migration_script m
+            where m.id_migration_script = %1$s
+            into m_text;
+          end;
+          $DDD$;$Q$, r_lp.id_migration_script, quote_literal(m_errmsg), quote_literal(coalesce(r_lp.result,''))  ), false);
+
+  end loop;
+
+  perform dblink_disconnect(m_dblink_conn);
+
+EXCEPTION
+WHEN others THEN
+  GET STACKED DIAGNOSTICS
+    m_errmsg = MESSAGE_TEXT
+   ,m_errcontext = PG_EXCEPTION_CONTEXT
+   ,m_errdetail = PG_EXCEPTION_DETAIL
+   ,m_errhint = PG_EXCEPTION_HINT
+   ,m_errstate = RETURNED_SQLSTATE;
+  
+  p_retval = 1;
+  p_errmsg = format('%s Context %s State: %s', m_errmsg, m_errcontext, m_errstate);
+
+  if exists (SELECT T.text from unnest(dblink_get_connections()) T where T.text = m_dblink_conn)
+  then
+    perform dblink_disconnect(m_dblink_conn);
+  end if;
+
+  perform pg_notify('upgrade.events', json_build_object('type','upgrade','status',3,'error',m_errmsg,'objecttype',m_funcname)::text);
+END;
+$$;