Safely Handling Inserts/Updates with Triggers

[jessehanley]

Hey everyone —

There's a few threads that have been closed around handling inserts/updates whilst we're preparing the table.

We handled this with a very simple trigger:

  # Enable dual-write from original table to intermediate via triggers.
  # This is optional; pgslice normally uses fill + swapped fill without dual-writing.
  # We mirror INSERT/UPDATE/DELETE into the intermediate table so swap has near-zero delta.

  def enable_mirroring_triggers(table: 'events')
    intermediate = "#{table}_intermediate"
    function = "#{table}_mirror_to_intermediate"
    trigger  = "#{table}_mirror_trigger"

    sql = <<~SQL
      DO $$
      BEGIN
        -- create function if not exists
        IF NOT EXISTS (
          SELECT 1 FROM pg_proc p
          JOIN pg_namespace n ON n.oid = p.pronamespace
          WHERE p.proname = '#{function}' AND n.nspname = 'public'
        ) THEN
          EXECUTE $$
            CREATE OR REPLACE FUNCTION public.#{function}()
            RETURNS trigger
            LANGUAGE plpgsql
            AS $$
            BEGIN
              IF TG_OP = 'INSERT' THEN
                INSERT INTO public.#{intermediate} SELECT (NEW).*;
              ELSIF TG_OP = 'UPDATE' THEN
                UPDATE public.#{intermediate} i SET
                  (#{mirror_column_list(table)}) = (#{mirror_new_tuple_list(table)})
                WHERE i.id = NEW.id;
              ELSIF TG_OP = 'DELETE' THEN
                DELETE FROM public.#{intermediate} WHERE id = OLD.id;
              END IF;
              RETURN NULL;
            END;
            $$;
          $$;
        END IF;

        -- create trigger if not exists
        IF NOT EXISTS (
          SELECT 1 FROM pg_trigger WHERE tgname = '#{trigger}'
        ) THEN
          EXECUTE $$
            CREATE TRIGGER #{trigger}
            AFTER INSERT OR UPDATE OR DELETE ON public.#{table}
            FOR EACH ROW EXECUTE FUNCTION public.#{function}();
          $$;
        END IF;
      END$$;
    SQL

    ActiveRecord::Base.connection.execute(sql)
  end

And then after we filled everything up we can then:

# Disable and drop mirroring trigger and function
  def disable_mirroring_triggers(table: 'events')
    function = "#{table}_mirror_to_intermediate"
    trigger  = "#{table}_mirror_trigger"

    sql = <<~SQL
      DO $$
      BEGIN
        IF EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = '#{trigger}') THEN
          EXECUTE 'DROP TRIGGER ' || quote_ident('#{trigger}') || ' ON public.#{table}';
        END IF;
        IF EXISTS (
          SELECT 1 FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
          WHERE p.proname = '#{function}' AND n.nspname = 'public'
        ) THEN
          EXECUTE 'DROP FUNCTION public.#{function}()';
        END IF;
      END$$;
    SQL

    ActiveRecord::Base.connection.execute(sql)
  end

Hope this helps others who are trying to partition live with an active database :)

[jessehanley]

Resolves #48 — maybe something worth merging in? A --live flag or something that adds the trigger?

For us, we needed to do this because we were actively batch reindexing into Elastic/Searchkick and didn't want any IDs missing when we did the switch.

[time-less-ness]

Is there any intention to merge this to main? It could be another sub-command if we worry it's not widely applicable.