text: /* pg_fact_loader--1.0.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  CREATE FUNCTION fact_loader._launch_worker(oid)
    RETURNS pg_catalog.INT4 STRICT
  AS 'MODULE_PATHNAME', 'pg_fact_loader_worker'
  LANGUAGE C;
  
  CREATE FUNCTION fact_loader.launch_worker()
    RETURNS pg_catalog.INT4 STRICT
  AS 'SELECT fact_loader._launch_worker(oid) FROM pg_database WHERE datname = current_database();'
  LANGUAGE SQL;
  
  CREATE TABLE fact_loader.fact_tables (
    fact_table_id SERIAL PRIMARY KEY,
    fact_table_relid REGCLASS NOT NULL,
    fact_table_agg_proid REGPROC NULL, --This may only be used to generate a merge function but is not used in automation
    enabled BOOLEAN NOT NULL DEFAULT FALSE,
    priority INT,
    attempt_number INT,
    retries_allowed INT DEFAULT 0,
    force_worker_priority BOOLEAN NOT NULL DEFAULT FALSE,
    last_refresh_source_cutoff TIMESTAMPTZ,
    last_refresh_attempted_at TIMESTAMPTZ,
    --TODO - answer if we want the worker to bail or record messages on ERROR (or both)
    last_refresh_succeeded BOOLEAN,
    row_created_at TIMESTAMPTZ DEFAULT NOW(),
    row_updated_at TIMESTAMPTZ,
    CONSTRAINT unique_fact_tables UNIQUE (fact_table_relid)
  );
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.fact_tables', '');
  
  CREATE TABLE fact_loader.fact_table_deps (
    fact_table_dep_id SERIAL PRIMARY KEY,
    parent_id INT NOT NULL REFERENCES fact_loader.fact_tables (fact_table_id),
    child_id INT NOT NULL REFERENCES fact_loader.fact_tables (fact_table_id),
    /*****
    In very many cases, you will use the same procs for insert, update, and delete
    even with multiple dependencies.  This is why you must give defaults here which
    will be used to auto-populate fact_loader.fact_table_dep_queue_table_deps which
    can be overridden if necessary for each queue table.
  
    After you configure all of your fact tables and queue tables, run the function
    refresh_fact_table_dep_queue_table_deps manually to populate fact_table_dep_queue_table_deps,
    then make any changes as necessary.
  
    You can see an example of this in the test suite
    "seeds" file.  You can also see an override example with order_emails_fact having a
    different proc for orders and reorders delete cases.
     */
    default_insert_merge_proid REGPROC NOT NULL,
    default_update_merge_proid REGPROC NOT NULL,
    default_delete_merge_proid REGPROC NOT NULL,
    row_created_at TIMESTAMPTZ DEFAULT NOW(),
    row_updated_at TIMESTAMPTZ,
    CONSTRAINT unique_fact_deps UNIQUE (parent_id, child_id)
  );
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.fact_table_deps', '');
  
  CREATE TABLE fact_loader.queue_tables (
    queue_table_id SERIAL PRIMARY KEY,
    queue_table_relid REGCLASS NOT NULL,
    queue_of_base_table_relid REGCLASS NOT NULL,
    /****
    NOTE - the reason for this config existing here is that we have no built-in way
    in pglogical to know which tables belong to which pglogical node.  Therefore, we
    need to configure that.  We hope that some time down the road, this will change,
    and we can derive this information.
     */
    pglogical_node_if_id INT NOT NULL,
    --This is the timezone for the changed_at column - if null, we assume it is timestamptz (we could check that actually)
    queue_table_tz TEXT,
    row_created_at TIMESTAMPTZ DEFAULT NOW(),
    row_updated_at TIMESTAMPTZ,
    CONSTRAINT unique_queue_table UNIQUE (queue_table_relid),
    CONSTRAINT unique_base_table UNIQUE (queue_of_base_table_relid)
  );
  COMMENT ON COLUMN fact_loader.queue_tables.pglogical_node_if_id IS $$The reason for this config existing here is that we have no built-in way
    in pglogical to know which tables belong to which pglogical node.  Therefore, we
    need to configure that.  We hope that some time down the road, this will change,
    and we can derive this information.$$;
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.queue_tables', '');
  
  CREATE TABLE fact_loader.queue_table_deps (
    queue_table_dep_id SERIAL PRIMARY KEY,
    fact_table_id INT NOT NULL REFERENCES fact_loader.fact_tables (fact_table_id),
    queue_table_id INT NOT NULL REFERENCES fact_loader.queue_tables (queue_table_id),
    relevant_change_columns NAME[],
    last_cutoff_id BIGINT,
    last_cutoff_source_time TIMESTAMPTZ,
    insert_merge_proid REGPROC NOT NULL,
    update_merge_proid REGPROC NOT NULL,
    delete_merge_proid REGPROC NOT NULL,
    row_created_at TIMESTAMPTZ DEFAULT NOW(),
    row_updated_at TIMESTAMPTZ,
    CONSTRAINT unique_queue_deps UNIQUE (fact_table_id, queue_table_id)
  );
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.queue_table_deps', '');
  
  CREATE TABLE fact_loader.key_retrieval_sequences (
    key_retrieval_sequence_id SERIAL PRIMARY KEY,
    queue_table_dep_id INT NOT NULL REFERENCES fact_loader.queue_table_deps (queue_table_dep_id),
    /****
    In almost all cases, we only need to write one way to retrieve keys.  The only exception is, for
    example, when in a delete case, you need to pass a different field (customer_id instead of order_id)
    to the delete_merge_proid function.  You then need a different key_retrieval_sequence to handle
    a different field name for this delete case.
  
    By default this is NULL, meaning there is no filter, meaning the sequence applies to all events I, U, D.
    Otherwise, you can add scopes in which case you must have one for each of 'I','U','D'.
     */
    filter_scope CHAR(1) NULL,
    level INT NOT NULL,
    return_columns NAME[] NOT NULL,
    is_fact_key BOOLEAN NOT NULL,
    join_to_relation REGCLASS NULL,
    join_to_column NAME NULL,
    return_columns_from_join NAME[] NULL,
    join_return_is_fact_key BOOLEAN NULL,
    CONSTRAINT unique_retrievals UNIQUE (queue_table_dep_id, filter_scope, level),
    CONSTRAINT valid_scopes CHECK (filter_scope IN ('I','U','D'))
  );
  
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.key_retrieval_sequences', '');
  
  CREATE TABLE fact_loader.fact_table_dep_queue_table_deps
  (
    fact_table_dep_queue_table_dep_id SERIAL PRIMARY KEY,
    fact_table_dep_id              INT REFERENCES fact_loader.fact_table_deps (fact_table_dep_id),
    queue_table_dep_id             INT REFERENCES fact_loader.queue_table_deps (queue_table_dep_id),
    last_cutoff_id                 BIGINT,
    last_cutoff_source_time        TIMESTAMPTZ,
    insert_merge_proid             REGPROC NOT NULL,
    update_merge_proid             REGPROC NOT NULL,
    delete_merge_proid             REGPROC NOT NULL,
    row_created_at                 TIMESTAMPTZ DEFAULT NOW(),
    row_updated_at                 TIMESTAMPTZ,
    CONSTRAINT unique_cutoffs UNIQUE (fact_table_dep_id, queue_table_dep_id)
  );
  
  CREATE OR REPLACE FUNCTION fact_loader.unique_scopes()
  RETURNS TRIGGER AS
  $BODY$
  BEGIN
    IF (NEW.filter_scope IS NULL AND EXISTS (
      SELECT 1
      FROM fact_loader.key_retrieval_sequences
      WHERE queue_table_dep_id <> NEW.queue_table_dep_id
        AND NEW.filter_scope IS NOT NULL
        )) OR
        (NEW.filter_scope IS NOT NULL AND EXISTS (
      SELECT 1
      FROM fact_loader.key_retrieval_sequences
      WHERE queue_table_dep_id <> NEW.queue_table_dep_id
        AND NEW.filter_scope IS NULL
        ))
    THEN
      RAISE EXCEPTION $$You must either use a NULL filter_scope to cover all 3 events I, U, D
      or you must specify all 3 events separately I, U, D (For queue_table_dep_id %).
      $$, NEW.queue_table_dep_id;
    END IF;
    RETURN NEW;
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE TRIGGER unique_scopes
  BEFORE INSERT OR UPDATE ON fact_loader.key_retrieval_sequences
  FOR EACH ROW EXECUTE PROCEDURE fact_loader.unique_scopes();
  
  /***
  This table is unlogged because it only has data mid-transaction and should always be empty
   */
  CREATE UNLOGGED TABLE fact_loader.process_queue (
    process_queue_id BIGSERIAL PRIMARY KEY,
    fact_table_id INT NOT NULL REFERENCES fact_loader.fact_tables (fact_table_id),
    proid REGPROC NOT NULL,
    key_value TEXT NOT NULL,
    row_created_at TIMESTAMPTZ DEFAULT NOW(),
    row_updated_at TIMESTAMPTZ
  );
  
  CREATE OR REPLACE FUNCTION fact_loader.set_row_updated_at_to_now()
  RETURNS TRIGGER AS
  $BODY$
  BEGIN
    NEW.row_updated_at = now();
    RETURN NEW;
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE TRIGGER set_row_updated_at_to_now
  BEFORE INSERT OR UPDATE ON fact_loader.fact_tables
  FOR EACH ROW
  WHEN (NEW.row_updated_at IS DISTINCT FROM now())
  EXECUTE PROCEDURE fact_loader.set_row_updated_at_to_now();
  
  CREATE TRIGGER set_row_updated_at_to_now
  BEFORE INSERT OR UPDATE ON fact_loader.fact_table_deps
  FOR EACH ROW
  WHEN (NEW.row_updated_at IS DISTINCT FROM now())
  EXECUTE PROCEDURE fact_loader.set_row_updated_at_to_now();
  
  CREATE TRIGGER set_row_updated_at_to_now
  BEFORE INSERT OR UPDATE ON fact_loader.queue_tables
  FOR EACH ROW
  WHEN (NEW.row_updated_at IS DISTINCT FROM now())
  EXECUTE PROCEDURE fact_loader.set_row_updated_at_to_now();
  
  CREATE TRIGGER set_row_updated_at_to_now
  BEFORE INSERT OR UPDATE ON fact_loader.queue_table_deps
  FOR EACH ROW
  WHEN (NEW.row_updated_at IS DISTINCT FROM now())
  EXECUTE PROCEDURE fact_loader.set_row_updated_at_to_now();
  
  CREATE TRIGGER set_row_updated_at_to_now
  BEFORE INSERT OR UPDATE ON fact_loader.fact_table_dep_queue_table_deps
  FOR EACH ROW
  WHEN (NEW.row_updated_at IS DISTINCT FROM now())
  EXECUTE PROCEDURE fact_loader.set_row_updated_at_to_now();
  
  CREATE TRIGGER set_row_updated_at_to_now
  BEFORE INSERT OR UPDATE ON fact_loader.process_queue
  FOR EACH ROW
  WHEN (NEW.row_updated_at IS DISTINCT FROM now())
  EXECUTE PROCEDURE fact_loader.set_row_updated_at_to_now();
  
  CREATE TYPE fact_loader.table_load_type AS ENUM('delta','full_refresh');
  
  
  CREATE OR REPLACE FUNCTION fact_loader.create_table_loader_function
  (p_source_proc REGPROC,
   p_destination_relation REGCLASS,
   p_ignore_diff_for_columns TEXT[])
  RETURNS REGPROC AS
  $BODY$
  DECLARE
    v_new_proc TEXT;
    v_sql TEXT;
  BEGIN
  
  /****
  Find the primary key for the destination table.  This is required.
  If the destination table does not have a primary key, it should.
  
  This is partly for simplicity, and partly to encourage good practice
  that we build and refresh tables based on chosen primary key to match
  records 1 for 1, which is basic DB design 101.
   */
    SELECT function_name, function_sql
    INTO v_new_proc, v_sql
    FROM fact_loader.table_loader_function(p_source_proc, p_destination_relation, p_ignore_diff_for_columns);
  
    EXECUTE v_sql;
  
    RETURN v_new_proc::REGPROC;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.execute_queue(p_fact_table_id INT)
  RETURNS TABLE (sql TEXT) AS
  $BODY$
  BEGIN
  
  RETURN QUERY
  WITH ordered_process_queue AS
  (SELECT process_queue_id, proid, key_value,
    --TODO - either infer the data type of the function args, which is not super easy with postgres,
    --or add configuration fields for the name and data type of these.  This will suffice for now
    --because we only have integer args for all functions
     'integer' AS queue_of_base_table_key_type
  FROM fact_loader.process_queue pq
  WHERE pq.fact_table_id = p_fact_table_id
  ORDER BY process_queue_id)
  
  , with_rank AS
  (SELECT format('%s(%s::%s)', proid::TEXT, 'key_value', queue_of_base_table_key_type) AS function_call,
    process_queue_id,
    RANK() OVER (PARTITION BY proid) AS execution_group
  FROM ordered_process_queue
  )
  
  , execute_sql_groups AS
  (
  SELECT execution_group,
  format($$
  WITH newly_processed AS (
  SELECT process_queue_id, %s
  FROM (
  /****
  Must wrap this to execute in order of ids
  ***/
  SELECT *
  FROM fact_loader.process_queue
  WHERE process_queue_id BETWEEN %s AND %s
    AND fact_table_id = %s
  ORDER BY process_queue_id) q
  )
  
  DELETE FROM fact_loader.process_queue pq USING newly_processed np
  WHERE np.process_queue_id = pq.process_queue_id;
  $$, function_call, MIN(process_queue_id), MAX(process_queue_id), p_fact_table_id) AS execute_sql
  FROM with_rank
  GROUP BY execution_group, function_call
  ORDER BY execution_group
  )
  
  SELECT COALESCE(string_agg(execute_sql,''),'SELECT NULL') AS final_execute_sql
  FROM execute_sql_groups;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.execute_table_loader
  (p_source_relation REGCLASS,
   p_destination_relation REGCLASS,
   p_ignore_diff_for_columns TEXT[],
   p_load_type fact_loader.table_load_type,
   p_ignore_unmapped_columns BOOLEAN = FALSE)
  RETURNS TABLE (upserted INT, deleted INT, truncated BOOLEAN, pct_dest NUMERIC(8,2)) AS
  $BODY$
  /***
    The SQL executed within this container is the actual
    load to the destination table, and assumes that 'prepare'
    phase has already been run, which is supposed to have gathered
    the actual minimal delta and determine what to do here.
   */
  DECLARE
    v_sql TEXT;
    v_unmapped_src_columns TEXT[];
    v_unmapped_dest_columns TEXT[];
  BEGIN
    SELECT execute_sql, unmapped_src_columns, unmapped_dest_columns INTO v_sql, v_unmapped_src_columns, v_unmapped_dest_columns
    FROM fact_loader.table_loader(
        p_source_relation,
        p_destination_relation,
        p_ignore_diff_for_columns,
        p_load_type);
    PERFORM fact_loader.table_loader_validator(p_source_relation,
                                          p_destination_relation,
                                          v_unmapped_src_columns,
                                          v_unmapped_dest_columns,
                                          p_ignore_unmapped_columns);
    RAISE LOG 'Executing SQL: %', v_sql;
    EXECUTE v_sql;
  
    RETURN QUERY
    SELECT * FROM count_tracker;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.load(p_fact_table_id INT)
  RETURNS VOID AS
  $BODY$
  DECLARE
      v_insert_to_process_queue_sql text;
      v_execute_sql text;
      v_metadata_update_sql text;
      v_debug_rec record;
      v_debug_text text = '';
  BEGIN
  /***
  There are 3 basic steps to this load:
      1. Gather all queue table changes and insert them into a consolidated process_queue
      2. Using the process_queue data, execute the delta load of the fact table
      3. Update the metadata indicating the last records updated for both the queue tables and fact table
   */
  
  /****
  Get SQL to insert new data into the consolidated process_queue,
  and SQL to update metadata for last_cutoffs.
   */
  SELECT insert_to_process_queue_sql, metadata_update_sql
  INTO v_insert_to_process_queue_sql, v_metadata_update_sql
  FROM fact_loader.sql_builder(p_fact_table_id);
  
  /****
  Populate the consolidated queue
   */
  RAISE LOG 'Populating Queue for fact_table_id %: %', p_fact_table_id, v_insert_to_process_queue_sql;
  EXECUTE COALESCE(v_insert_to_process_queue_sql, $$SELECT 'No queue data' AS result$$);
  
  /****
  For DEBUG purposes only to view the actual process_queue.  Requires setting log_min_messages to DEBUG.
   */
  IF current_setting('log_min_messages') LIKE 'debug%' THEN
      FOR v_debug_rec IN
          SELECT * FROM fact_loader.process_queue
      LOOP
          v_debug_text = v_debug_text||E'\n'||format('%s', v_debug_rec.process_queue_id||chr(9)||v_debug_rec.fact_table_id||chr(9)||v_debug_rec.proid||chr(9)||v_debug_rec.key_value);
      END LOOP;
      IF v_debug_text <> '' THEN
          v_debug_text = E'\n'||format('%s',
                                       (SELECT string_agg(column_name,chr(9))
                                        FROM information_schema.columns
                                        WHERE table_name = 'process_queue'
                                          AND table_schema = 'fact_loader'
                                          AND column_name NOT LIKE 'row_%_at'))
              ||v_debug_text;
          RAISE DEBUG '%', v_debug_text;
      END IF;
  END IF;
  
  /****
  With data now in the process_queue, the execute_queue function builds the SQL to execute.
  Save this SQL in a variable and execute it.
  If there is no data to execute, this is a no-op select statement.
   */
  SELECT sql INTO v_execute_sql FROM fact_loader.execute_queue(p_fact_table_id);
  RAISE LOG 'Executing Queue for fact_table_id %: %', p_fact_table_id, v_execute_sql;
  EXECUTE COALESCE(v_execute_sql, $$SELECT 'No queue data to execute' AS result$$);
  
  /****
  With everything finished, we now update the metadata for the fact_table.
  Even if no data was processed, we will still move forward last_refresh_attempted_at.
  
  last_refresh_succeeded will be marked true always for now.  It could in the future
  be used to indicate a failure in case of a caught error.
   */
  RAISE LOG 'Updating metadata for fact_table_id %: %', p_fact_table_id, v_metadata_update_sql;
  EXECUTE COALESCE(v_metadata_update_sql,
      format(
      $$UPDATE fact_loader.fact_tables ft
          SET last_refresh_attempted_at = now(),
            last_refresh_succeeded = TRUE
       WHERE fact_table_id = %s;
      $$, p_fact_table_id));
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.prepare_table_loader
  (p_source_relation REGCLASS,
   p_destination_relation REGCLASS,
   p_ignore_diff_for_columns TEXT[],
   p_load_type fact_loader.table_load_type,
   p_ignore_unmapped_columns BOOLEAN = FALSE)
  RETURNS TABLE (upserted INT, deleted INT, truncated BOOLEAN, pct_dest NUMERIC(8,2)) AS
  $BODY$
  /***
    The SQL executed within this container is not going
    to lock any of the destination table for writing, which
    is precisely why it is separated from the 'execute' phase
    which actually writes to the table in the shortest transaction
    possible.
   */
  DECLARE
    v_sql TEXT;
    v_unmapped_src_columns TEXT[];
    v_unmapped_dest_columns TEXT[];
  BEGIN
    SELECT prepare_sql, unmapped_src_columns, unmapped_dest_columns INTO v_sql, v_unmapped_src_columns, v_unmapped_dest_columns
    FROM fact_loader.table_loader(
        p_source_relation,
        p_destination_relation,
        p_ignore_diff_for_columns,
        p_load_type);
    PERFORM fact_loader.table_loader_validator(p_source_relation,
                                          p_destination_relation,
                                          v_unmapped_src_columns,
                                          v_unmapped_dest_columns,
                                          p_ignore_unmapped_columns);
    RAISE LOG 'Executing SQL: %', v_sql;
    EXECUTE v_sql;
  
    RETURN QUERY
    SELECT * FROM count_tracker;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.purge_queues
  (p_add_interval INTERVAL = '1 hour')
  RETURNS VOID AS
  $BODY$
  /*****
  The interval overlap is only important for delete cases in which you may need to join
  to another audit table in order to get a deleted row's data.  1 hour is somewhat arbitrary,
  but in the delete case, any related deleted rows would seem to normally appear very close to
  another relation's deleted rows.  1 hour is probably generous but also safe.
   */
  DECLARE
    v_sql TEXT;
  BEGIN
  
  WITH eligible_queue_tables_for_purge AS
  (SELECT
    /****
    This logic should handle dependent fact tables as well,
    because they share the same queue tables but they have separately
    logged last_cutoffs.
     */
     qt.queue_table_relid
     , qt.queue_table_id_field
     , queue_table_timestamp
     , queue_table_tz
     , MIN(last_cutoff_id)          AS min_cutoff_id
     , MIN(last_cutoff_source_time) AS min_source_time
   FROM fact_loader.queue_deps_all qt
   WHERE qt.last_cutoff_id IS NOT NULL
    /***
    There must be no other fact tables using the same queue
    which have not yet been processed at all
     */
    AND NOT EXISTS
     (SELECT 1
      FROM fact_loader.queue_deps_all qtdx
      WHERE qtdx.queue_table_id = qt.queue_table_id
            AND qtdx.last_cutoff_id IS NULL)
   GROUP BY qt.queue_table_relid
     , qt.queue_table_id_field
     , queue_table_timestamp
     , queue_table_tz)
  
  SELECT
    string_agg(
      format($$
      DELETE FROM %s
      WHERE %s <= %s
        AND %s %s < (%s::TIMESTAMPTZ - interval %s);
      $$,
        queue_table_relid,
        queue_table_id_field,
        min_cutoff_id,
        quote_ident(queue_table_timestamp),
        CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END,
        quote_literal(min_source_time),
        quote_literal(p_add_interval::TEXT)
        )
      , E'\n\n')
    INTO v_sql
  FROM eligible_queue_tables_for_purge;
  
  IF v_sql IS NOT NULL THEN
    RAISE LOG 'Purging Queue: %', v_sql;
    EXECUTE v_sql;
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE FUNCTION fact_loader.refresh_fact_table_dep_queue_table_deps()
  RETURNS VOID AS
  $BODY$
  BEGIN
  /****
  This function will be used to refresh the fact_table_dep_queue_table_deps table.
  The purpose of this table is to easily figure out queue data for fact tables that depend on other fact tables.
  This will be run with every call of load().
  This may not be the most efficient method, but it is certainly reliable and fast.
   */
  
  /****
  Recursively find all fact table deps including nested ones (fact tables that depend on other fact tables)
  to build the fact_table_dep_queue_table_deps table.
   */
  WITH RECURSIVE all_fact_table_deps AS (
    SELECT
      qtd.queue_table_dep_id
      , ftd.fact_table_dep_id
      , parent_id                                 AS parent_fact_table_id
      , child_id                                  AS fact_table_id
      , qtd.queue_table_id
      , qt.queue_table_relid
      , ftp.fact_table_relid AS parent_fact_table
      , ftc.fact_table_relid AS child_fact_table
      , ftd.default_insert_merge_proid
      , ftd.default_update_merge_proid
      , ftd.default_delete_merge_proid
    FROM fact_loader.queue_table_deps qtd
    INNER JOIN fact_loader.queue_tables qt ON qtd.queue_table_id = qt.queue_table_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.parent_id = qtd.fact_table_id
    INNER JOIN fact_loader.fact_tables ftp USING (fact_table_id)
    INNER JOIN fact_loader.fact_tables ftc ON ftc.fact_table_id = ftd.child_id
    UNION ALL
    SELECT
      qtd.queue_table_dep_id
      , ftd.fact_table_dep_id
      , parent_id                                 AS parent_fact_table_id
      , child_id                                  AS fact_table_id
      , qtd.queue_table_id
      , qt.queue_table_relid
      , ftp.fact_table_relid AS parent_fact_table
      , ft.fact_table_relid AS child_fact_table
      , ftd.default_insert_merge_proid
      , ftd.default_update_merge_proid
      , ftd.default_delete_merge_proid
    FROM all_fact_table_deps qtd
    INNER JOIN fact_loader.queue_tables qt ON qtd.queue_table_id = qt.queue_table_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.parent_id = qtd.fact_table_id
    INNER JOIN fact_loader.fact_tables ftp ON ftp.fact_table_id = ftd.parent_id
    INNER JOIN fact_loader.fact_tables ft ON ft.fact_table_id = ftd.child_id
  )
  
  /****
  Remove fact_table_dep_queue_table_deps that no longer exist if applicable
   */
  , removed AS (
    DELETE FROM fact_loader.fact_table_dep_queue_table_deps ftdqc
    WHERE NOT EXISTS(SELECT 1
                     FROM all_fact_table_deps aftd
                     WHERE aftd.fact_table_dep_id = ftdqc.fact_table_dep_id
                      AND aftd.queue_table_dep_id = ftdqc.queue_table_dep_id)
  )
  
  /****
  Add any new keys or ignore if they already exist
   */
  INSERT INTO fact_loader.fact_table_dep_queue_table_deps
  (fact_table_dep_id, queue_table_dep_id, insert_merge_proid, update_merge_proid, delete_merge_proid)
  SELECT fact_table_dep_id, queue_table_dep_id, default_insert_merge_proid, default_update_merge_proid, default_delete_merge_proid
  FROM all_fact_table_deps
  ON CONFLICT (fact_table_dep_id, queue_table_dep_id)
  DO NOTHING;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  
  CREATE OR REPLACE FUNCTION fact_loader.table_loader
  (p_source_relation REGCLASS,
   p_destination_relation REGCLASS,
   p_ignore_diff_for_columns TEXT[],
   p_load_type fact_loader.table_load_type)
  RETURNS TABLE (prepare_sql text, execute_sql text, unmapped_src_columns text[], unmapped_dest_columns text[]) AS
  $BODY$
  DECLARE
    v_pkey_fields TEXT[];
  BEGIN
  
  /****
  Find the primary key for the destination table.  This is required.
  If the destination table does not have a primary key, it should.
  
  This is partly for simplicity, and partly to encourage good practice
  that we build and refresh tables based on chosen primary key to match
  records 1 for 1, which is basic DB design 101.
   */
  SELECT array_agg(a.attname ORDER BY pk.rn) INTO v_pkey_fields
  FROM (SELECT
          i.indrelid
          , unnest(indkey) AS ik
          , row_number()
            OVER ()        AS rn
        FROM pg_index i
        WHERE i.indrelid = p_destination_relation AND i.indisprimary) pk
  INNER JOIN pg_attribute a
    ON a.attrelid = pk.indrelid AND a.attnum = pk.ik;
  
  RETURN QUERY
  WITH source_columns AS
  (
    SELECT column_name, ordinal_position, CASE WHEN column_name = ANY(v_pkey_fields) THEN TRUE ELSE FALSE END AS pkey_field
    FROM information_schema.columns c
    INNER JOIN pg_class pc ON pc.relname = c.table_name AND pc.oid = p_source_relation
    INNER JOIN pg_namespace n ON n.oid = pc.relnamespace AND c.table_schema = n.nspname
    ORDER BY ordinal_position
  )
  
  , destination_columns AS
  (
    SELECT column_name, ordinal_position, CASE WHEN column_name = ANY(v_pkey_fields) THEN TRUE ELSE FALSE END AS pkey_field
    FROM information_schema.columns c
    INNER JOIN pg_class pc ON pc.relname = c.table_name AND pc.oid = p_destination_relation
    INNER JOIN pg_namespace n ON n.oid = pc.relnamespace AND c.table_schema = n.nspname
    ORDER BY ordinal_position
  )
  
  , unmapped_source_columns AS
  (
    SELECT array_agg(s.column_name::text) AS unmapped_columns_src
    FROM source_columns s
    WHERE NOT EXISTS
    (SELECT 1 FROM destination_columns d WHERE d.column_name = s.column_name)
  )
  
  , unmapped_dest_columns AS
  (
    SELECT array_agg(d.column_name::text) AS unmapped_columns_dest
    FROM destination_columns d
    WHERE NOT EXISTS
    (SELECT 1 FROM source_columns s WHERE d.column_name = s.column_name)
  )
  
  , pkeys AS
  (
    SELECT
      string_agg(quote_ident(pkey_field),E'\n, ') AS pkey_fields,
      string_agg($$s.$$||quote_ident(pkey_field)||$$ = d.$$||quote_ident(pkey_field),E'\nAND ') AS pkey_join
    FROM
    (SELECT unnest AS pkey_field
    FROM unnest(v_pkey_fields)) pk
  )
  
  , info AS
  (
      SELECT
          string_agg(
            CASE
              WHEN sc.column_name IS NOT NULL
                THEN dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
            AS matching_column_list
        , string_agg(
            CASE
              WHEN sc.column_name IS NOT NULL
               AND (p_ignore_diff_for_columns IS NULL
                    OR sc.column_name != ALL (p_ignore_diff_for_columns)
                   )
                THEN dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
          AS matching_column_list_without_ignored
        , string_agg(
            CASE
              WHEN sc.column_name IS NOT NULL
               AND NOT dc.pkey_field
                THEN dc.column_name || ' = EXCLUDED.' || dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
          AS upsert_list
        , pkeys.pkey_fields
        , pkeys.pkey_join
      FROM destination_columns dc
        CROSS JOIN pkeys
        LEFT JOIN source_columns sc ON dc.column_name = sc.column_name
      GROUP BY pkeys.pkey_fields,
        pkeys.pkey_join
  )
  
  , sql_snippets AS
  (
    SELECT
      $$
      DROP TABLE IF EXISTS count_tracker;
      CREATE TEMP TABLE count_tracker (upserted INT, deleted INT, truncated BOOLEAN, pct_dest NUMERIC(8,2));
      INSERT INTO count_tracker VALUES (NULL, NULL, FALSE, NULL);
      $$::TEXT
        AS count_tracker_sql
      , $$
      DROP TABLE IF EXISTS actual_delta;
      CREATE TEMP TABLE actual_delta AS
      WITH final_diff AS (
      SELECT $$||pkey_fields||$$
      FROM
      (SELECT $$||matching_column_list_without_ignored||$$
      FROM $$||p_source_relation::TEXT||$$
      EXCEPT
      SELECT $$||matching_column_list_without_ignored||$$
      FROM $$||p_destination_relation::TEXT||$$ d $$
        AS actual_delta_sql
  
      , $$
      DROP TABLE IF EXISTS removed_keys;
      CREATE TEMP TABLE removed_keys AS
      SELECT $$||pkey_fields||$$
      FROM $$||p_destination_relation::TEXT||$$ d
      WHERE NOT EXISTS (SELECT 1 FROM $$||p_source_relation::TEXT||$$ s WHERE $$||pkey_join||$$);
      $$
        AS removed_keys_sql
  
      , $$
      ) full_diff)
  
      --This extra step is necessarily precisely because we may want to not except every column, like load_dttm
      SELECT *
      FROM $$||p_source_relation::TEXT||$$ s
      WHERE EXISTS (
        SELECT 1
        FROM final_diff d
        WHERE $$||pkey_join||$$
      );
      $$
        AS except_join_to_source_sql
  
      , $$
      /***
        We add the exists here because we are only looking for column-level differences
        for the given keys that have changed.  This may be a very small portion of the
        table.  Without the exists clause, this second part of EXCEPT would do a full
        table scan unnecessarily.
      ***/
      WHERE EXISTS (SELECT 1 FROM $$||p_source_relation::TEXT||$$ s WHERE $$||pkey_join||$$)$$
        AS key_join_exists_sql
  
      , $$
      /***
        We add a primary key to the actual_delta table to ensure there are no duplicate keys.
      ***/
      ALTER TABLE actual_delta ADD PRIMARY KEY ($$||pkey_fields||$$);
      $$
        AS add_delta_pkey_sql
  
      , $$
      /****
        This part is not implemented yet, but partially complete.
        If we decide we want to figure out that >50% of the table will be updated, we could decide
        to truncate.  But then we have to balance the desire for that with more read queries to
        figure it out.
  
        To implement, add the type full_refresh_truncate to fact_loader.table_load_type, and uncomment code.
        We would also have to add the logic to find actual keys added, then subtract it from actual_delta
        to get the net updates expected.  If this is over 50%, we should truncate and re-insert all data.
      ***/
      DROP TABLE IF EXISTS percent_of_destination;
      CREATE TEMP TABLE percent_of_destination AS
      SELECT
      (((SELECT COUNT(1) FROM actual_delta) - (SELECT COUNT(1) FROM added_keys))::NUMERIC /
      (SELECT COUNT(1) FROM $$||p_destination_relation::TEXT||$$)::NUMERIC)::NUMERIC(8,2) AS pct;
  
      UPDATE count_tracker SET pct_dest = (SELECT pct FROM percent_of_destination);
      $$
        AS percent_change_sql
  
      ,$$
      DO $LOCK_SAFE_DDL$
      BEGIN
      SET lock_timeout TO '10ms';
      IF (SELECT pct FROM percent_of_destination) >= 0.5 THEN
        LOOP
          BEGIN
            TRUNCATE $$||p_destination_relation::TEXT||$$;
            UPDATE count_tracker SET truncated = true;
            EXIT;
          EXCEPTION
            WHEN lock_not_available
              THEN RAISE WARNING 'Could not obtain immediate lock for SQL %, retrying', p_sql;
              PERFORM pg_sleep(3);
            WHEN OTHERS THEN
              RAISE;
          END;
        END LOOP;
      END IF;
      RESET lock_timeout;
      END
      $LOCK_SAFE_DDL$
      ;
      $$
        AS lock_safe_truncate_sql
  
      ,$$
      --Delete keys that are no longer in your new version
      DELETE FROM $$||p_destination_relation::TEXT||$$ d
      WHERE EXISTS
          (SELECT 1 FROM removed_keys s WHERE $$||pkey_join||$$);
  
      GET DIAGNOSTICS v_row_count = ROW_COUNT;
      UPDATE count_tracker SET deleted = v_row_count;
      $$
        AS delete_sql
  
      ,$$
      INSERT INTO $$||p_destination_relation::TEXT||$$ AS t ($$||
      matching_column_list||$$)
      SELECT $$||matching_column_list||
      $$ FROM actual_delta
      ON CONFLICT ($$||pkey_fields||$$)
      DO UPDATE
      SET $$||upsert_list||$$
      ;
  
      GET DIAGNOSTICS v_row_count = ROW_COUNT;
      UPDATE count_tracker SET upserted = v_row_count;
      $$
        AS upsert_sql
    FROM info
  )
  
  SELECT
    count_tracker_sql||
    CASE
  /*** not implemented truncate pattern
      WHEN p_load_type IN('full_refresh','full_refresh_truncate') THEN
  ***/
      WHEN p_load_type = 'full_refresh' THEN
        removed_keys_sql||actual_delta_sql||except_join_to_source_sql||add_delta_pkey_sql||$$;$$
      WHEN p_load_type = 'delta' THEN
        actual_delta_sql||key_join_exists_sql||except_join_to_source_sql||add_delta_pkey_sql||$$;$$
    END||$$
  
      $$||
  /*** not implemented truncate pattern
      CASE
      WHEN p_load_type = 'full_refresh_truncate' THEN
        percent_change_sql
      ELSE
      ''
    END
  ***/
    ''
    AS prepare_sql
    , $$
    --THIS SHOULD BE RUN IN A TRANSACTION
    DO $SCRIPT$
    DECLARE
      v_row_count INT;
      v_results RECORD;
    BEGIN
    $$||
    CASE
  /*** not implemented truncate pattern
      WHEN p_load_type = 'full_refresh_truncate' THEN
        lock_safe_truncate_sql||delete_sql||upsert_sql
  ***/
      WHEN p_load_type = 'full_refresh' THEN
        delete_sql||upsert_sql
    WHEN p_load_type = 'delta' THEN
        upsert_sql
    END||$$
  
    FOR v_results IN SELECT * FROM count_tracker LOOP
      RAISE LOG 'upserted: %, deleted: %, truncated: %, pct_dest: %',
      v_results.upserted, v_results.deleted, v_results.truncated, v_results.pct_dest;
    END LOOP;
  
    END
    $SCRIPT$;
  
    $$ AS execute_sql
  
    , (SELECT unmapped_columns_src FROM unmapped_source_columns) AS unmapped_src_columns
    , (SELECT unmapped_columns_dest FROM unmapped_dest_columns) AS unmapped_dest_columns
  FROM sql_snippets;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.table_loader_function
  (p_source_proc REGPROC,
   p_destination_relation REGCLASS,
   p_ignore_diff_for_columns TEXT[])
  RETURNS TABLE (function_name text, function_sql text) AS
  $BODY$
  BEGIN
  
  /****
  Find the primary key for the destination table.  This is required.
  If the destination table does not have a primary key, it should.
  
  This is partly for simplicity, and partly to encourage good practice
  that we build and refresh tables based on chosen primary key to match
  records 1 for 1, which is basic DB design 101.
   */
  RETURN QUERY
  WITH get_pkey_fields AS (
  SELECT
    a.attname,
    format_type(a.atttypid, a.atttypmod) AS atttype,
    pk.rn
  FROM (SELECT
          i.indrelid
          , unnest(indkey) AS ik
          , row_number()
            OVER ()        AS rn
        FROM pg_index i
        WHERE i.indrelid = p_destination_relation AND i.indisprimary) pk
  INNER JOIN pg_attribute a
    ON a.attrelid = pk.indrelid AND a.attnum = pk.ik)
  
  , pkey_fields_sorted AS
  (SELECT array_agg(attname ORDER BY rn) AS pkey_fields FROM get_pkey_fields)
  
  , function_args AS
  (SELECT regexp_matches(pg_get_function_identity_arguments(p_source_proc),'(?:^|, )(\w+)','g') AS arg)
  
  , function_schema AS
  (SELECT string_agg(arg[1],', ') AS arg_params,
   pg_get_function_identity_arguments(p_source_proc) AS arg_defs
   FROM function_args)
  
  , destination_columns AS
  (
    SELECT c.table_schema, c.table_name, column_name, ordinal_position, CASE WHEN gpf.attname IS NOT NULL THEN TRUE ELSE FALSE END AS pkey_field
    FROM information_schema.columns c
    INNER JOIN pg_class pc ON pc.relname = c.table_name AND pc.oid = p_destination_relation
    INNER JOIN pg_namespace n ON n.oid = pc.relnamespace AND c.table_schema = n.nspname
    LEFT JOIN get_pkey_fields gpf ON gpf.attname = c.column_name
    ORDER BY ordinal_position
  )
  
  , pkeys AS
  (
    SELECT
      string_agg(quote_ident(pkey_field),E'\n, ') AS pkey_fields,
      string_agg(quote_ident(pkey_field)||' '||pkey_type,', ') AS pkey_fields_ddl,
      string_agg($$s.$$||quote_ident(pkey_field)||$$ = d.$$||quote_ident(pkey_field),E'\nAND ') AS pkey_join
    FROM
    (SELECT attname AS pkey_field,
      atttype AS pkey_type
    FROM get_pkey_fields
    ORDER BY rn) pk
  )
  
  , info AS
  (
      SELECT
          string_agg(
            dc.column_name, E'\n  , '
            ORDER BY dc.ordinal_position
          )
            AS matching_column_list
        , string_agg(
            CASE
              WHEN (p_ignore_diff_for_columns IS NULL
                    OR dc.column_name != ALL (p_ignore_diff_for_columns)
                   )
                THEN dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
          AS matching_column_list_without_ignored
        , string_agg(
            CASE
              WHEN NOT dc.pkey_field
                THEN dc.column_name || ' = EXCLUDED.' || dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
          AS upsert_list
        , pkeys.pkey_fields
        , pkeys.pkey_fields_ddl
        , pkeys.pkey_join
        , quote_ident(dc.table_schema)||'.'||quote_ident(table_name||'_merge') AS proposed_function_name
        , fs.arg_params
        , fs.arg_defs
      FROM destination_columns dc
        CROSS JOIN pkeys
        CROSS JOIN function_schema fs
      GROUP BY pkeys.pkey_fields,
        pkeys.pkey_fields_ddl,
        pkeys.pkey_join,
        quote_ident(dc.table_schema)||'.'||quote_ident(table_name||'_merge'),
        fs.arg_params,
        fs.arg_defs
  )
  
  , sql_snippets AS
  (
    SELECT
      proposed_function_name
      , $$
      CREATE OR REPLACE FUNCTION $$||proposed_function_name||$$($$||arg_defs||$$)
      RETURNS VOID AS
      $FUNC$
      BEGIN
      $$::TEXT AS function_start
      , $$
      END;
      $FUNC$
      LANGUAGE plpgsql;
      $$::TEXT AS function_end
      , $$
      WITH actual_delta AS (
      $$::TEXT AS actual_delta_cte
  
      , $$
      WITH data AS (
      SELECT * FROM $$||p_source_proc::TEXT||$$($$||arg_params||$$)
      )
  
      , final_diff AS (
      SELECT $$||pkey_fields||$$
      FROM
      (SELECT $$||matching_column_list_without_ignored||$$
      FROM data
      EXCEPT
      SELECT $$||matching_column_list_without_ignored||$$
      FROM $$||p_destination_relation::TEXT||$$ d $$
        AS actual_delta_sql
  
      , $$
      ) full_diff)
  
      --This extra step is necessarily precisely because we may want to not except every column, like load_dttm
      SELECT *
      FROM data s
      WHERE EXISTS (
        SELECT 1
        FROM final_diff d
        WHERE $$||pkey_join||$$
      )
      $$
        AS except_join_to_source_sql
  
      , $$
      /***
        We add the exists here because we are only looking for column-level differences
        for the given keys that have changed.  This may be a very small portion of the
        table.  Without the exists clause, this second part of EXCEPT would do a full
        table scan unnecessarily.
      ***/
      WHERE EXISTS (SELECT 1 FROM data s WHERE $$||pkey_join||$$)$$
        AS key_join_exists_sql
  
      ,$$
      INSERT INTO $$||p_destination_relation::TEXT||$$ AS t ($$||
      matching_column_list||$$)
      SELECT $$||matching_column_list||
      $$ FROM actual_delta
      ON CONFLICT ($$||pkey_fields||$$)
      DO UPDATE
      SET $$||upsert_list||$$
      ;
      $$
        AS upsert_sql
    FROM info
  )
  
  SELECT
    proposed_function_name AS function_name
    , function_start||actual_delta_cte||actual_delta_sql||except_join_to_source_sql||')'||upsert_sql||function_end
        AS function_sql
  FROM sql_snippets;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.table_loader_validator
  (p_source_relation REGCLASS,
    p_destination_relation REGCLASS,
    p_unmapped_src_columns TEXT[],
    p_unmapped_dest_columns TEXT[],
    p_ignore_unmapped_columns BOOLEAN)
  RETURNS VOID AS
  $BODY$
  DECLARE v_messages TEXT = '';
  BEGIN
  
  IF NOT p_ignore_unmapped_columns AND p_unmapped_src_columns IS NOT NULL THEN
    v_messages = format($$You have unmapped columns (%s) in the source table %s.
    All source columns must be named identically to destination
    in order to map.
  
    If you are certain you want to ignore these columns, meaning they
    will not update anything in destination table %s, add the final argument
    to this function as TRUE.  $$
      , array_to_string(p_unmapped_src_columns,', ')
      , p_source_relation::TEXT
      , p_destination_relation::TEXT);
  END IF;
  IF NOT p_ignore_unmapped_columns AND p_unmapped_dest_columns IS NOT NULL THEN
    v_messages = v_messages||format($$
  
    You have unmapped columns (%s) in the destination table %s.
    All destination columns must be named identically to source
    in order to map.
  
    If you are certain you want to ignore these columns, meaning the source
    table %s does not contain all columns in destination table, add the final argument
    to this function as TRUE.$$
      , array_to_string(p_unmapped_dest_columns,', ')
      , p_destination_relation::TEXT
      , p_source_relation::TEXT);
  END IF;
  IF v_messages <> '' THEN
    RAISE EXCEPTION '%', v_messages;
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.worker()
  RETURNS BOOLEAN AS
  $BODY$
  DECLARE
    v_fact_record RECORD;
    c_lock_cutoff_refresh INT = 99995;
  BEGIN
  
  
  /****
  Attempt to refresh fact_table_dep_queue_table_deps or ignore if refresh is in progress.
   */
  IF (SELECT pg_try_advisory_xact_lock(c_lock_cutoff_refresh)) THEN
    PERFORM fact_loader.refresh_fact_table_dep_queue_table_deps();
  END IF;
  
  /****
  Acquire an advisory lock on the row indicating this job, which will cause the function
  to simply return false if another session is running it concurrently.
  It will be released upon transaction commit or rollback.
   */
  FOR v_fact_record IN
    SELECT fact_table_id
    FROM fact_loader.fact_tables
    WHERE enabled
    ORDER BY
      CASE WHEN force_worker_priority THEN 0 ELSE 1 END,
      --This may be improved in the future but is a good start
      last_refresh_attempted_at NULLS FIRST,
      priority
  LOOP
  
  IF (SELECT pg_try_advisory_xact_lock(fact_table_id)
          FROM fact_loader.fact_tables
          WHERE fact_table_id = v_fact_record.fact_table_id) THEN
    --Load fact table
    PERFORM fact_loader.load(v_fact_record.fact_table_id);
  
    /***
    Run purge process.  This need not run every launch of worker but it should not hurt.
    It is better for it to run after the fact table load is successful so as to avoid a
    rollback and more dead bloat
     */
    PERFORM fact_loader.purge_queues();
  
    RETURN TRUE;
  END IF;
  
  END LOOP;
  
  RETURN FALSE;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  /* pg_fact_loader--1.0--1.1.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  ALTER TABLE fact_loader.key_retrieval_sequences ADD COLUMN pass_queue_table_change_date_at_tz TEXT NULL;
  COMMENT ON COLUMN fact_loader.key_retrieval_sequences.pass_queue_table_change_date_at_tz
  IS
  $$If this is set to a time zone, then the changed_at field will be cast to this time zone and then cast to a date, for the purpose of creating a date-range based fact table.
  For casting queue_table_timestamp to a date, we first ensure we have it as timestamptz (objective UTC time).
  Then, we cast it to the timezone of interest on which the date should be based.
  For example, 02:00:00 UTC time on 2018-05-02 is actually 2018-05-01 in America/Chicago time.
  Thus, any date-based fact table must decide in what time zone to consider the date.$$;
  
  ALTER TABLE fact_loader.key_retrieval_sequences ADD CONSTRAINT verify_valid_tz CHECK (pass_queue_table_change_date_at_tz IS NULL OR (now() AT TIME ZONE pass_queue_table_change_date_at_tz IS NOT NULL));
  
  --This check constraint could have been added in v. 1.0
  ALTER TABLE fact_loader.queue_tables ADD CONSTRAINT verify_valid_tz CHECK (queue_table_tz IS NULL OR (now() AT TIME ZONE queue_table_tz IS NOT NULL));
  
  ALTER TABLE fact_loader.process_queue ADD COLUMN source_change_date DATE NULL;
  COMMENT ON COLUMN fact_loader.process_queue.source_change_date
  IS
  'Corresponds to fact_loader.key_retrieval_sequences.pass_queue_table_change_date_at_tz.  If this field is populated, a function will be expected that has args (key_value, source_change_date) based on this process_queue table.';
  
  --This should have already been added in v. 1.0
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.fact_table_dep_queue_table_deps', '');
  
  ALTER TABLE fact_loader.queue_table_deps
    ALTER COLUMN insert_merge_proid DROP NOT NULL,
    ALTER COLUMN update_merge_proid DROP NOT NULL,
    ALTER COLUMN delete_merge_proid DROP NOT NULL;
  
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps
    ALTER COLUMN insert_merge_proid DROP NOT NULL,
    ALTER COLUMN update_merge_proid DROP NOT NULL,
    ALTER COLUMN delete_merge_proid DROP NOT NULL;
  
  ALTER TABLE fact_loader.fact_table_deps
    ALTER COLUMN default_insert_merge_proid DROP NOT NULL,
    ALTER COLUMN default_update_merge_proid DROP NOT NULL,
    ALTER COLUMN default_delete_merge_proid DROP NOT NULL;
  
  
  
  
  
  CREATE OR REPLACE FUNCTION fact_loader.execute_queue(p_fact_table_id INT)
  RETURNS TABLE (sql TEXT) AS
  $BODY$
  BEGIN
  
  RETURN QUERY
  WITH ordered_process_queue AS
  (SELECT
     process_queue_id
     , proid
     , key_value
     , source_change_date
     , (pp.proargtypes::REGTYPE[])[0] AS proid_first_arg
   FROM fact_loader.process_queue pq
     LEFT JOIN pg_proc pp ON pp.oid = proid
   WHERE pq.fact_table_id = p_fact_table_id
   ORDER BY process_queue_id)
  
  , with_rank AS
  (SELECT
    /****
    If source_change_date is NULL, we assume the proid has one arg and pass it.
    If not, we assume the proid has two args and pass source_change_date as the second.
    */
     format('%s(%s::%s%s)'
            , proid::TEXT
            , 'key_value'
            , proid_first_arg
            , CASE
                WHEN source_change_date IS NOT NULL
                  THEN format(', %s::DATE',quote_literal(source_change_date))
                ELSE ''
              END
          ) AS function_call,
    proid,
    process_queue_id,
    RANK() OVER (PARTITION BY proid) AS execution_group
  FROM ordered_process_queue
  )
  
  , execute_sql_groups AS
  (
  SELECT execution_group,
  format($$
  WITH newly_processed AS (
  SELECT process_queue_id, %s
  FROM (
  /****
  Must wrap this to execute in order of ids
  ***/
  SELECT *
  FROM fact_loader.process_queue
  WHERE process_queue_id BETWEEN %s AND %s
    AND fact_table_id = %s
    AND proid = %s::REGPROC
  ORDER BY process_queue_id) q
  )
  
  DELETE FROM fact_loader.process_queue pq USING newly_processed np
  WHERE np.process_queue_id = pq.process_queue_id;
  $$, function_call, MIN(process_queue_id), MAX(process_queue_id), p_fact_table_id, quote_literal(proid::TEXT)) AS execute_sql
  FROM with_rank
  GROUP BY execution_group, function_call, proid
  ORDER BY execution_group
  )
  
  SELECT COALESCE(string_agg(execute_sql,''),'SELECT NULL') AS final_execute_sql
  FROM execute_sql_groups;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.table_loader_function
  (p_source_proc REGPROC,
   p_destination_relation REGCLASS,
   p_ignore_diff_for_columns TEXT[])
  RETURNS TABLE (function_name text, function_sql text) AS
  $BODY$
  BEGIN
  
  /****
  Find the primary key for the destination table.  This is required.
  If the destination table does not have a primary key, it should.
  
  This is partly for simplicity, and partly to encourage good practice
  that we build and refresh tables based on chosen primary key to match
  records 1 for 1, which is basic DB design 101.
   */
  RETURN QUERY
  WITH get_pkey_fields AS (
  SELECT
    a.attname,
    format_type(a.atttypid, a.atttypmod) AS atttype,
    pk.rn
  FROM (SELECT
          i.indrelid
          , unnest(indkey) AS ik
          , row_number()
            OVER ()        AS rn
        FROM pg_index i
        WHERE i.indrelid = p_destination_relation AND i.indisprimary) pk
  INNER JOIN pg_attribute a
    ON a.attrelid = pk.indrelid AND a.attnum = pk.ik)
  
  , pkey_fields_sorted AS
  (SELECT array_agg(attname ORDER BY rn) AS pkey_fields FROM get_pkey_fields)
  
  , function_args AS
  (SELECT regexp_matches(pg_get_function_identity_arguments(p_source_proc),'(?:^|, )(\w+)','g') AS arg)
  
  , function_schema AS
  (SELECT string_agg(arg[1],', ') AS arg_params,
   pg_get_function_identity_arguments(p_source_proc) AS arg_defs
   FROM function_args)
  
  , destination_columns AS
  (
    SELECT c.table_schema, c.table_name, column_name, ordinal_position, CASE WHEN gpf.attname IS NOT NULL THEN TRUE ELSE FALSE END AS pkey_field
    FROM information_schema.columns c
    INNER JOIN pg_class pc ON pc.relname = c.table_name AND pc.oid = p_destination_relation
    INNER JOIN pg_namespace n ON n.oid = pc.relnamespace AND c.table_schema = n.nspname
    LEFT JOIN get_pkey_fields gpf ON gpf.attname = c.column_name
    ORDER BY ordinal_position
  )
  
  , pkeys AS
  (
    SELECT
      string_agg(quote_ident(pkey_field),E'\n, ') AS pkey_fields,
      string_agg(quote_ident(pkey_field)||' '||pkey_type,', ') AS pkey_fields_ddl,
      string_agg($$s.$$||quote_ident(pkey_field)||$$ = d.$$||quote_ident(pkey_field),E'\nAND ') AS pkey_join,
      string_agg($$d.$$||quote_ident(pkey_field)||$$ = $$||(SELECT arg_params FROM function_schema),E'\nAND ') AS pkey_join_to_arg
    FROM
    (SELECT attname AS pkey_field,
      atttype AS pkey_type
    FROM get_pkey_fields
    ORDER BY rn) pk
  )
  
  , info AS
  (
      SELECT
          string_agg(
            dc.column_name, E'\n  , '
            ORDER BY dc.ordinal_position
          )
            AS matching_column_list
        , string_agg(
            CASE
              WHEN (p_ignore_diff_for_columns IS NULL
                    OR dc.column_name != ALL (p_ignore_diff_for_columns)
                   )
                THEN dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
          AS matching_column_list_without_ignored
        , string_agg(
            CASE
              WHEN NOT dc.pkey_field
                THEN dc.column_name || ' = EXCLUDED.' || dc.column_name
              ELSE
                NULL
            END, E'\n  , '
            ORDER BY dc.ordinal_position
          )
          AS upsert_list
        , pkeys.pkey_fields
        , pkeys.pkey_fields_ddl
        , pkeys.pkey_join
        , quote_ident(dc.table_schema)||'.'||quote_ident(table_name||'_merge') AS proposed_function_name
        , fs.arg_params
        , fs.arg_defs
        , pkey_join_to_arg
      FROM destination_columns dc
        CROSS JOIN pkeys
        CROSS JOIN function_schema fs
      GROUP BY pkeys.pkey_fields,
        pkeys.pkey_fields_ddl,
        pkeys.pkey_join,
        quote_ident(dc.table_schema)||'.'||quote_ident(table_name||'_merge'),
        fs.arg_params,
        fs.arg_defs,
        pkey_join_to_arg
  )
  
  , sql_snippets AS
  (
    SELECT
      proposed_function_name
      , $$
      CREATE OR REPLACE FUNCTION $$||proposed_function_name||$$($$||arg_defs||$$)
      RETURNS VOID AS
      $FUNC$
      BEGIN
      $$::TEXT AS function_start
      , $$
      END;
      $FUNC$
      LANGUAGE plpgsql;
      $$::TEXT AS function_end
      , $$
      WITH actual_delta AS (
      $$::TEXT AS actual_delta_cte
  
      , $$
      WITH data AS (
      SELECT * FROM $$||p_source_proc::TEXT||$$($$||arg_params||$$)
      )
  
      , final_diff AS (
      SELECT $$||pkey_fields||$$
      FROM
      (SELECT $$||matching_column_list_without_ignored||$$
      FROM data
      EXCEPT
      SELECT $$||matching_column_list_without_ignored||$$
      FROM $$||p_destination_relation::TEXT||$$ d
      WHERE $$||pkey_join_to_arg
        AS actual_delta_sql
  
      , $$
      ) full_diff)
  
      --This extra step is necessarily precisely because we may want to not except every column, like load_dttm
      SELECT *
      FROM data s
      WHERE EXISTS (
        SELECT 1
        FROM final_diff d
        WHERE $$||pkey_join||$$
      )
      $$
        AS except_join_to_source_sql
  
      , $$
      /***
        We add the exists here because we are only looking for column-level differences
        for the given keys that have changed.  This may be a very small portion of the
        table.  Without the exists clause, this second part of EXCEPT would do a full
        table scan unnecessarily.
      ***/
      WHERE EXISTS (SELECT 1 FROM data s WHERE $$||pkey_join||$$)$$
        AS key_join_exists_sql
  
      ,$$
      INSERT INTO $$||p_destination_relation::TEXT||$$ AS t ($$||
      matching_column_list||$$)
      SELECT $$||matching_column_list||
      $$ FROM actual_delta
      ON CONFLICT ($$||pkey_fields||$$)
      DO UPDATE
      SET $$||upsert_list||$$
      ;
      $$
        AS upsert_sql
    FROM info
  )
  
  SELECT
    proposed_function_name AS function_name
    , function_start||actual_delta_cte||actual_delta_sql||except_join_to_source_sql||')'||upsert_sql||function_end
        AS function_sql
  FROM sql_snippets;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  /* pg_fact_loader--1.1--1.2.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  --To support non-replicated queue tables
  ALTER TABLE fact_loader.queue_tables
    ALTER COLUMN pglogical_node_if_id DROP NOT NULL;
  
  CREATE TABLE fact_loader.fact_table_refresh_logs
  (fact_table_refresh_log_id SERIAL PRIMARY KEY,
  fact_table_id INT REFERENCES fact_loader.fact_tables (fact_table_id),
  refresh_attempted_at TIMESTAMPTZ,
  messages TEXT);
  
  ALTER TABLE fact_loader.fact_tables
    ADD COLUMN use_daily_schedule BOOLEAN NOT NULL DEFAULT FALSE,
    ADD COLUMN daily_scheduled_time TIME NULL,
    ADD COLUMN daily_scheduled_tz TEXT NULL,
    ADD COLUMN daily_scheduled_proid REGPROC,
    ADD CONSTRAINT verify_valid_daily_tz
      CHECK (daily_scheduled_tz IS NULL OR
             (now() AT TIME ZONE daily_scheduled_tz IS NOT NULL)),
    ADD CONSTRAINT daily_schedule_configured_correctly
      CHECK ((NOT use_daily_schedule) OR
             (use_daily_schedule
              AND daily_scheduled_time IS NOT NULL
              AND daily_scheduled_tz IS NOT NULL
              AND daily_scheduled_proid IS NOT NULL));
  
  
  
  CREATE OR REPLACE VIEW fact_loader.unresolved_failures AS
  SELECT ft.fact_table_id,
    fact_table_relid,
    refresh_attempted_at,
    messages
  FROM fact_loader.fact_tables ft
  INNER JOIN fact_loader.fact_table_refresh_logs ftrl
    ON ft.fact_table_id = ftrl.fact_table_id
    AND ft.last_refresh_attempted_at = ftrl.refresh_attempted_at
  WHERE NOT enabled
    AND NOT last_refresh_succeeded;
  
  CREATE OR REPLACE VIEW fact_loader.prioritized_jobs AS
  SELECT *
  FROM fact_loader.fact_tables
  WHERE enabled
    AND (NOT use_daily_schedule OR
         --Only run this job according to the same day of the daily_scheduled_time
         --according to configured timezone
         (
           (last_refresh_attempted_at IS NULL
            OR last_refresh_attempted_at::DATE < (now() AT TIME ZONE daily_scheduled_tz)::DATE
           )
            AND (now() AT TIME ZONE daily_scheduled_tz)::TIME
              BETWEEN daily_scheduled_time AND '23:59:59.999999'::TIME
         )
        )
  ORDER BY
    CASE WHEN force_worker_priority THEN 0 ELSE 1 END,
    --If a job has a daily schedule, once the time has come for the next refresh,
    --prioritize it first
    CASE
      WHEN (use_daily_schedule AND
            (last_refresh_attempted_at IS NULL
              OR last_refresh_attempted_at::DATE < (now() AT TIME ZONE daily_scheduled_tz)::DATE
             )
            AND (now() AT TIME ZONE daily_scheduled_tz)::TIME
              BETWEEN daily_scheduled_time AND '23:59:59.999999'::TIME)
      THEN (now() AT TIME ZONE daily_scheduled_tz)::TIME
      ELSE NULL
    END NULLS LAST,
    --This may be improved in the future but is a good start
    last_refresh_attempted_at NULLS FIRST,
    priority
  ;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.daily_scheduled_load(p_fact_table_id INT)
  RETURNS BOOLEAN AS
  $BODY$
  DECLARE
      v_execute_sql text;
  BEGIN
  /***
  There are 2 basic steps to this load:
      1. Execute the single daily-refresh function
      2. Update the metadata indicating the last attempt time
   */
  SELECT 'SELECT '||daily_scheduled_proid::TEXT||'()' INTO v_execute_sql
  FROM fact_loader.fact_tables
  WHERE fact_table_id = p_fact_table_id
    AND use_daily_schedule;
  
  IF v_execute_sql IS NULL THEN
      RETURN FALSE;
  END IF;
  
  EXECUTE v_execute_sql;
  
  UPDATE fact_loader.fact_tables ft
  SET last_refresh_attempted_at = now(),
      last_refresh_succeeded = TRUE
  WHERE fact_table_id = p_fact_table_id;
  
  RETURN TRUE;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.try_load(p_fact_table_id INT)
  RETURNS BOOLEAN AS
  $BODY$
  /***
  This will be used by the worker, but can also be used safely if a DBA
  wants to run a job manually.
   */
  DECLARE
    c_lock_cutoff_refresh INT = 99995;
  BEGIN
  
  IF (SELECT pg_try_advisory_xact_lock(fact_table_id)
          FROM fact_loader.fact_tables
          WHERE fact_table_id = p_fact_table_id) THEN
    /****
    Attempt to refresh fact_table_dep_queue_table_deps or ignore if refresh is in progress.
     */
    IF (SELECT pg_try_advisory_xact_lock(c_lock_cutoff_refresh)) THEN
      PERFORM fact_loader.refresh_fact_table_dep_queue_table_deps();
    END IF;
  
    --Load fact table and handle exceptions to auto-disable job and log errors in case of error
    BEGIN
      --Scheduled daily job
      IF (SELECT use_daily_schedule
          FROM fact_loader.fact_tables
          WHERE fact_table_id = p_fact_table_id) THEN
  
        PERFORM fact_loader.daily_scheduled_load(p_fact_table_id);
  
      --Queue-based job
      ELSE
        PERFORM fact_loader.load(p_fact_table_id);
  
        /***
        Run purge process.  This need not run every launch of worker but it should not hurt.
        It is better for it to run after the fact table load is successful so as to avoid a
        rollback and more dead bloat
         */
        PERFORM fact_loader.purge_queues();
  
      END IF;
  
      RETURN TRUE;
  
    EXCEPTION
      WHEN OTHERS THEN
        UPDATE fact_loader.fact_tables
        SET last_refresh_succeeded = FALSE,
            last_refresh_attempted_at = now(),
            enabled = FALSE
        WHERE fact_table_id = p_fact_table_id;
  
        INSERT INTO fact_loader.fact_table_refresh_logs (fact_table_id, refresh_attempted_at, messages)
        VALUES (p_fact_table_id, now(), SQLERRM);
  
        RETURN FALSE;
    END;
  ELSE
    RETURN FALSE;
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.worker()
  RETURNS BOOLEAN AS
  $BODY$
  DECLARE
    v_fact_record RECORD;
  BEGIN
  
  /****
  Acquire an advisory lock on the row indicating this job, which will cause the function
  to simply return false if another session is running it concurrently.
  It will be released upon transaction commit or rollback.
   */
  FOR v_fact_record IN
    SELECT fact_table_id
    FROM fact_loader.prioritized_jobs
  LOOP
  
  IF fact_loader.try_load(v_fact_record.fact_table_id) THEN
    RETURN TRUE;
  END IF;
  
  END LOOP;
  
  --If no jobs returned true, then return false
  RETURN FALSE;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  /* pg_fact_loader--1.2--1.3.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  DROP VIEW IF EXISTS fact_loader.queue_deps_all_with_retrieval;
  DROP VIEW IF EXISTS fact_loader.queue_deps_all;
  DROP VIEW IF EXISTS fact_loader.logical_subscription;
  DROP VIEW IF EXISTS fact_loader.prioritized_jobs;
  DROP VIEW IF EXISTS fact_loader.unresolved_failures;
  DROP FUNCTION IF EXISTS fact_loader.sql_builder(int);
  CREATE OR REPLACE FUNCTION fact_loader.add_batch_id_fields()
  RETURNS VOID
  AS $BODY$
  DECLARE
    v_rec RECORD;
    v_sql TEXT;
  BEGIN
  
  FOR v_rec IN
    SELECT queue_table_relid
    FROM fact_loader.queue_tables qt
    INNER JOIN pg_class c ON c.oid = qt.queue_table_relid
    INNER JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE NOT EXISTS
      (SELECT 1
      FROM information_schema.columns col
      WHERE col.column_name = 'fact_loader_batch_id'
        AND col.table_schema = n.nspname
        AND col.table_name = c.relname)
  LOOP
  
  v_sql = format($F$
  ALTER TABLE %s
  ADD COLUMN fact_loader_batch_id
  BIGINT
  DEFAULT nextval('fact_loader.batch_id');
  $F$,
  v_rec.queue_table_relid::text,
  v_rec.queue_table_relid::text);
  
  RAISE LOG 'ADDING fact_loader_batch_id COLUMN TO queue table %: %', v_rec.queue_table_relid::text, v_sql;
  EXECUTE v_sql;
  
  END LOOP;
  
  END
  $BODY$
  LANGUAGE plpgsql;
  
  
  ALTER TABLE fact_loader.queue_tables ADD COLUMN purge BOOLEAN NOT NULL DEFAULT TRUE;
  UPDATE fact_loader.fact_table_refresh_logs SET messages = jsonb_build_object('Message', messages) WHERE messages IS NOT NULL;
  --Will be re-added via \i in sql file
  ALTER TABLE fact_loader.fact_table_refresh_logs ALTER COLUMN messages TYPE jsonb USING messages::jsonb;
  
  --This was a problem from the start
  ALTER TABLE fact_loader.queue_tables ALTER COLUMN pglogical_node_if_id TYPE OID;
  
  --This should have been done from the start
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.fact_table_dep_queue_table_de_fact_table_dep_queue_table_de_seq', '');
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.fact_table_deps_fact_table_dep_id_seq', '');
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.fact_tables_fact_table_id_seq', '');
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.key_retrieval_sequences_key_retrieval_sequence_id_seq', '');
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.queue_table_deps_queue_table_dep_id_seq', '');
  SELECT pg_catalog.pg_extension_config_dump('fact_loader.queue_tables_queue_table_id_seq', '');
  
  --No indexes or anything but allow debugging
  CREATE UNLOGGED TABLE fact_loader.debug_process_queue (LIKE fact_loader.process_queue);
  ALTER TABLE fact_loader.debug_process_queue ADD PRIMARY KEY (process_queue_id);
  
  -- Now a temp table to avoid serialization contention
  DROP TABLE fact_loader.process_queue;
  
  --Make this a trigger to check dep fact tables
  ALTER TABLE fact_loader.fact_tables ADD COLUMN depends_on_base_daily_job_id INT REFERENCES fact_loader.fact_tables (fact_table_id);
  ALTER TABLE fact_loader.fact_tables ADD COLUMN depends_on_parent_daily_job_id INT REFERENCES fact_loader.fact_tables (fact_table_id);
  ALTER TABLE fact_loader.fact_tables DROP CONSTRAINT daily_schedule_configured_correctly;
  ALTER TABLE fact_loader.fact_tables ADD CONSTRAINT daily_schedule_configured_correctly CHECK (NOT use_daily_schedule OR (use_daily_schedule AND ((daily_scheduled_time IS NOT NULL AND daily_scheduled_tz IS NOT NULL AND daily_scheduled_proid IS NOT NULL) OR (depends_on_base_daily_job_id IS NOT NULL AND depends_on_parent_daily_job_id IS NOT NULL))));
  
  --These columns have never been used
  ALTER TABLE fact_loader.fact_tables DROP COLUMN attempt_number, DROP COLUMN retries_allowed;
  
  --This is the usual case and makes sense
  ALTER TABLE fact_loader.key_retrieval_sequences ALTER COLUMN level SET DEFAULT 1;
  
  --Need to have a more reliable dependency knowledge for scheduled jobs
  ALTER TABLE fact_loader.fact_tables ADD COLUMN daily_scheduled_deps REGCLASS[];
  ALTER TABLE fact_loader.fact_tables ADD COLUMN daily_scheduled_dep_delay_tolerance INTERVAL;
  ALTER TABLE fact_loader.fact_tables ADD CONSTRAINT daily_deps_correctly_configured CHECK ((daily_scheduled_deps IS NULL AND daily_scheduled_dep_delay_tolerance IS NULL) OR (daily_scheduled_deps IS NOT NULL AND daily_scheduled_dep_delay_tolerance IS NOT NULL));
  
  --Log all events and add pruning
  ALTER TABLE fact_loader.fact_table_refresh_logs ADD COLUMN refresh_finished_at TIMESTAMPTZ;
  ALTER TABLE fact_loader.fact_table_refresh_logs ALTER COLUMN fact_table_refresh_log_id TYPE BIGINT;
  
  -- Handle race conditions by changing to batch usage
  CREATE SEQUENCE fact_loader.batch_id;
  SELECT fact_loader.add_batch_id_fields();
  
  
  CREATE OR REPLACE FUNCTION fact_loader.queue_table_delay_info()
  RETURNS TABLE("replication_set_name" text,
             "queue_of_base_table_relid" regclass,
             "if_id" oid,
             "if_name" name,
             "source_time" timestamp with time zone)
  AS
  $BODY$
  /***
  This function exists to allow no necessary dependency
  to exist on pglogical_ticker.  If the extension is used,
  it will return data from its native functions, if not,
  it will return a null data set matching the structure
  ***/
  BEGIN
  
  IF EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pglogical_ticker') THEN
      RETURN QUERY EXECUTE $$
      SELECT
          unnest(coalesce(sub_replication_sets,'{NULL}')) AS replication_set_name
        , qt.queue_of_base_table_relid
        , n.if_id
        , n.if_name
        --source_time is now() for local tables (pglogical_node_if_id is null), and based on pglogical_ticker time otherwise
        , CASE
          WHEN qt.pglogical_node_if_id IS NULL
            THEN now()
          ELSE t.source_time
          END                          AS source_time
      FROM fact_loader.queue_tables qt
        LEFT JOIN fact_loader.logical_subscription() s ON qt.pglogical_node_if_id = s.sub_origin_if
        LEFT JOIN pglogical.node_interface n ON n.if_id = qt.pglogical_node_if_id
        LEFT JOIN pglogical_ticker.all_subscription_tickers() t ON t.provider_name = n.if_name;$$;
  ELSE
      RETURN QUERY
      SELECT
          NULL::TEXT AS replication_set_name
        , qt.queue_of_base_table_relid
        , NULL::OID AS if_id
        , NULL::NAME AS if_name
        --source_time is now() if queue tables are not pglogical-replicated, which is assumed if no ticker
        , now() AS source_time
      FROM fact_loader.queue_tables qt;
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE VIEW fact_loader.queue_deps_all AS
  WITH RECURSIVE fact_table_dep_cutoffs AS
  (SELECT
      1 AS level
      , qtd.queue_table_dep_id
      , ftdqc.fact_table_dep_id
      , ftdqc.fact_table_dep_queue_table_dep_id
      --This dep_maximum_cutoff_time is being taken from the queue_table_deps, because we cannot go past when the
      --fact table has been updated
      , qtd.last_cutoff_id                        AS dep_maximum_cutoff_id
      , qtd.last_cutoff_source_time               AS dep_maximum_cutoff_time
      , ftd.parent_id                                 AS parent_fact_table_id
      , ftd.child_id                                  AS child_fact_table_id
      , ftd.child_id                                  AS base_fact_table_id
      , queue_table_id
      , relevant_change_columns
      , ftdqc.last_cutoff_id
      , ftdqc.last_cutoff_source_time
      , ftdqc.insert_merge_proid
      , ftdqc.update_merge_proid
      , ftdqc.delete_merge_proid
    FROM fact_loader.queue_table_deps qtd
    INNER JOIN fact_loader.fact_table_dep_queue_table_deps ftdqc ON ftdqc.queue_table_dep_id = qtd.queue_table_dep_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.fact_table_dep_id = ftdqc.fact_table_dep_id
    UNION ALL
    /****
    In this recursive part, we walk UP the chain to the base level in order to get the
    last_cutoff_id and last_cutoff_source_time of parent_ids because children must never surpass those.
  
    The ONLY difference between this recursive part and the non-recursive part are the dep_maximum_cutoffs.
    That means we can get our resultant data below by simply selecting distinct ON the right fields and order
    by dep_maximum_cutoffs to get the most conservative cutoff window, that is, the minimum cutoff amongst
    the queue tables and any PARENT fact table cutoffs.
  
    That means if, for example,
      - IF a queue table has been cutoff up until 11:00:00
      - AND IF a level 1 fact table dependent on that queue table was last cutoff at 10:55:00
      - THEN a level 2 fact table dependent on level 1 fact table must not go past 10:55:00 when it is processed.
     */
    SELECT
      ftdc.level + 1 AS level
      , ftdc.queue_table_dep_id
      , ftdc.fact_table_dep_id
      , ftdc.fact_table_dep_queue_table_dep_id
      --This dep_maximum_cutoff_time is being taken from the queue_table_deps, because we cannot go past when the
      --fact table has been updated
      , ftdqc.last_cutoff_id                        AS dep_maximum_cutoff_id
      , ftdqc.last_cutoff_source_time               AS dep_maximum_cutoff_time
      , ftd.parent_id                                 AS parent_fact_table_id
      , ftd.child_id                                  AS child_fact_table_id
      , ftdc.base_fact_table_id
      , ftdc.queue_table_id
      , ftdc.relevant_change_columns
      , ftdc.last_cutoff_id
      , ftdc.last_cutoff_source_time
      , ftdc.insert_merge_proid
      , ftdc.update_merge_proid
      , ftdc.delete_merge_proid
    FROM fact_loader.queue_table_deps qtd
    INNER JOIN fact_loader.fact_table_dep_queue_table_deps ftdqc ON ftdqc.queue_table_dep_id = qtd.queue_table_dep_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.fact_table_dep_id = ftdqc.fact_table_dep_id
    INNER JOIN fact_table_dep_cutoffs ftdc ON ftdc.parent_fact_table_id = ftd.child_id
  )
  
  , adjusted_fact_table_deps AS (
  /****
  The reason we look at distinct queue_table_dep_id and not simply queue_table_id
  is because two parent fact tables could have differing logic for retrieving changes
  for the same base queue_tables.
   */
  SELECT DISTINCT ON(base_fact_table_id, queue_table_dep_id)
  *
  FROM fact_table_dep_cutoffs
  ORDER BY base_fact_table_id, queue_table_dep_id, dep_maximum_cutoff_time
  )
  
  , queue_table_info AS (
      SELECT * FROM fact_loader.queue_table_delay_info()
  )
  
  /****
  For fact tables that depend on other fact tables, we join the child fact table to the queue_table_deps of the parent
  fact table, and just reuse this exactly, with these distinctions:
    - From the fact_table_dep table, we do use the proids, and the last_cutoff_id
    - We use the parent last_cutoff_source_time as the maximum_cutoff, because we can only update those records already updated on the parent
    - We pass the information of which table for which to update metadata in the end
   */
  , queue_table_deps_with_nested AS (
    /****
    This part of the union is for the base level of queue_table_deps - for fact tables with no other dependent fact tables
     */
    SELECT
      queue_table_dep_id
      , NULL :: INT                                AS fact_table_dep_id
      , NULL :: INT                                AS fact_table_dep_queue_table_dep_id
      , NULL :: BIGINT                             AS dep_maximum_cutoff_id
      , NULL :: TIMESTAMPTZ                        AS dep_maximum_cutoff_time
      , fact_table_id
      , queue_table_id
      , relevant_change_columns
      , last_cutoff_id
      , last_cutoff_source_time
      , insert_merge_proid
      , update_merge_proid
      , delete_merge_proid
    FROM fact_loader.queue_table_deps
    UNION ALL
    /****
    This part of the union is for fact tables with other dependent fact tables
     */
    SELECT
      queue_table_dep_id
      , fact_table_dep_id
      , fact_table_dep_queue_table_dep_id
      , aftd.dep_maximum_cutoff_id
      , aftd.dep_maximum_cutoff_time
      , base_fact_table_id                                  AS fact_table_id
      , queue_table_id
      , relevant_change_columns
      , aftd.last_cutoff_id
      , aftd.last_cutoff_source_time
      , aftd.insert_merge_proid
      , aftd.update_merge_proid
      , aftd.delete_merge_proid
    FROM adjusted_fact_table_deps aftd
  )
  
  SELECT
    ft.fact_table_id,
    ft.fact_table_relid,
    ft.fact_table_agg_proid,
    qt.queue_table_id,
    qt.queue_table_relid,
    qt.queue_of_base_table_relid,
    qtd.relevant_change_columns,
    qtd.last_cutoff_id,
    qtd.last_cutoff_source_time,
    rt.if_name AS provider_name,
    rt.replication_set_name,
    qtd.dep_maximum_cutoff_id,  --Not used yet - TODO - think about if it needs to be used to filter as cutoff MAX in addition to the time filter
    LEAST(
        MIN(qtd.dep_maximum_cutoff_time)
        OVER (
          PARTITION BY qtd.fact_table_id ),
        MIN(rt.source_time)
        OVER (
          PARTITION BY qtd.fact_table_id )
    ) AS maximum_cutoff_time,
    aqt.queue_table_id_field,
    'primary_key'::name AS queue_table_key,
    'operation'::name AS queue_table_op,
    'change'::name AS queue_table_change,
    'changed_at'::name AS queue_table_timestamp,
    qt.queue_table_tz,
    aqbt.queue_of_base_table_key,
    aqbt.queue_of_base_table_key_type,
    queue_table_dep_id,
    fact_table_dep_id,
    fact_table_dep_queue_table_dep_id,
    insert_merge_proid,
    update_merge_proid,
    delete_merge_proid,
    qt.purge
  FROM queue_table_deps_with_nested qtd
  INNER JOIN fact_loader.fact_tables ft ON ft.fact_table_id = qtd.fact_table_id
  INNER JOIN fact_loader.queue_tables qt ON qt.queue_table_id = qtd.queue_table_id
  INNER JOIN queue_table_info rt ON rt.queue_of_base_table_relid = qt.queue_of_base_table_relid
  INNER JOIN LATERAL
    (SELECT a.attname AS queue_of_base_table_key, format_type(atttypid, atttypmod) AS queue_of_base_table_key_type
    FROM (SELECT
            i.indrelid
            , unnest(indkey) AS ik
            , row_number()
              OVER ()        AS rn
          FROM pg_index i
          WHERE i.indrelid = qt.queue_of_base_table_relid AND i.indisprimary) pk
    INNER JOIN pg_attribute a
      ON a.attrelid = pk.indrelid AND a.attnum = pk.ik) aqbt ON TRUE
  INNER JOIN LATERAL
    (SELECT a.attname AS queue_table_id_field
    FROM (SELECT
            i.indrelid
            , unnest(indkey) AS ik
            , row_number()
              OVER ()        AS rn
          FROM pg_index i
          WHERE i.indrelid = qt.queue_table_relid AND i.indisprimary) pk
    INNER JOIN pg_attribute a
      ON a.attrelid = pk.indrelid AND a.attnum = pk.ik) aqt ON TRUE
  ORDER BY ft.fact_table_relid;
  
  
  CREATE OR REPLACE VIEW fact_loader.queue_deps_all_with_retrieval AS
  SELECT
    qtd.*,
    krs.filter_scope,
    krs.level,
    krs.return_columns, --we need not get the type separately.  It must match queue_of_base_table_key_type
    krs.is_fact_key,
    krs.join_to_relation,
    qtk.queue_table_relid AS join_to_relation_queue,
    krs.join_to_column,
    ctypes.join_column_type,
    krs.return_columns_from_join,
    ctypes.return_columns_from_join_type,
    krs.join_return_is_fact_key,
    /***
    We include this in this view def to be easily shared by all events (I, U, D) in sql_builder,
    as those may be different in terms of passing source_change_date.
     */
    format(', %s::DATE AS source_change_date',
      CASE
        WHEN krs.pass_queue_table_change_date_at_tz IS NOT NULL
          /***
          For casting queue_table_timestamp to a date, we first ensure we have it as timestamptz (objective UTC time).
          Then, we cast it to the timezone of interest on which the date should be based.
          For example, 02:00:00 UTC time on 2018-05-02 is actually 2018-05-01 in America/Chicago time.
          Thus, any date-based fact table must decide in what time zone to consider the date.
           */
          THEN format('(%s %s AT TIME ZONE %s)',
                      'q.'||quote_ident(qtd.queue_table_timestamp),
                      CASE WHEN qtd.queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(qtd.queue_table_tz) END,
                      quote_literal(krs.pass_queue_table_change_date_at_tz))
          ELSE 'NULL'
        END) AS source_change_date_select
  FROM fact_loader.queue_deps_all qtd
  INNER JOIN fact_loader.key_retrieval_sequences krs ON qtd.queue_table_dep_id = krs.queue_table_dep_id
  LEFT JOIN fact_loader.queue_tables qtk ON qtk.queue_of_base_table_relid = krs.join_to_relation
  LEFT JOIN LATERAL
    (SELECT MAX(CASE WHEN attname = krs.join_to_column THEN format_type(atttypid, atttypmod) ELSE NULL END) AS join_column_type,
      MAX(CASE WHEN attname = krs.return_columns_from_join[1] THEN format_type(atttypid, atttypmod) ELSE NULL END) AS return_columns_from_join_type
    FROM pg_attribute a
    WHERE a.attrelid IN(krs.join_to_relation)
      /****
      We stubbornly assume that if there are multiple columns in return_columns_from_join, they all have the same type.
      Undue complexity would ensue if we did away with that rule.
       */
      AND a.attname IN(krs.join_to_column,krs.return_columns_from_join[1])) ctypes ON TRUE;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.purge_queues
  (p_add_interval INTERVAL = '1 hour')
  RETURNS VOID AS
  $BODY$
  /*****
  The interval overlap is only important for delete cases in which you may need to join
  to another audit table in order to get a deleted row's data.  1 hour is somewhat arbitrary,
  but in the delete case, any related deleted rows would seem to normally appear very close to
  another relation's deleted rows.  1 hour is probably generous but also safe.
   */
  DECLARE
    v_sql TEXT;
  BEGIN
  
  WITH eligible_queue_tables_for_purge AS
  (SELECT
    /****
    This logic should handle dependent fact tables as well,
    because they share the same queue tables but they have separately
    logged last_cutoffs.
     */
     qt.queue_table_relid
     , queue_table_timestamp
     , queue_table_tz
     , MIN(last_cutoff_id)          AS min_cutoff_id
     , MIN(last_cutoff_source_time) AS min_source_time
   FROM fact_loader.queue_deps_all qt
   WHERE qt.last_cutoff_id IS NOT NULL AND qt.purge
    /***
    There must be no other fact tables using the same queue
    which have not yet been processed at all
     */
    AND NOT EXISTS
     (SELECT 1
      FROM fact_loader.queue_deps_all qtdx
      WHERE qtdx.queue_table_id = qt.queue_table_id
            AND qtdx.last_cutoff_id IS NULL)
   GROUP BY qt.queue_table_relid
     , queue_table_timestamp
     , queue_table_tz)
  
  SELECT
    string_agg(
      format($$
      DELETE FROM %s
      WHERE %s IN
        (SELECT %s
        FROM %s
        WHERE %s <= %s
        AND %s %s < (%s::TIMESTAMPTZ - interval %s)
        FOR UPDATE SKIP LOCKED
        );
      $$,
        queue_table_relid,
        'fact_loader_batch_id',
        'fact_loader_batch_id',
        queue_table_relid,
        'fact_loader_batch_id',
        min_cutoff_id,
        quote_ident(queue_table_timestamp),
        CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END,
        quote_literal(min_source_time),
        quote_literal(p_add_interval::TEXT)
        )
      , E'\n\n')
    INTO v_sql
  FROM eligible_queue_tables_for_purge;
  
  IF v_sql IS NOT NULL THEN
      RAISE DEBUG 'Purging Queue: %', v_sql;
      BEGIN
          EXECUTE v_sql;
      EXCEPTION
          WHEN serialization_failure THEN
              RAISE LOG 'Serialization failure in queue purging for transaction % - skipping.', txid_current()::text;
          WHEN OTHERS THEN
              RAISE;
      END;
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  
  CREATE OR REPLACE FUNCTION fact_loader.worker()
  RETURNS BOOLEAN AS
  $BODY$
  DECLARE
    v_fact_record RECORD;
  BEGIN
  
  /****
  Acquire an advisory lock on the row indicating this job, which will cause the function
  to simply return false if another session is running it concurrently.
  It will be released upon transaction commit or rollback.
   */
  FOR v_fact_record IN
    SELECT fact_table_id
    FROM fact_loader.prioritized_jobs
  LOOP
  
  IF fact_loader.try_load(v_fact_record.fact_table_id) THEN
  --If any configured functions use temp tables,
  --must discard to avoid them hanging around in the idle background worker session
    DISCARD TEMP;
  
  --Log job times
    INSERT INTO fact_loader.fact_table_refresh_logs (fact_table_id, refresh_attempted_at, refresh_finished_at)
    VALUES (v_fact_record.fact_table_id, now(), clock_timestamp());
  
  --Return true meaning the fact table was refreshed (this applies even if there was no new data)
    RETURN TRUE;
  END IF;
  
  END LOOP;
  
  --If no jobs returned true, then return false
  RETURN FALSE;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.try_load(p_fact_table_id INT)
  RETURNS BOOLEAN AS
  $BODY$
  /***
  This will be used by the worker, but can also be used safely if a DBA
  wants to run a job manually.
   */
  DECLARE
    c_lock_cutoff_refresh INT = 99995;
    v_err JSONB;
    v_errmsg TEXT;
    v_errdetail TEXT;
    v_errhint TEXT;
    v_errcontext TEXT;
  BEGIN
  
  -- We except rare serialization failures here which we will ignore and move to the next record
  -- Anything else should be raised
  BEGIN
  IF EXISTS (SELECT fact_table_id
          FROM fact_loader.fact_tables
          WHERE fact_table_id = p_fact_table_id
          FOR UPDATE SKIP LOCKED) THEN
    /****
    Attempt to refresh fact_table_dep_queue_table_deps or ignore if refresh is in progress.
     */
    IF (SELECT pg_try_advisory_xact_lock(c_lock_cutoff_refresh)) THEN
      PERFORM fact_loader.refresh_fact_table_dep_queue_table_deps();
    END IF;
  
    --Load fact table and handle exceptions to auto-disable job and log errors in case of error
    BEGIN
      --Scheduled daily job
      IF (SELECT use_daily_schedule
          FROM fact_loader.fact_tables
          WHERE fact_table_id = p_fact_table_id) THEN
  
        PERFORM fact_loader.daily_scheduled_load(p_fact_table_id);
  
      --Queue-based job
      ELSE
        PERFORM fact_loader.load(p_fact_table_id);
  
        /***
        Run purge process.  This need not run every launch of worker but it should not hurt.
        It is better for it to run after the fact table load is successful so as to avoid a
        rollback and more dead bloat
         */
        PERFORM fact_loader.purge_queues();
  
      END IF;
  
      RETURN TRUE;
  
    EXCEPTION
      WHEN OTHERS THEN
        GET STACKED DIAGNOSTICS
          v_errmsg = MESSAGE_TEXT,
          v_errdetail = PG_EXCEPTION_DETAIL,
          v_errhint = PG_EXCEPTION_HINT,
          v_errcontext = PG_EXCEPTION_CONTEXT;
  
        UPDATE fact_loader.fact_tables
        SET last_refresh_succeeded = FALSE,
            last_refresh_attempted_at = now(),
            enabled = FALSE
        WHERE fact_table_id = p_fact_table_id;
  
        v_err = jsonb_strip_nulls(
              jsonb_build_object(
              'Message', v_errmsg,
              'Detail', case when v_errdetail = '' then null else v_errdetail end,
              'Hint', case when v_errhint = '' then null else v_errhint end,
              'Context', case when v_errcontext = '' then null else v_errcontext end)
              );
  
        INSERT INTO fact_loader.fact_table_refresh_logs (fact_table_id, refresh_attempted_at, refresh_finished_at, messages)
        VALUES (p_fact_table_id, now(), clock_timestamp(), v_err);
  
        RETURN FALSE;
    END;
  ELSE
    RETURN FALSE;
  END IF;
  
  EXCEPTION
      WHEN serialization_failure THEN
          RAISE LOG 'Serialization failure on transaction % attempting to lock % - skipping.', txid_current()::text, p_fact_table_id::text;
          RETURN FALSE;
      WHEN OTHERS THEN
          RAISE;
  
  END;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.load(p_fact_table_id INT)
  RETURNS VOID AS
  $BODY$
  DECLARE
      v_process_queue_sql text;
      v_execute_sql text;
      v_metadata_update_sql text;
      v_debug_rec record;
      v_debug_text text = '';
  BEGIN
  /***
  There are 3 basic steps to this load:
      1. Gather all queue table changes and insert them into a consolidated process_queue
      2. Update the metadata indicating the last records updated for both the queue tables and fact table
   */
  
  /****
  Get SQL to insert new data into the consolidated process_queue,
  and SQL to update metadata for last_cutoffs.
   */
  SELECT process_queue_sql, metadata_update_sql
  INTO v_process_queue_sql, v_metadata_update_sql
  FROM fact_loader.sql_builder(p_fact_table_id);
  
  /****
  Populate the consolidated queue
  This just creates a temp table with all changes to be processed
   */
  RAISE DEBUG 'Populating Queue for fact_table_id %: %', p_fact_table_id, v_process_queue_sql;
  EXECUTE COALESCE(v_process_queue_sql, $$SELECT 'No queue data' AS result$$);
  
  /****
  For DEBUG purposes only to view the actual process_queue.  Requires setting log_min_messages to DEBUG.
   */
  IF current_setting('log_min_messages') = 'debug3' THEN
      INSERT INTO fact_loader.debug_process_queue
      SELECT * FROM process_queue;
  END IF;
  
  /****
  With data now in the process_queue, the execute_queue function builds the SQL to execute.
  Save this SQL in a variable and execute it.
  If there is no data to execute, this is a no-op select statement.
   */
  SELECT sql INTO v_execute_sql FROM fact_loader.execute_queue(p_fact_table_id);
  RAISE DEBUG 'Executing Queue for fact_table_id %: %', p_fact_table_id, v_execute_sql;
  EXECUTE COALESCE(v_execute_sql, $$SELECT 'No queue data to execute' AS result$$);
  
  /****
  With everything finished, we now update the metadata for the fact_table.
  Even if no data was processed, we will still move forward last_refresh_attempted_at.
  
  last_refresh_succeeded will be marked true always for now.  It could in the future
  be used to indicate a failure in case of a caught error.
   */
  RAISE DEBUG 'Updating metadata for fact_table_id %: %', p_fact_table_id, v_metadata_update_sql;
  EXECUTE COALESCE(v_metadata_update_sql,
      format(
      $$UPDATE fact_loader.fact_tables ft
          SET last_refresh_attempted_at = now(),
            last_refresh_succeeded = TRUE
       WHERE fact_table_id = %s;
      $$, p_fact_table_id));
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE VIEW fact_loader.unresolved_failures AS
  SELECT ft.fact_table_id,
    fact_table_relid,
    refresh_attempted_at,
    messages
  FROM fact_loader.fact_tables ft
  INNER JOIN fact_loader.fact_table_refresh_logs ftrl
    ON ft.fact_table_id = ftrl.fact_table_id
    AND ft.last_refresh_attempted_at = ftrl.refresh_attempted_at
  WHERE NOT enabled
    AND NOT last_refresh_succeeded;
  
  CREATE OR REPLACE FUNCTION fact_loader.safely_terminate_workers()
  RETURNS TABLE (number_terminated INT, number_still_live INT, pids_still_live INT[]) AS
  $BODY$
  /****
  It is not a requirement to use this function to terminate workers.  Because workers are transactional,
  you can simply terminate them and no data loss will result in pg_fact_loader.
  
  Likewise, a hard crash of any system using pg_fact_loader will recover just fine upon re-launching workers.
  
  Still, it is ideal to avoid bloat to cleanly terminate workers and restart them using this function to kill
  them, and launch_workers(int) to re-launch them.
   */
  BEGIN
  
  RETURN QUERY
  WITH try_term_pids AS (
  SELECT
    pid,
    CASE WHEN
      state = 'idle' AND
      state_change BETWEEN SYMMETRIC
        now() - interval '5 seconds' AND
        now() - interval '55 seconds'
      THEN
      pg_terminate_backend(pid)
      ELSE FALSE
    END AS terminated
  FROM pg_stat_activity
  WHERE usename = 'postgres'
    AND query = 'SELECT fact_loader.worker();')
  
  SELECT SUM(CASE WHEN terminated THEN 1 ELSE 0 END)::INT AS number_terminated_out,
    SUM(CASE WHEN NOT terminated THEN 1 ELSE 0 END)::INT AS number_still_live_out,
    (SELECT array_agg(pid) FROM try_term_pids WHERE NOT terminated) AS pids_still_live_out
  FROM try_term_pids;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.launch_workers(number_to_launch int)
  RETURNS INT[] AS
  $BODY$
  DECLARE
    v_pids INT[];
  BEGIN
  
  FOR i IN 1..number_to_launch
  LOOP
    v_pids = array_append(v_pids, fact_loader.launch_worker());
  /*
  It's not strictly required to not launch all workers simultaneously, but it's
  also a little more invasive to do that, probably requiring more advisory lock skips.
  Better to just sleep 1 second between launches.
   */
  PERFORM pg_sleep(1);
  END LOOP;
  
  RETURN v_pids;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE VIEW fact_loader.prioritized_jobs AS
  WITH jobs_with_daily_variables AS (
  SELECT
     ft.*,
  /***
  Keep all this logic of daily jobs as variables to ease visualization of logic in the next cte below!!
   */
      (--If this is the first run of a scheduled job, it is eligible
              ft.last_refresh_attempted_at IS NULL
            OR (
                 --If it was last attempted successfully prior to this scheduled time only - meaning yesterday, it is eligible
                  (
                   ft.last_refresh_succeeded AND
                   ft.last_refresh_attempted_at::DATE <
                    -- Timezone taken from daily_scheduled_tz if base job, otherwise look up the timezone of the base job if this is dependent
                      (now() AT TIME ZONE COALESCE(
                                            ft.daily_scheduled_tz,
                                            base.daily_scheduled_tz
                                            )
                      )::DATE
                  )
                OR
                 --If a job has failed and been re-enabled, it is eligible again even though it has been attempted at or after the scheduled time
                   NOT ft.last_refresh_succeeded
               )
       ) AS daily_not_attempted_today,
  
    (now() AT TIME ZONE ft.daily_scheduled_tz)::TIME
        BETWEEN daily_scheduled_time AND '23:59:59.999999'::TIME AS daily_scheduled_time_passed,
  
    base.use_daily_schedule
    AND base.last_refresh_succeeded
    AND base.last_refresh_attempted_at :: DATE = (now() AT TIME ZONE base.daily_scheduled_tz) :: DATE
      AS daily_base_job_finished,
  
    ft.depends_on_base_daily_job_id = ft.depends_on_parent_daily_job_id AS daily_has_only_one_parent,
  
    -- This should only be used in combination with daily_has_only_one_parent
    parent.use_daily_schedule
    AND parent.last_refresh_succeeded
    AND parent.last_refresh_attempted_at :: DATE = (now() AT TIME ZONE COALESCE(parent.daily_scheduled_tz, base.daily_scheduled_tz)) :: DATE
      AS parent_job_finished
  FROM fact_loader.fact_tables ft
  LEFT JOIN LATERAL
    (SELECT ftb.use_daily_schedule,
      ftb.last_refresh_succeeded,
      ftb.last_refresh_attempted_at,
      ftb.daily_scheduled_tz
    FROM fact_loader.fact_tables ftb
    WHERE ftb.fact_table_id = ft.depends_on_base_daily_job_id) base ON TRUE
  LEFT JOIN LATERAL
    (SELECT ftp.use_daily_schedule,
      ftp.last_refresh_succeeded,
      ftp.last_refresh_attempted_at,
      ftp.daily_scheduled_tz
    FROM fact_loader.fact_tables ftp
    WHERE ftp.fact_table_id = ft.depends_on_parent_daily_job_id) parent ON TRUE
  WHERE enabled
  )
  
  , jobs_with_daily_schedule_eligibility AS (
  SELECT
     *,
     --Only run this job according to the same day of the daily_scheduled_time
     --according to configured timezone
     (use_daily_schedule AND daily_not_attempted_today
       AND
      (
        daily_scheduled_time_passed
        OR
       (daily_base_job_finished AND (daily_has_only_one_parent OR parent_job_finished))
      )
     ) AS daily_schedule_eligible
  FROM jobs_with_daily_variables)
  
  SELECT *
  FROM jobs_with_daily_schedule_eligibility
  WHERE NOT use_daily_schedule OR daily_schedule_eligible
  ORDER BY
    CASE WHEN force_worker_priority THEN 0 ELSE 1 END,
    --If a job has a daily schedule, once the time has come for the next refresh,
    --prioritize it first
    CASE
      WHEN daily_schedule_eligible
      THEN (now() AT TIME ZONE daily_scheduled_tz)::TIME
      ELSE NULL
    END NULLS LAST,
    --This may be improved in the future but is a good start
    last_refresh_attempted_at NULLS FIRST,
    priority
  ;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.sql_builder(p_fact_table_id INT)
    RETURNS TABLE(raw_queued_changes_sql text,
                  gathered_queued_changes_sql text,
                  process_queue_sql text,
                  metadata_update_sql text) AS
  $BODY$
  
  /****
  The recursive part of this CTE are only the sql_builder parts.
  In Postgres, if any of your CTEs are recursive, you only use the RECURSIVE keyword on the first of a set.
  
  The retrieval info may be the same for all 3 events (insert, update, delete), in which case filter_scope is null
  Otherwise, they must be specified separately.
   */
  WITH RECURSIVE queue_deps_with_insert_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'I' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  , queue_deps_with_update_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'U' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  , queue_deps_with_delete_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'D' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  /****
  Recursively build the SQL for any INSERT events found in the queues.
  
  The recursive part ONLY applies to cases where multiple joins have to be made to get at the source data,
  in which case there are multiple levels of key_retrieval_sequences for a given queue_table_dep_id. For an
  example of this, see the test cases involving the test.order_product_promos table.
   */
  , insert_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', b.'||return_columns[1]||'::TEXT AS key'
              ELSE ', unnest(array[b.'||array_to_string(return_columns, ',b.')||'])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||return_columns_from_join[1]||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level)||'])::TEXT AS key'
            END
          ELSE ''
      END AS key_select_column,
      CASE
        WHEN is_fact_key
          THEN ''
          ELSE 'INNER JOIN '||join_to_relation::TEXT||' j'||level||
               E'\n  ON b.'||quote_ident(return_columns[1])||' = j'||level||'.'||quote_ident(join_to_column)
      END AS key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_insert_retrieval c
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.'||return_columns[1]||'::TEXT AS key'
            ELSE ', unnest(b.'||array_to_string(return_columns,',j'||r.level)||')::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||return_columns_from_join[1]||'::TEXT AS key'
            ELSE ', unnest(j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS key_select_column,
      key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE E'\nINNER JOIN '||join_to_relation::TEXT||' j'||c.level||
             E'\n  ON j'||r.level||'.'||quote_ident(return_columns[1])||' = j'||c.level||'.'||quote_ident(join_to_column) END AS key_retrieval_sql,
      r.source_change_date_select
    FROM insert_sql_builder r
    INNER JOIN queue_deps_with_insert_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , update_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', b.'||return_columns[1]||'::TEXT AS key'
              ELSE ', unnest(array[b.'||array_to_string(return_columns, ',b.')||'])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||return_columns_from_join[1]||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level)||'])::TEXT AS key'
            END
          ELSE ''
      END AS key_select_column,
      CASE
        WHEN is_fact_key
          THEN ''
          ELSE 'INNER JOIN '||join_to_relation::TEXT||' j'||level||
               E'\n  ON b.'||quote_ident(return_columns[1])||' = j'||level||'.'||quote_ident(join_to_column)
      END AS key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_update_retrieval c
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.'||return_columns[1]||'::TEXT AS key'
            ELSE ', unnest(b.'||array_to_string(return_columns,',j'||r.level)||')::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||return_columns_from_join[1]||'::TEXT AS key'
            ELSE ', unnest(j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS key_select_column,
      key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE E'\nINNER JOIN '||join_to_relation::TEXT||' j'||c.level||
             E'\n  ON j'||r.level||'.'||quote_ident(return_columns[1])||' = j'||c.level||'.'||quote_ident(join_to_column) END AS key_retrieval_sql,
      r.source_change_date_select
    FROM update_sql_builder r
    INNER JOIN queue_deps_with_update_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , delete_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      --For deletes, same pattern as key_select_column but instead, we may be selecting from the audit tables instead
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', q.before_change->>'||quote_literal(return_columns[1])||'::TEXT AS key'
              ELSE ', unnest(array[before_change->>'''||array_to_string(return_columns, ''', before_change->>''')||'''])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||quote_ident(return_columns_from_join[1])||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level||'.')||'])::TEXT AS key'
            END
          ELSE ''
      END AS delete_key_select_column,
      CASE
      WHEN is_fact_key
        THEN ''
        ELSE format($$
          --Join to either the base table, or the audit table, one of which
          --will be missing the key in a delete case
            INNER JOIN LATERAL (
             SELECT %s FROM %s jb
             WHERE %s = %s
             UNION ALL
             SELECT %s FROM %s jq
             WHERE operation = 'D'
              AND %s = %s) %s ON TRUE
             $$, quote_ident(return_columns_from_join[1]),
              join_to_relation::TEXT,
              (CASE WHEN level = 1 THEN '(q'||'.before_change->>'||quote_literal(return_columns[1])||')::'||join_column_type ELSE 'j'||level||'.'||quote_ident(return_columns[1]) END),
              'jb.'||quote_ident(join_to_column),
              '(before_change->>'||quote_literal(return_columns_from_join[1])||')::'||return_columns_from_join_type,
              join_to_relation_queue::TEXT,
              (CASE WHEN level = 1 THEN '(q'||'.before_change->>'||quote_literal(return_columns[1])||')::'||join_column_type ELSE 'j'||level||'.'||quote_ident(return_columns[1]) END),
              '(jq.before_change->>'||quote_literal(join_to_column)||')::'||join_column_type,
              /****
              We use the higher level here just to be consistent with aliases from insert/update key retrieval
               */
              'j'||level
              )
      END
        AS delete_key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_delete_retrieval
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      delete_key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.before_change->>'||quote_literal(return_columns[1])||'::TEXT AS key'
            ELSE ', unnest(array[before_change->>'''||array_to_string(return_columns,',j'||r.level||'.before_change->>''')||'''])::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||quote_ident(return_columns_from_join[1])||'::TEXT AS key'
            ELSE ', unnest(array[j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS delete_key_select_column,
      delete_key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE format($$
          --Join to either the base table, or the audit table, one of which
          --will be missing the key in a delete case
            INNER JOIN LATERAL (
             SELECT %s FROM %s jb
             WHERE %s = %s
             UNION ALL
             SELECT %s FROM %s jq
             WHERE operation = 'D'
              AND %s = %s) %s ON TRUE
             $$, quote_ident(return_columns_from_join[1]),
              join_to_relation::TEXT,
              'j'||r.level||'.'||quote_ident(return_columns[1]),
              'jb.'||quote_ident(join_to_column),
              '(before_change->>'||quote_literal(return_columns_from_join[1])||')::'||return_columns_from_join_type,
              join_to_relation_queue::TEXT,
              'j'||r.level||'.'||quote_ident(return_columns[1]),
              '(jq.before_change->>'||quote_literal(join_to_column)||')::'||join_column_type,
              /****
              We use the higher level here just to be consistent with aliases from insert/update key retrieval
               */
              'j'||c.level
              )
      END
        AS delete_key_retrieval_sql,
      r.source_change_date_select
    FROM delete_sql_builder r
    INNER JOIN queue_deps_with_delete_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , non_recursive_sql AS (
    SELECT
    /****
    Separate select list for:
      - raw queue_ids from queue tables
      - gathered data from joining queue_ids to source tables to get actual keys to update in fact tables
     */
    -- gathering all queue_ids from queue tables
      queue_table_dep_id,
      format($$
      %s AS fact_table_id,
      %s AS queue_table_id,
      %s AS queue_table_dep_id,
      %s::INT AS fact_table_dep_id,
      %s::INT AS fact_table_dep_queue_table_dep_id,
      %s AS queue_table_id_field,
      q.fact_loader_batch_id,
      %s::TIMESTAMPTZ AS maximum_cutoff_time $$,
      fact_table_id,
      queue_table_id,
      queue_table_dep_id,
      (CASE WHEN fact_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_id::TEXT END),
      (CASE WHEN fact_table_dep_queue_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_queue_table_dep_id::TEXT END),
      'q.'||quote_ident(queue_table_id_field),
      quote_literal(maximum_cutoff_time))
        AS metadata_select_columns,
  
    -- gathering actual keys to update in fact tables by joining from queue_ids to source tables
      format($$
      %s AS fact_table_id,
      %s AS queue_table_dep_id,
      %s::INT AS fact_table_dep_id,
      %s::INT AS fact_table_dep_queue_table_dep_id,
      %s AS queue_table_id_field,
      q.fact_loader_batch_id,
      %s AS operation,
      %s %s AS changed_at,
      %s::REGPROC AS insert_merge_proid,
      %s::REGPROC AS update_merge_proid,
      %s::REGPROC AS delete_merge_proid,
      %s::TIMESTAMPTZ AS maximum_cutoff_time $$,
      fact_table_id,
      queue_table_dep_id,
      (CASE WHEN fact_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_id::TEXT END),
      (CASE WHEN fact_table_dep_queue_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_queue_table_dep_id::TEXT END),
      'q.'||quote_ident(queue_table_id_field),
      'q.'||quote_ident(queue_table_op),
      'q.'||quote_ident(queue_table_timestamp),
      CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END,
      CASE WHEN insert_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(insert_merge_proid) END,
      CASE WHEN update_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(update_merge_proid) END,
      CASE WHEN delete_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(delete_merge_proid) END,
      quote_literal(maximum_cutoff_time))
        AS global_select_columns,
  
    -- This is simply the queue table aliased as q
      format('%s q', queue_table_relid::TEXT) AS queue_table_aliased,
  
    -- This is the SQL to join from the queue table to the base table
      E'\nINNER JOIN '||queue_of_base_table_relid::TEXT||' b'||
      E'\n  ON q.'||quote_ident(queue_table_key)||'::'||queue_of_base_table_key_type||' = b.'||quote_ident(queue_of_base_table_key)
             AS base_join_sql,
  
    -- This is a WHERE statement to be added to ALL gathering of new queue_ids to process.
      format($$ %s
          AND q.%s < %s %s
          $$,
          CASE
            WHEN last_cutoff_id IS NOT NULL
            THEN 'q.fact_loader_batch_id > '||last_cutoff_id
            ELSE
              'TRUE'
          END,
          quote_ident(c.queue_table_timestamp),
          quote_literal(c.maximum_cutoff_time),
          CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END) AS global_where_sql,
      format($$
        AND q.%s = 'I'
        $$,
          queue_table_op)
             AS where_for_insert_sql,
      format($$
        AND (q.%s = 'U' AND %s)
        $$,
          queue_table_op,
          CASE
            WHEN relevant_change_columns IS NULL
              THEN 'TRUE'
            ELSE
              format($$q.%s ?| '{%s}'$$, queue_table_change, array_to_string(relevant_change_columns,','))
          END)
             AS where_for_update_sql,
      format($$
        AND q.%s = 'D'
        $$,
          queue_table_op)
             AS where_for_delete_sql
    FROM fact_loader.queue_deps_all c
    WHERE c.fact_table_id = p_fact_table_id
  )
  
  , insert_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM insert_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , update_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM update_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , delete_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM delete_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , all_queues_sql AS (
  SELECT
    format($$
    SELECT %s
    FROM %s
    %s
    WHERE %s
    $$,
      nrs.global_select_columns||isbf.key_select_column||isbf.source_change_date_select,
      nrs.queue_table_aliased||nrs.base_join_sql,
      isbf.key_retrieval_sql,
      nrs.global_where_sql||nrs.where_for_insert_sql) AS queue_insert_sql,
    format($$
    SELECT %s
    FROM %s
    %s
    WHERE %s
    $$,
      nrs.global_select_columns||usbf.key_select_column||usbf.source_change_date_select,
      nrs.queue_table_aliased||nrs.base_join_sql,
      usbf.key_retrieval_sql,
      nrs.global_where_sql||nrs.where_for_update_sql) AS queue_update_sql,
    format($$
    SELECT %s
    FROM %s
    %s
    WHERE %s
    $$,
      nrs.global_select_columns||dsbf.delete_key_select_column||usbf.source_change_date_select,
      nrs.queue_table_aliased,
      dsbf.delete_key_retrieval_sql,
      nrs.global_where_sql||nrs.where_for_delete_sql) AS queue_delete_sql,
    format($$
    SELECT %s
    FROM %s
    WHERE %s
    $$,
      nrs.metadata_select_columns,
      nrs.queue_table_aliased,
      nrs.global_where_sql) AS queue_ids_sql
  FROM non_recursive_sql nrs
  INNER JOIN insert_sql_builder_final isbf ON isbf.queue_table_dep_id = nrs.queue_table_dep_id
  INNER JOIN update_sql_builder_final usbf ON usbf.queue_table_dep_id = nrs.queue_table_dep_id
  INNER JOIN delete_sql_builder_final dsbf ON dsbf.queue_table_dep_id = nrs.queue_table_dep_id
  )
  
  , final_queue_sql AS
  (SELECT string_agg(
    /****
    This first UNION is to union together INSERT, UPDATE, and DELETE events for a single queue table
     */
    format($$
    %s
    UNION ALL
    %s
    UNION ALL
    %s
    $$,
      queue_insert_sql,
      queue_update_sql,
      queue_delete_sql)
    /****
    This second UNION as the second arg of string_agg is the union together ALL queue tables for this fact table
     */
    , E'\nUNION ALL\n') AS event_sql,
    string_agg(queue_ids_sql, E'\nUNION ALL\n') AS raw_queued_changes_sql_out
  FROM all_queues_sql)
  
  , final_outputs AS (
  SELECT raw_queued_changes_sql_out,
  $$
  WITH all_changes AS (
  ($$||event_sql||$$)
  ORDER BY changed_at)
  
  , base_execution_groups AS
  (SELECT fact_table_id,
      queue_table_dep_id,
      queue_table_id_field,
      operation,
      changed_at,
      source_change_date,
      insert_merge_proid,
      update_merge_proid,
      delete_merge_proid,
      maximum_cutoff_time,
      key,
      CASE WHEN operation = 'I' THEN insert_merge_proid
      WHEN operation = 'U' THEN update_merge_proid
      WHEN operation = 'D' THEN delete_merge_proid
          END AS proid,
    RANK() OVER (
      PARTITION BY
        CASE
          WHEN operation = 'I' THEN insert_merge_proid
          WHEN operation = 'U' THEN update_merge_proid
          WHEN operation = 'D' THEN delete_merge_proid
        END
    ) AS execution_group
    FROM all_changes
    WHERE key IS NOT NULL)
  
  SELECT fact_table_id, proid, key, source_change_date
  FROM base_execution_groups beg
  WHERE proid IS NOT NULL
  GROUP BY execution_group, fact_table_id, proid, key, source_change_date
  /****
  This ordering is particularly important for date-range history tables
  where order of inserts is critical and usually expected to follow a pattern
  ***/
  ORDER BY execution_group, MIN(changed_at), MIN(queue_table_id_field);
  $$ AS gathered_queued_changes_sql_out
    ,
  
  $$
  DROP TABLE IF EXISTS process_queue;
  CREATE TEMP TABLE process_queue
  (process_queue_id serial,
   fact_table_id int,
   proid regproc,
   key_value text,
   source_change_date date);
  
  INSERT INTO process_queue
  (fact_table_id, proid, key_value, source_change_date)
  $$ AS process_queue_snippet,
  
  $$
  WITH all_ids AS
  ($$||raw_queued_changes_sql_out||$$)
  
  , new_metadata AS
  (SELECT MAX(fact_loader_batch_id) AS last_cutoff_id,
    maximum_cutoff_time,
    queue_table_dep_id
  FROM all_ids
  --Exclude dependent fact tables from updates directly to queue_table_deps
  WHERE fact_table_dep_id IS NULL
  GROUP BY queue_table_dep_id, maximum_cutoff_time)
  
  /****
  The dependent fact table uses the same queue_table_id_field as last_cutoff
  We are going to update fact_table_deps metadata instead of queue_table_deps
  ****/
  , new_metadata_fact_dep AS
  (SELECT MAX(fact_loader_batch_id) AS last_cutoff_id,
    maximum_cutoff_time,
    fact_table_dep_queue_table_dep_id
  FROM all_ids
  --Include dependent fact tables only
  WHERE fact_table_dep_id IS NOT NULL
  GROUP BY fact_table_dep_queue_table_dep_id, maximum_cutoff_time)
  
  , update_key AS (
  SELECT qdwr.queue_table_dep_id,
    --Cutoff the id to that newly found, otherwise default to last value
    COALESCE(mu.last_cutoff_id, qdwr.last_cutoff_id) AS last_cutoff_id,
    --This cutoff time must always be the same for all queue tables for given fact table.
    --Even if there are no new records, we move this forward to wherever the stream is at
    qdwr.maximum_cutoff_time AS last_cutoff_source_time
  FROM fact_loader.queue_deps_all qdwr
  LEFT JOIN new_metadata mu ON mu.queue_table_dep_id = qdwr.queue_table_dep_id
  WHERE qdwr.fact_table_id = $$||p_fact_table_id||$$
  --Exclude dependent fact tables from updates directly to queue_table_deps
    AND qdwr.fact_table_dep_id IS NULL
  )
  
  /****
  This SQL also nearly matches that for the queue_table_deps but would be a little ugly to try to DRY up
  ****/
  , update_key_fact_dep AS (
  SELECT qdwr.fact_table_dep_queue_table_dep_id,
    qdwr.fact_table_id,
    COALESCE(mu.last_cutoff_id, qdwr.last_cutoff_id) AS last_cutoff_id,
    qdwr.maximum_cutoff_time AS last_cutoff_source_time
  FROM fact_loader.queue_deps_all qdwr
  LEFT JOIN new_metadata_fact_dep mu ON mu.fact_table_dep_queue_table_dep_id = qdwr.fact_table_dep_queue_table_dep_id
  WHERE qdwr.fact_table_id = $$||p_fact_table_id||$$
  --Include dependent fact tables only
    AND qdwr.fact_table_dep_id IS NOT NULL
  )
  
  , updated_queue_table_deps AS (
  UPDATE fact_loader.queue_table_deps qtd
  SET last_cutoff_id = uk.last_cutoff_id,
    last_cutoff_source_time = uk.last_cutoff_source_time
  FROM update_key uk
  WHERE qtd.queue_table_dep_id = uk.queue_table_dep_id
  RETURNING qtd.*)
  
  , updated_fact_table_deps AS (
  UPDATE fact_loader.fact_table_dep_queue_table_deps ftd
  SET last_cutoff_id = uk.last_cutoff_id,
    last_cutoff_source_time = uk.last_cutoff_source_time
  FROM update_key_fact_dep uk
  WHERE ftd.fact_table_dep_queue_table_dep_id = uk.fact_table_dep_queue_table_dep_id
  RETURNING uk.*)
  
  UPDATE fact_loader.fact_tables ft
  SET last_refresh_source_cutoff = uqtd.last_cutoff_source_time,
    last_refresh_attempted_at = now(),
    last_refresh_succeeded = TRUE
  FROM
  (SELECT fact_table_id, last_cutoff_source_time
  FROM updated_queue_table_deps
  --Must use UNION to get only distinct values
  UNION
  SELECT fact_table_id, last_cutoff_source_time
  FROM updated_fact_table_deps) uqtd
  WHERE uqtd.fact_table_id = ft.fact_table_id;
  $$ AS metadata_update_sql_out
  FROM final_queue_sql)
  
  SELECT raw_queued_changes_sql_out,
   gathered_queued_changes_sql_out
    ,
   format($$
    %s
    %s$$, process_queue_snippet, gathered_queued_changes_sql_out) AS process_queue_sql_out,
   metadata_update_sql_out
  FROM final_outputs;
  
  $BODY$
  LANGUAGE SQL;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.execute_queue(p_fact_table_id INT)
  RETURNS TABLE (sql TEXT) AS
  $BODY$
  BEGIN
  
  RETURN QUERY
  WITH ordered_process_queue AS
  (SELECT
     process_queue_id
     , proid
     , key_value
     , source_change_date
     , (pp.proargtypes::REGTYPE[])[0] AS proid_first_arg
   FROM process_queue pq
     LEFT JOIN pg_proc pp ON pp.oid = proid
   WHERE pq.fact_table_id = p_fact_table_id
   ORDER BY process_queue_id)
  
  , with_rank AS
  (SELECT
    /****
    If source_change_date is NULL, we assume the proid has one arg and pass it.
    If not, we assume the proid has two args and pass source_change_date as the second.
    */
     format('%s(%s::%s%s)'
            , proid::TEXT
            , 'key_value'
            , proid_first_arg
            , CASE
                WHEN source_change_date IS NOT NULL
                  THEN format(', %s::DATE',quote_literal(source_change_date))
                ELSE ''
              END
          ) AS function_call,
    proid,
    process_queue_id,
    RANK() OVER (PARTITION BY proid) AS execution_group
  FROM ordered_process_queue
  )
  
  , execute_sql_groups AS
  (
  SELECT execution_group,
  format($$
  SELECT process_queue_id, %s
  FROM (
  /****
  Must wrap this to execute in order of ids
  ***/
  SELECT *
  FROM process_queue
  WHERE process_queue_id BETWEEN %s AND %s
    AND fact_table_id = %s
    AND proid = %s::REGPROC
  ORDER BY process_queue_id) q;
  $$, function_call, MIN(process_queue_id), MAX(process_queue_id), p_fact_table_id, quote_literal(proid::TEXT)) AS execute_sql
  FROM with_rank
  GROUP BY execution_group, function_call, proid
  ORDER BY execution_group
  )
  
  SELECT COALESCE(string_agg(execute_sql,''),'SELECT NULL') AS final_execute_sql
  FROM execute_sql_groups;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.raw_queued_changes(p_fact_table_id INT)
  RETURNS TABLE (fact_table_id INT,
      queue_table_id INT,
      queue_table_dep_id INT,
      fact_table_dep_id INT,
      fact_table_dep_queue_table_dep_id INT,
      queue_table_id_field BIGINT,
      fact_loader_batch_id BIGINT,
      maximum_cutoff_time TIMESTAMPTZ) AS
  $BODY$
  DECLARE
      v_raw_sql text;
  BEGIN
  
  SELECT raw_queued_changes_sql
  INTO v_raw_sql
  FROM fact_loader.sql_builder(p_fact_table_id);
  
  RETURN QUERY EXECUTE v_raw_sql;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.gathered_queued_changes(p_fact_table_id INT)
  RETURNS TABLE (fact_table_id INT, proid REGPROC, key_value TEXT, source_change_date DATE) AS 
  $BODY$
  DECLARE
      v_gather_sql text;
  BEGIN
  
  SELECT gathered_queued_changes_sql
  INTO v_gather_sql
  FROM fact_loader.sql_builder(p_fact_table_id);
  
  RETURN QUERY EXECUTE v_gather_sql;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.daily_scheduled_load(p_fact_table_id INT)
  RETURNS BOOLEAN AS
  $BODY$
  DECLARE
      v_execute_sql text;
      v_deps regclass[];
      v_dep_delay_tolerance interval;
      v_delayed_msg text;
  BEGIN
  /***
  There are 3 basic steps to this load:
      1. If dependencies are listed, verify they are up to date enough
      2. Execute the single daily-refresh function
      3. Update the metadata indicating the last attempt time
   */
  SELECT 'SELECT '||daily_scheduled_proid::TEXT||'()',
      daily_scheduled_deps,
      daily_scheduled_dep_delay_tolerance
       INTO
      v_execute_sql,
      v_deps,
      v_dep_delay_tolerance
  FROM fact_loader.fact_tables
  WHERE fact_table_id = p_fact_table_id
    AND use_daily_schedule;
  
  IF v_execute_sql IS NULL THEN
      RETURN FALSE;
  END IF;
  
  IF v_deps IS NOT NULL THEN
      WITH deps AS 
      (SELECT unnest(v_deps) AS dep)
      
      , delays AS (
      SELECT dep, now() - source_time as delay_interval 
      FROM fact_loader.queue_table_delay_info() qtd
      INNER JOIN deps d ON d.dep = qtd.queue_of_base_table_relid
      UNION ALL
      SELECT dep, now() - last_refresh_source_cutoff as delay_interval
      FROM fact_loader.fact_tables ft
      INNER JOIN deps d ON d.dep = ft.fact_table_relid
      )
  
      SELECT string_agg(dep::text||': Delayed '||delay_interval::text, ', ')
          INTO v_delayed_msg
      FROM delays
      WHERE delay_interval > v_dep_delay_tolerance;
  
      IF v_delayed_msg IS NOT NULL THEN
          RAISE EXCEPTION '%', v_delayed_msg;
      END IF;
  END IF;
  
  EXECUTE v_execute_sql;
  
  UPDATE fact_loader.fact_tables ft
  SET last_refresh_attempted_at = now(),
      last_refresh_succeeded = TRUE
  WHERE fact_table_id = p_fact_table_id;
  
  RETURN TRUE;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.fact_table_refresh_logs_pruner() RETURNS trigger
  LANGUAGE plpgsql
  AS $$
  
  declare
          step int := 1000;
          -- step should equal the firing frequency in trigger definition
          overdrive int := 2;
          -- overdrive times step = max rows (see below)
  
          max_rows int := step * overdrive;
          rows int;
  
  begin
          delete from fact_loader.fact_table_refresh_logs
          where fact_table_refresh_log_id in (
                  select fact_table_refresh_log_id
                  from fact_loader.fact_table_refresh_logs
                  where refresh_attempted_at < now() - '90 days'::interval
                  -- do not do the literal interval value above as a declare parameter
                  order by fact_table_refresh_log_id
                  limit max_rows
                  for update skip locked
          );
  
          get diagnostics rows = row_count;
          return null;
  end
  $$;
  
  CREATE TRIGGER fact_table_refresh_logs_pruner
  AFTER INSERT ON fact_loader.fact_table_refresh_logs
  FOR EACH ROW WHEN ((new.fact_table_refresh_log_id % 1000::bigint) = 0)
  EXECUTE PROCEDURE fact_loader.fact_table_refresh_logs_pruner();
  
  
  /***
  This function exists mostly to easily mock out for testing purposes.
   */
  CREATE FUNCTION fact_loader.logical_subscription()
  RETURNS TABLE (sub_origin_if OID, sub_replication_sets text[])
  AS $BODY$
  BEGIN
  
  IF EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pglogical') THEN
  
    RETURN QUERY EXECUTE $$
    SELECT sub_origin_if, sub_replication_sets
    FROM pglogical.subscription;
    $$;
  ELSE
    RETURN QUERY
    SELECT NULL::OID, NULL::TEXT[];
  
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  CREATE OR REPLACE FUNCTION fact_loader.queue_table_delay_info()
  RETURNS TABLE("replication_set_name" text,
             "queue_of_base_table_relid" regclass,
             "if_id" oid,
             "if_name" name,
             "source_time" timestamp with time zone)
  AS
  $BODY$
  /***
  This function exists to allow no necessary dependency
  to exist on pglogical_ticker.  If the extension is used,
  it will return data from its native functions, if not,
  it will return a null data set matching the structure
  ***/
  BEGIN
  
  IF EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pglogical_ticker') THEN
      RETURN QUERY EXECUTE $$
      SELECT
          unnest(coalesce(sub_replication_sets,'{NULL}')) AS replication_set_name
        , qt.queue_of_base_table_relid
        , n.if_id
        , n.if_name
        --source_time is now() for local tables (pglogical_node_if_id is null), and based on pglogical_ticker time otherwise
        , CASE
          WHEN qt.pglogical_node_if_id IS NULL
            THEN now()
          ELSE t.source_time
          END                          AS source_time
      FROM fact_loader.queue_tables qt
        LEFT JOIN fact_loader.logical_subscription() s ON qt.pglogical_node_if_id = s.sub_origin_if
        LEFT JOIN pglogical.node_interface n ON n.if_id = qt.pglogical_node_if_id
        LEFT JOIN pglogical_ticker.all_subscription_tickers() t ON t.provider_name = n.if_name;$$;
  ELSE
      RETURN QUERY
      SELECT
          NULL::TEXT AS replication_set_name
        , qt.queue_of_base_table_relid
        , NULL::OID AS if_id
        , NULL::NAME AS if_name
        --source_time is now() if queue tables are not pglogical-replicated, which is assumed if no ticker
        , now() AS source_time
      FROM fact_loader.queue_tables qt;
  END IF;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.refresh_fact_table_dep_queue_table_deps()
  RETURNS VOID AS
  $BODY$
  BEGIN
  /****
  This function will be used to refresh the fact_table_dep_queue_table_deps table.
  The purpose of this table is to easily figure out queue data for fact tables that depend on other fact tables.
  This will be run with every call of load().
  This may not be the most efficient method, but it is certainly reliable and fast.
   */
  
  /****
  Recursively find all fact table deps including nested ones (fact tables that depend on other fact tables)
  to build the fact_table_dep_queue_table_deps table.
   */
  WITH RECURSIVE all_fact_table_deps AS (
    SELECT
      qtd.queue_table_dep_id
      , ftd.fact_table_dep_id
      , parent_id                                 AS parent_fact_table_id
      , child_id                                  AS fact_table_id
      , qtd.queue_table_id
      , qt.queue_table_relid
      , ftp.fact_table_relid AS parent_fact_table
      , ftc.fact_table_relid AS child_fact_table
      , ftd.default_insert_merge_proid
      , ftd.default_update_merge_proid
      , ftd.default_delete_merge_proid
    FROM fact_loader.queue_table_deps qtd
    INNER JOIN fact_loader.queue_tables qt ON qtd.queue_table_id = qt.queue_table_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.parent_id = qtd.fact_table_id
    INNER JOIN fact_loader.fact_tables ftp USING (fact_table_id)
    INNER JOIN fact_loader.fact_tables ftc ON ftc.fact_table_id = ftd.child_id
    UNION ALL
    SELECT
      qtd.queue_table_dep_id
      , ftd.fact_table_dep_id
      , parent_id                                 AS parent_fact_table_id
      , child_id                                  AS fact_table_id
      , qtd.queue_table_id
      , qt.queue_table_relid
      , ftp.fact_table_relid AS parent_fact_table
      , ft.fact_table_relid AS child_fact_table
      , ftd.default_insert_merge_proid
      , ftd.default_update_merge_proid
      , ftd.default_delete_merge_proid
    FROM all_fact_table_deps qtd
    INNER JOIN fact_loader.queue_tables qt ON qtd.queue_table_id = qt.queue_table_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.parent_id = qtd.fact_table_id
    INNER JOIN fact_loader.fact_tables ftp ON ftp.fact_table_id = ftd.parent_id
    INNER JOIN fact_loader.fact_tables ft ON ft.fact_table_id = ftd.child_id
  )
  
  /****
  Remove fact_table_dep_queue_table_deps that no longer exist if applicable
   */
  , removed AS (
    DELETE FROM fact_loader.fact_table_dep_queue_table_deps ftdqc
    WHERE NOT EXISTS(SELECT 1
                     FROM all_fact_table_deps aftd
                     WHERE aftd.fact_table_dep_id = ftdqc.fact_table_dep_id
                      AND aftd.queue_table_dep_id = ftdqc.queue_table_dep_id)
  )
  
  /****
  Add any new keys or ignore if they already exist
  Add not exists because we think allowing all records to insert and conflict could be cause
  of serialization errors in repeatable read isolation.
   */
  INSERT INTO fact_loader.fact_table_dep_queue_table_deps
  (fact_table_dep_id, queue_table_dep_id, insert_merge_proid, update_merge_proid, delete_merge_proid)
  SELECT fact_table_dep_id, queue_table_dep_id, default_insert_merge_proid, default_update_merge_proid, default_delete_merge_proid
  FROM all_fact_table_deps new
  WHERE NOT EXISTS
      (SELECT 1
      FROM fact_loader.fact_table_dep_queue_table_deps existing
      WHERE existing.fact_table_dep_id = new.fact_table_dep_id
        AND existing.queue_table_dep_id = new.queue_table_dep_id)
  ON CONFLICT (fact_table_dep_id, queue_table_dep_id)
  DO NOTHING;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  -- These fields now becomes based on batch, not based on queue_table_id_field
  DO $BODY$
  DECLARE
      v_rec RECORD;
      v_sql TEXT;
  BEGIN
  
  FOR v_rec IN
      SELECT
        format($$
        UPDATE fact_loader.%s
        SET last_cutoff_id =
        (SELECT fact_loader_batch_id
        FROM %s
        WHERE %s = %s)
        WHERE %s = %s;
        $$,
          CASE WHEN fact_table_dep_id IS NULL THEN 'queue_table_deps' ELSE 'fact_table_dep_queue_table_deps' END,
          queue_table_relid::text,
          queue_table_id_field::text,
          last_cutoff_id::text,
          CASE WHEN fact_table_dep_id IS NULL THEN 'queue_table_dep_id' ELSE 'fact_table_dep_queue_table_dep_id' END,
          CASE WHEN fact_table_dep_id IS NULL THEN queue_table_dep_id ELSE fact_table_dep_queue_table_dep_id END
        ) AS sql
      FROM fact_loader.queue_deps_all
      WHERE last_cutoff_id IS NOT NULL
  LOOP
  
  v_sql = v_rec.sql;
  RAISE LOG 'Updating Extension pg_fact_loader Executed: %', v_sql;
  EXECUTE v_sql;
  
  END LOOP;
  
  END$BODY$;
  
  
  COMMENT ON TABLE fact_loader.debug_process_queue IS 'A mirror of process_queue for debugging only (unlogged) - only populated with log_min_duration set to DEBUG.';
  
  
  COMMENT ON TABLE fact_loader.fact_table_dep_queue_table_deps IS
  $$Data in this table is by default auto-generated by refresh_fact_table_dep_queue_table_deps() only for queue-based fact tables that depend on other fact table changes.
  Each row represents a parent's queue_table_dep, updates of which will trickle down to this dependent fact table.  Even though the default proids
  from fact_table_deps are used initially, they may not be appropriate as generalized across all of these queue_table_deps.
  The proids may need to be overridden for individual fact_table_dep_queue_table_deps if that generalization isn't possible.
  See the regression suite in ./sql and ./expected for examples of this.
  $$;
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.fact_table_dep_queue_table_dep_id IS 'Unique identifier';
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.fact_table_dep_id IS 'fact_table_dep for this specific dependency.';
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.queue_table_dep_id IS 'Inherited queue_table_dep that this dependent fact table depends on.';
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.last_cutoff_id IS
    $$This is unique and maintained separately from last_cutoff_id in queue_table_deps,
    as it refers to the last_cutoff_id for this dependent fact table.  It is the last fact_loader_batch_id of
    the queue table that was processed for this queue table - dependent fact table pair.
    After this job runs, records that have this id and lower are eligible to be pruned,
    assuming no other fact tables also depend on those same records.
    The next time the job runs, only records after this id are considered.$$;
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.last_cutoff_source_time IS
    $$This is unique and maintained separately from last_cutoff_source_time in queue_table_deps,
    as it refers to the last_cutoff_source_time for this dependent fact table.  It is the source data
    change time of the last queue table record that was processed for this queue table - dependent fact table pair.
    This helps pg_fact_loader synchronize time across multiple queue tables and only pull changes
    that are early enough, and not purge records that are later than these cutoff times.  It will also
    never go past its parent(s) in time.  THIS DOES NOT DETERMINE filter conditions
    for the starting point at which to pull new records as does last_cutoff_id - it is only
    used as an ending-point barrier.
    $$;
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.insert_merge_proid IS
    $$Initially populated by default_insert_merge_proid from fact_table_deps, but can be
    overridden if a different proid is required. This is the function oid to execute on
    INSERT events *for this dependent fact table* - it accepts a single value as its arg
    which is typically the key that has changed and needs to be updated.
    The way to retrieve this key for this queue table - dependent fact table pair
    is configured in key_retrieval_sequences *for the parent(s)*. NULL to ignore insert events.
    See the regression suite in ./sql and ./expected for examples of this.$$;
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.update_merge_proid IS
    $$Initially populated by default_update_merge_proid from fact_table_deps, but can be
    overridden if a different proid is required. This is the function oid to execute on
    UPDATE events *for this dependent fact table* - it accepts a single value as its arg
    which is typically the key that has changed and needs to be updated.
    The way to retrieve this key for this queue table - dependent fact table pair
    is configured in key_retrieval_sequences *for the parent(s)*. NULL to ignore insert events.
    See the regression suite in ./sql and ./expected for examples of this.$$;
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.delete_merge_proid IS
    $$Initially populated by default_delete_merge_proid from fact_table_deps, but can be
    overridden if a different proid is required. This is the function oid to execute on
    DELETE events *for this dependent fact table* - it accepts a single value as its arg
    which is typically the key that has changed and needs to be updated.
    The way to retrieve this key for this queue table - dependent fact table pair
    is configured in key_retrieval_sequences *for the parent(s)*. NULL to ignore insert events.
    See the regression suite in ./sql and ./expected for examples of this.$$;
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.row_created_at IS 'Timestamp of when this row was first created.';
    COMMENT ON COLUMN fact_loader.fact_table_dep_queue_table_deps.row_updated_at IS 'Timestamp of when this row was last updated (this is updated via trigger).';
  
  
  COMMENT ON TABLE fact_loader.fact_table_deps IS 'For queue-based fact tables that depend on other fact table changes ONLY. Add those dependencies here.';
    COMMENT ON COLUMN fact_loader.fact_table_deps.fact_table_dep_id IS 'Unique identifier.';
    COMMENT ON COLUMN fact_loader.fact_table_deps.parent_id IS 'The parent fact_table_id that the child depends on.';
    COMMENT ON COLUMN fact_loader.fact_table_deps.child_id IS 'The child fact_table_id that will run only after the parent is updated.';
    COMMENT ON COLUMN fact_loader.fact_table_deps.default_insert_merge_proid IS
    $$Default function to use for insert events to update child tables.
    This may need to be modified for each individual inherited fact_table_dep_queue_table_deps
    if that generalization isn't possible. See the regression suite in ./sql and ./expected for examples.$$;
    COMMENT ON COLUMN fact_loader.fact_table_deps.default_update_merge_proid IS
    $$Default function to use for update events to update child tables.
    This may need to be modified for each individual inherited fact_table_dep_queue_table_deps
    if that generalization isn't possible. See the regression suite in ./sql and ./expected for examples.$$;
    COMMENT ON COLUMN fact_loader.fact_table_deps.default_delete_merge_proid IS
    $$Default function to use for delete events to update child tables.
    This may need to be modified for each individual inherited fact_table_dep_queue_table_deps
    if that generalization isn't possible. See the regression suite in ./sql and ./expected for examples.$$;
    COMMENT ON COLUMN fact_loader.fact_table_deps.row_created_at IS 'Timestamp of when this row was first created.';
    COMMENT ON COLUMN fact_loader.fact_table_deps.row_updated_at IS 'Timestamp of when this row was last updated (this is updated via trigger).';
  
  
  COMMENT ON TABLE fact_loader.fact_table_refresh_logs IS 'Used to log both job run times and exceptions.';
    COMMENT ON COLUMN fact_loader.fact_table_refresh_logs.fact_table_refresh_log_id IS 'Unique identifier,';
    COMMENT ON COLUMN fact_loader.fact_table_refresh_logs.fact_table_id IS 'Fact table that created the log.';
    COMMENT ON COLUMN fact_loader.fact_table_refresh_logs.refresh_attempted_at IS 'The time of the attempt (transaction begin time), which can be correlated to fact_table.last_refresh_attempted_at (see also unresolved_failures).';
    COMMENT ON COLUMN fact_loader.fact_table_refresh_logs.refresh_finished_at IS 'The transaction commit time of the attempt, which can be used with refresh_attempted_at to get actual run time.';
    COMMENT ON COLUMN fact_loader.fact_table_refresh_logs.messages IS 'Only for failures - Error message content in JSON format - including message, message detail, context, and hint.';
  
  
  COMMENT ON TABLE fact_loader.fact_tables IS 'Each fact table to be built via pg_fact_loader, which also drives the worker.  These are also referred to as "jobs".';
    COMMENT ON COLUMN fact_loader.fact_tables.fact_table_id IS 'Unique identifier for the fact table or job - also referred to as job_id';
    COMMENT ON COLUMN fact_loader.fact_tables.fact_table_relid IS 'The oid of the fact table itself regclass type to accept only valid relations.';
    COMMENT ON COLUMN fact_loader.fact_tables.fact_table_agg_proid IS
    $$NOT REQUIRED.  The aggregate function definition for the fact table.
    This can be used when passed to create_table_loader_function to auto-create a merge function.
    It can also be a reference for dq checks because it indicates what function returns
    the correct results for a fact table as it should appear now.$$;
    COMMENT ON COLUMN fact_loader.fact_tables.enabled IS 'Indicates whether or not the job is enabled.  The worker will skip this table unless marked TRUE.';
    COMMENT ON COLUMN fact_loader.fact_tables.priority IS 'Determines the order in which the job runs (in combination with other sorting factors)';
    COMMENT ON COLUMN fact_loader.fact_tables.force_worker_priority IS 'If marked TRUE, this fact table will be prioritized in execution order above all other factors.';
    COMMENT ON COLUMN fact_loader.fact_tables.last_refresh_source_cutoff IS 'The data cutoff time of the last refresh - only records older than this have been updated.';
    COMMENT ON COLUMN fact_loader.fact_tables.last_refresh_attempted_at IS 'The last time the worker ran on this fact table.  The oldest will be prioritized first, ahead of priority.';
    COMMENT ON COLUMN fact_loader.fact_tables.last_refresh_succeeded IS 'Whether or not the last run of the job succeeded.  NULL if it has never been run.';
    COMMENT ON COLUMN fact_loader.fact_tables.row_created_at IS 'Timestamp of when this row was first created.';
    COMMENT ON COLUMN fact_loader.fact_tables.row_updated_at IS 'Timestamp of when this row was last updated (this is updated via trigger).';
    COMMENT ON COLUMN fact_loader.fact_tables.use_daily_schedule IS 'If TRUE, this job is scheduled to run daily instead of using queue tables according to other daily column configuration.  Also must be marked TRUE for dependent jobs.';
    COMMENT ON COLUMN fact_loader.fact_tables.daily_scheduled_time IS 'The time of day *after which* to run the job (the system will attempt to run until midnight). If you have a chain of daily scheduled jobs, only the base job has time filled in.';
    COMMENT ON COLUMN fact_loader.fact_tables.daily_scheduled_tz IS 'The timezone your time is in.  This is critical to know when to allow a daily refresh from the standpoint of the business logic you require for a timezone-based date.';
    COMMENT ON COLUMN fact_loader.fact_tables.daily_scheduled_proid IS
    $$The single function oid to execute at the scheduled time.  No arguments supported. It is assumed to contain all the
    logic necessary to add any new daily entries, if applicable.  See the unit tests in sql/16_1_2_features.sql for examples.$$;
    COMMENT ON COLUMN fact_loader.fact_tables.depends_on_base_daily_job_id IS 'For jobs that depend on other daily scheduled jobs only. This is the fact_table_id of the FIRST job in a chain which is actually the only one with a scheduled_time.';
    COMMENT ON COLUMN fact_loader.fact_tables.depends_on_parent_daily_job_id IS 'For jobs that depend on other daily scheduled jobs only. Immediate parent which must complete before this job will run.';
    COMMENT ON COLUMN fact_loader.fact_tables.daily_scheduled_deps IS 'OPTIONAL for daily scheduled jobs.  The only purpose of this column is to consider if we should wait to run a scheduled job because dependent tables are out of date.  This is a regclass array of tables that this scheduled job depends on, which will only be considered if they are either listed in fact_loader.queue_tables or fact_loader.fact_tables.  If the former, replication delay will be considered (if table is not local).  If the latter, last_refresh_source_cutoff will be considered.  Works in combination with daily_scheduled_dep_delay_tolerance which says how much time delay is tolerated.  Job will FAIL if the time delay constraint is not met for all tables - this is intended to be configured as a rare occurrence and thus we want to raise an alarm about it.';
    COMMENT ON COLUMN fact_loader.fact_tables.daily_scheduled_dep_delay_tolerance IS 'OPTIONAL for daily scheduled jobs.  Amount of time interval allowed that dependent tables can be out of date before running this job.  For example, if 10 minutes, then if ANY of the dependent tables are more than 10 minutes out of date, this job will FAIL if the time delay constraint is not met for all tables - this is intended to be configured as a rare occurrence and thus we want to raise an alarm about it.';
  
  
  COMMENT ON TABLE fact_loader.key_retrieval_sequences IS
  $$How to go from a change in the queue table itself to retrieve the key
  that needs to be updated in the fact table.  That key specifically will be passed
  to the insert/update/delete merge_proids configured in queue_table_deps.  When multiple joins
  are required to get there, you will have more than one key_retrieval_sequence for a
  single queue_table_dep.  You can also optionally have a different key_retrieval_sequence
  if your insert/update/delete merge_proids don't all accept the exact same field as an arg.
  NOTE - The regression suite in ./sql and ./expected has abundant examples of different configurations.$$;
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.key_retrieval_sequence_id IS 'Unique identifier.';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.queue_table_dep_id IS 'Which fact table - queue table record this is for (queue_table_deps)';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.filter_scope IS
    $$NULL or one of I, U, D.  Optional and likely rare.  By default, this key_retrieval_sequence
    will tell pg_fact_loader how to get the key for all events - insert, update, delete.
    But if your insert/update/delete merge_proids don't all accept the exact same field as an arg,
    you will have to tell it a different way to retrieve the different I, U, D events on separate rows.
    The regression suite has examples of this.$$;
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.level IS
    $$Default 1. When there are multiple joins required to retrieve a key,
    this indicates the order in which to perform the joins.  It will start at level 1,
    then the return_columns_from_join field will be used to join to the join_to_relation - join_to_column
    for the level 2 record, and so on.$$;
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.return_columns IS
    $$What field to return from the base table (if this is level 1), or (if this level 2+)
    this should be the same as the return_columns_from_join from the previous level.$$;
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.is_fact_key IS 'Only true if the base table itself contains the key. If return_columns contains the keys to pass into the functions without any additional join, TRUE.  Otherwise, FALSE if you need to join to get more information.';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.join_to_relation IS 'Join from the base table (or if this is level 2+, the join_to_relation from the previous level) to this table to get the key or to do yet a further join.';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.join_to_column IS 'Join to this column of join_to_relation.';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.return_columns_from_join IS 'Return these columns from join_to_relation.';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.join_return_is_fact_key IS 'If return_columns_from_join are your fact keys, true.  Otherwise false, and that means you need another level to get your key.';
    COMMENT ON COLUMN fact_loader.key_retrieval_sequences.pass_queue_table_change_date_at_tz IS
    $$If this is set to a time zone, then the changed_at field will be cast to this time zone and then cast to a date,
    for the purpose of creating a date-range based fact table.
    For casting queue_table_timestamp to a date, we first ensure we have it as timestamptz (objective UTC time).
    Then, we cast it to the timezone of interest on which the date should be based.
    For example, 02:00:00 UTC time on 2018-05-02 is actually 2018-05-01 in America/Chicago time.
    Thus, any date-based fact table must decide in what time zone to consider the date.$$;
  
  
  COMMENT ON VIEW fact_loader.queue_deps_all IS 'A view which gathers all fact table data in order to process queued changes and update it, including nested dependencies.';
  
  
  COMMENT ON VIEW fact_loader.queue_deps_all_with_retrieval IS 'The master view which builds on queue_deps_all to include key_retrieval_sequences.  This is the main view used by sql_builder(int) to gather all queued changes.';
  
  
  COMMENT ON TABLE fact_loader.queue_table_deps IS
  $$Ties together which fact tables depend on which queue tables, along with holding
  information on the last cutoff ids for each queue table.  **NOTE** that anything that exists in
  queue_table_dep is assumed to be require its queue data not to be pruned even if the fact_tables
  job is disabled.  That means that even if a job is disabled, you will not lose data, but you will also
  have your queue tables building up in size until you either enable (successfully) or drop the job.
  The regression suite in ./sql and ./expected has abundant examples of different configurations.$$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.queue_table_dep_id IS 'Unique identifier.';
    COMMENT ON COLUMN fact_loader.queue_table_deps.fact_table_id IS 'Fact table to tie together with a queue table it depends on.';
    COMMENT ON COLUMN fact_loader.queue_table_deps.queue_table_id IS 'Queue table to tie together with a fact table that needs its changes.';
    COMMENT ON COLUMN fact_loader.queue_table_deps.relevant_change_columns IS
    $$Optional. For UPDATE changes to data, you can specify to only consider changes
    to these columns as sufficient to update the fact table.
    If NULL, all columns will be considered as potentially changing the fact table data.$$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.last_cutoff_id IS
    $$The last fact_loader_batch_id of the queue table that was processed for this queue table - fact table pair.
    After this job runs, records that have this id and lower are eligible to be pruned,
    assuming no other fact tables also depend on those same records.
    The next time the job runs, only records after this id are considered.$$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.last_cutoff_source_time IS
    $$The source data change time of the last queue table record that was processed for this
    queue table - fact table pair.  This helps pg_fact_loader synchronize time across
    multiple queue tables and only pull changes that are early enough, and not purge
    records that are later than these cutoff times.  THIS DOES NOT DETERMINE filter conditions
    for the starting point at which to pull new records as does last_cutoff_id - it is only
    used as an ending-point barrier.
    $$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.insert_merge_proid IS
    $$Function oid to execute on insert events - accepts a single value as its arg
    which is typically the key that has changed and needs to be updated.
    The way to retrieve this key for this queue table - fact table pair
    is configured in key_retrieval_sequences. NULL to ignore insert events.$$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.update_merge_proid IS
    $$Function oid to execute on update events - accepts a single value as its arg
    which is typically the key that has changed and needs to be updated.
    The way to retrieve this key for this queue table - fact table pair
    is configured in key_retrieval_sequences. NULL to ignore update events.$$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.delete_merge_proid IS
    $$Function oid to execute on delete events - accepts a single value as its arg
    which is typically the key that has changed and needs to be updated.
    The way to retrieve this key for this queue table - fact table pair
    is configured in key_retrieval_sequences. NULL to ignore delete events.$$;
    COMMENT ON COLUMN fact_loader.queue_table_deps.row_created_at IS 'Timestamp of when this row was first created.';
    COMMENT ON COLUMN fact_loader.queue_table_deps.row_updated_at IS 'Timestamp of when this row was last updated (this is updated via trigger).';
  
  
  COMMENT ON TABLE fact_loader.queue_tables IS 'Each queue table along with the base table to which it belongs.';
    COMMENT ON COLUMN fact_loader.queue_tables.queue_table_id IS 'Unique identifier for queue tables.';
    COMMENT ON COLUMN fact_loader.queue_tables.queue_table_relid IS 'The oid of the queue table itself regclass type to accept only valid relations.';
    COMMENT ON COLUMN fact_loader.queue_tables.queue_of_base_table_relid IS 'The oid of the base table for which the queue table contains an audited log of changes.  regclass type to accept only valid relations.';
    COMMENT ON COLUMN fact_loader.queue_tables.pglogical_node_if_id IS
    $$Optional - If NULL, we assume this is a local queue table and we need not synchronize time
    for potential replication delay.  For use with tables that are replicated via pglogical.
    This is the pglogical.node_interface of the table.  This also requires pglogical_ticker
    and is used to synchronize time and ensure we don't continue to move forward
    in time when replication is delayed for this queue table.$$;
    COMMENT ON COLUMN fact_loader.queue_tables.queue_table_tz IS
    $$**NOTE CAREFULLY** - If this is NULL, it assumes that changed_at in the queue
    tables is stored in TIMESTAMPTZ.  If it IS set, it assumes you are telling it that changed_at is
    of TIMESTAMP data type which is stored in the provided time zone of queue_table_tz.$$;
    COMMENT ON COLUMN fact_loader.queue_tables.row_created_at IS 'Timestamp of when this row was first created.';
    COMMENT ON COLUMN fact_loader.queue_tables.row_updated_at IS 'Timestamp of when this row was last updated (this is updated via trigger).';
    COMMENT ON COLUMN fact_loader.queue_tables.purge IS 'Default is true because we prune queue tables as data is no longer needed. Can be set to false and no pruning will happen on this table.';
  
  
  COMMENT ON VIEW fact_loader.unresolved_failures IS 'Will only show fact table and error messages for a job that just failed and has not been re-enabled since last failure.  Useful for monitoring.';
  
  
  /* pg_fact_loader--1.3--1.4.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  DROP FUNCTION fact_loader.raw_queued_changes(int);
  ALTER TABLE fact_loader.debug_process_queue DROP CONSTRAINT debug_process_queue_pkey;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.load(p_fact_table_id INT)
  RETURNS VOID AS
  $BODY$
  DECLARE
      v_process_queue_sql text;
      v_execute_sql text;
      v_metadata_update_sql text;
      v_debug_rec record;
      v_debug_text text = '';
  BEGIN
  /***
  There are 3 basic steps to this load:
      1. Gather all queue table changes and insert them into a consolidated process_queue
      2. Update the metadata indicating the last records updated for both the queue tables and fact table
   */
  
  /****
  Get SQL to insert new data into the consolidated process_queue,
  and SQL to update metadata for last_cutoffs.
   */
  SELECT process_queue_sql, metadata_update_sql
  INTO v_process_queue_sql, v_metadata_update_sql
  FROM fact_loader.sql_builder(p_fact_table_id);
  
  /****
  Populate the consolidated queue
  This just creates a temp table with all changes to be processed
   */
  RAISE DEBUG 'Populating Queue for fact_table_id %: %', p_fact_table_id, v_process_queue_sql;
  EXECUTE COALESCE(v_process_queue_sql, $$SELECT 'No queue data' AS result$$);
  
  /****
  For DEBUG purposes only to view the actual process_queue.  Requires setting log_min_messages to DEBUG.
   */
  IF current_setting('log_min_messages') = 'debug3' THEN
      INSERT INTO fact_loader.debug_process_queue (process_queue_id, fact_table_id, proid, key_value, row_created_at, row_updated_at, source_change_date)
      -- the row timestamps are not populated, so we set them here
      SELECT process_queue_id, fact_table_id, proid, key_value, now(), now(), source_change_date FROM process_queue;
  END IF;
  
  /****
  With data now in the process_queue, the execute_queue function builds the SQL to execute.
  Save this SQL in a variable and execute it.
  If there is no data to execute, this is a no-op select statement.
   */
  SELECT sql INTO v_execute_sql FROM fact_loader.execute_queue(p_fact_table_id);
  RAISE DEBUG 'Executing Queue for fact_table_id %: %', p_fact_table_id, v_execute_sql;
  EXECUTE COALESCE(v_execute_sql, $$SELECT 'No queue data to execute' AS result$$);
  
  /****
  With everything finished, we now update the metadata for the fact_table.
  Even if no data was processed, we will still move forward last_refresh_attempted_at.
  
  last_refresh_succeeded will be marked true always for now.  It could in the future
  be used to indicate a failure in case of a caught error.
   */
  RAISE DEBUG 'Updating metadata for fact_table_id %: %', p_fact_table_id, v_metadata_update_sql;
  EXECUTE COALESCE(v_metadata_update_sql,
      format(
      $$UPDATE fact_loader.fact_tables ft
          SET last_refresh_attempted_at = now(),
            last_refresh_succeeded = TRUE
       WHERE fact_table_id = %s;
      $$, p_fact_table_id));
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.sql_builder(p_fact_table_id INT)
    RETURNS TABLE(raw_queued_changes_sql text,
                  gathered_queued_changes_sql text,
                  process_queue_sql text,
                  metadata_update_sql text) AS
  $BODY$
  
  /****
  The recursive part of this CTE are only the sql_builder parts.
  In Postgres, if any of your CTEs are recursive, you only use the RECURSIVE keyword on the first of a set.
  
  The retrieval info may be the same for all 3 events (insert, update, delete), in which case filter_scope is null
  Otherwise, they must be specified separately.
   */
  WITH RECURSIVE queue_deps_with_insert_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'I' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  , queue_deps_with_update_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'U' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  , queue_deps_with_delete_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'D' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  /****
  Recursively build the SQL for any INSERT events found in the queues.
  
  The recursive part ONLY applies to cases where multiple joins have to be made to get at the source data,
  in which case there are multiple levels of key_retrieval_sequences for a given queue_table_dep_id. For an
  example of this, see the test cases involving the test.order_product_promos table.
   */
  , insert_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', b.'||return_columns[1]||'::TEXT AS key'
              ELSE ', unnest(array[b.'||array_to_string(return_columns, ',b.')||'])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||return_columns_from_join[1]||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level)||'])::TEXT AS key'
            END
          ELSE ''
      END AS key_select_column,
      CASE
        WHEN is_fact_key
          THEN ''
          ELSE 'INNER JOIN '||join_to_relation::TEXT||' j'||level||
               E'\n  ON b.'||quote_ident(return_columns[1])||' = j'||level||'.'||quote_ident(join_to_column)
      END AS key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_insert_retrieval c
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.'||return_columns[1]||'::TEXT AS key'
            ELSE ', unnest(b.'||array_to_string(return_columns,',j'||r.level)||')::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||return_columns_from_join[1]||'::TEXT AS key'
            ELSE ', unnest(j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS key_select_column,
      key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE E'\nINNER JOIN '||join_to_relation::TEXT||' j'||c.level||
             E'\n  ON j'||r.level||'.'||quote_ident(return_columns[1])||' = j'||c.level||'.'||quote_ident(join_to_column) END AS key_retrieval_sql,
      r.source_change_date_select
    FROM insert_sql_builder r
    INNER JOIN queue_deps_with_insert_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , update_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', b.'||return_columns[1]||'::TEXT AS key'
              ELSE ', unnest(array[b.'||array_to_string(return_columns, ',b.')||'])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||return_columns_from_join[1]||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level)||'])::TEXT AS key'
            END
          ELSE ''
      END AS key_select_column,
      CASE
        WHEN is_fact_key
          THEN ''
          ELSE 'INNER JOIN '||join_to_relation::TEXT||' j'||level||
               E'\n  ON b.'||quote_ident(return_columns[1])||' = j'||level||'.'||quote_ident(join_to_column)
      END AS key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_update_retrieval c
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.'||return_columns[1]||'::TEXT AS key'
            ELSE ', unnest(b.'||array_to_string(return_columns,',j'||r.level)||')::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||return_columns_from_join[1]||'::TEXT AS key'
            ELSE ', unnest(j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS key_select_column,
      key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE E'\nINNER JOIN '||join_to_relation::TEXT||' j'||c.level||
             E'\n  ON j'||r.level||'.'||quote_ident(return_columns[1])||' = j'||c.level||'.'||quote_ident(join_to_column) END AS key_retrieval_sql,
      r.source_change_date_select
    FROM update_sql_builder r
    INNER JOIN queue_deps_with_update_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , delete_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      --For deletes, same pattern as key_select_column but instead, we may be selecting from the audit tables instead
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', q.before_change->>'||quote_literal(return_columns[1])||'::TEXT AS key'
              ELSE ', unnest(array[before_change->>'''||array_to_string(return_columns, ''', before_change->>''')||'''])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||quote_ident(return_columns_from_join[1])||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level||'.')||'])::TEXT AS key'
            END
          ELSE ''
      END AS delete_key_select_column,
      CASE
      WHEN is_fact_key
        THEN ''
        ELSE format($$
          --Join to either the base table, or the audit table, one of which
          --will be missing the key in a delete case
            INNER JOIN LATERAL (
             SELECT %s FROM %s jb
             WHERE %s = %s
             UNION ALL
             SELECT %s FROM %s jq
             WHERE operation = 'D'
              AND %s = %s) %s ON TRUE
             $$, quote_ident(return_columns_from_join[1]),
              join_to_relation::TEXT,
              (CASE WHEN level = 1 THEN '(q'||'.before_change->>'||quote_literal(return_columns[1])||')::'||join_column_type ELSE 'j'||level||'.'||quote_ident(return_columns[1]) END),
              'jb.'||quote_ident(join_to_column),
              '(before_change->>'||quote_literal(return_columns_from_join[1])||')::'||return_columns_from_join_type,
              join_to_relation_queue::TEXT,
              (CASE WHEN level = 1 THEN '(q'||'.before_change->>'||quote_literal(return_columns[1])||')::'||join_column_type ELSE 'j'||level||'.'||quote_ident(return_columns[1]) END),
              '(jq.before_change->>'||quote_literal(join_to_column)||')::'||join_column_type,
              /****
              We use the higher level here just to be consistent with aliases from insert/update key retrieval
               */
              'j'||level
              )
      END
        AS delete_key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_delete_retrieval
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      delete_key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.before_change->>'||quote_literal(return_columns[1])||'::TEXT AS key'
            ELSE ', unnest(array[before_change->>'''||array_to_string(return_columns,',j'||r.level||'.before_change->>''')||'''])::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||quote_ident(return_columns_from_join[1])||'::TEXT AS key'
            ELSE ', unnest(array[j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS delete_key_select_column,
      delete_key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE format($$
          --Join to either the base table, or the audit table, one of which
          --will be missing the key in a delete case
            INNER JOIN LATERAL (
             SELECT %s FROM %s jb
             WHERE %s = %s
             UNION ALL
             SELECT %s FROM %s jq
             WHERE operation = 'D'
              AND %s = %s) %s ON TRUE
             $$, quote_ident(return_columns_from_join[1]),
              join_to_relation::TEXT,
              'j'||r.level||'.'||quote_ident(return_columns[1]),
              'jb.'||quote_ident(join_to_column),
              '(before_change->>'||quote_literal(return_columns_from_join[1])||')::'||return_columns_from_join_type,
              join_to_relation_queue::TEXT,
              'j'||r.level||'.'||quote_ident(return_columns[1]),
              '(jq.before_change->>'||quote_literal(join_to_column)||')::'||join_column_type,
              /****
              We use the higher level here just to be consistent with aliases from insert/update key retrieval
               */
              'j'||c.level
              )
      END
        AS delete_key_retrieval_sql,
      r.source_change_date_select
    FROM delete_sql_builder r
    INNER JOIN queue_deps_with_delete_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , field_vars AS (
    SELECT
      *,
      format($$
        %s AS fact_table_id,
        %s AS queue_table_dep_id,
        %s::INT AS fact_table_dep_id,
        %s::INT AS fact_table_dep_queue_table_dep_id,
        %s AS queue_table_id_field,
        q.fact_loader_batch_id,
        %s::TIMESTAMPTZ AS maximum_cutoff_time,
      -- We must not ignore ids which are above maximum_cutoff_time
      -- but below the highest id which is below maximum_cutoff_time
        MIN(q.fact_loader_batch_id)
        FILTER (
        WHERE %s %s > %s::TIMESTAMPTZ)
        OVER() AS min_missed_id
      $$,
        fact_table_id,
        queue_table_dep_id,
        (CASE WHEN fact_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_id::TEXT END),
        (CASE WHEN fact_table_dep_queue_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_queue_table_dep_id::TEXT END),
        'q.'||quote_ident(queue_table_id_field),
        quote_literal(maximum_cutoff_time),
        'q.'||quote_ident(queue_table_timestamp),
        CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END,
        quote_literal(maximum_cutoff_time)
      )
        AS inner_shared_select_columns,
      $$
        fact_table_id,
        queue_table_dep_id,
        fact_table_dep_id,
        fact_table_dep_queue_table_dep_id,
        queue_table_id_field,
        fact_loader_batch_id,
        maximum_cutoff_time,
        min_missed_id
      $$
        AS outer_shared_select_columns,
      CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END
        AS changed_at_tz_correction
    FROM fact_loader.queue_deps_all c
    WHERE c.fact_table_id = p_fact_table_id
  )
  
  , non_recursive_sql AS (
    SELECT
    /****
    Separate select list for:
      - raw queue_ids from queue tables
      - gathered data from joining queue_ids to source tables to get actual keys to update in fact tables
     */
    -- gathering all queue_ids from queue tables
      queue_table_dep_id,
      outer_shared_select_columns,
      format($$
        %s,
        %s %s AS changed_at,
        %s AS queue_table_id
      $$,
        inner_shared_select_columns,
        'q.'||quote_ident(queue_table_timestamp),
        changed_at_tz_correction,
        queue_table_id
      )
        AS inner_metadata_select_columns,
      format($$
        %s,
        queue_table_id
      $$,
        outer_shared_select_columns
      )
        AS outer_metadata_select_columns,
  
    -- gathering actual keys to update in fact tables by joining from queue_ids to source tables
      format($$
        %s,
        %s AS operation,
        %s %s AS changed_at,
        %s::REGPROC AS insert_merge_proid,
        %s::REGPROC AS update_merge_proid,
        %s::REGPROC AS delete_merge_proid
      $$,
        inner_shared_select_columns,
        'q.'||quote_ident(queue_table_op),
        'q.'||quote_ident(queue_table_timestamp),
        changed_at_tz_correction,
        CASE WHEN insert_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(insert_merge_proid) END,
        CASE WHEN update_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(update_merge_proid) END,
        CASE WHEN delete_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(delete_merge_proid) END
      )
        AS inner_data_select_columns,
      format($$
        %s,
        operation,
        changed_at,
        insert_merge_proid,
        update_merge_proid,
        delete_merge_proid,
        key,
        source_change_date
      $$,
        outer_shared_select_columns
      )
        AS outer_data_select_columns,
  
    -- This is simply the queue table aliased as q
      format('%s q', queue_table_relid::TEXT) AS queue_table_aliased,
  
    -- This is the SQL to join from the queue table to the base table
      format($$
        INNER JOIN %s b
          ON q.%s::%s = b.%s
      $$,
        queue_of_base_table_relid::TEXT,
        quote_ident(queue_table_key),
        queue_of_base_table_key_type,
        quote_ident(queue_of_base_table_key))
             AS base_join_sql,
  
    -- This is a WHERE statement to be added to ALL gathering of new queue_ids to process.
    -- There is a further filter based on the window min_missed_id after this subquery
      format($$ %s
          $$,
          CASE
            WHEN last_cutoff_id IS NOT NULL
            THEN 'q.fact_loader_batch_id > '||last_cutoff_id
            ELSE
              'TRUE'
          END)
             AS inner_global_where_sql,
      format($$
          %s < %s %s
          AND (min_missed_id IS NULL OR (fact_loader_batch_id < min_missed_id))
          $$,
          quote_ident(c.queue_table_timestamp),
          quote_literal(c.maximum_cutoff_time),
          changed_at_tz_correction)
             AS outer_global_where_sql,
      format($$
        AND q.%s = 'I'
        $$,
          queue_table_op)
             AS where_for_insert_sql,
      format($$
        AND (q.%s = 'U' AND %s)
        $$,
          queue_table_op,
          CASE
            WHEN relevant_change_columns IS NULL
              THEN 'TRUE'
            ELSE
              format($$q.%s ?| '{%s}'$$, queue_table_change, array_to_string(relevant_change_columns,','))
          END)
             AS where_for_update_sql,
      format($$
        AND q.%s = 'D'
        $$,
          queue_table_op)
             AS where_for_delete_sql
    FROM field_vars c
  )
  
  , insert_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM insert_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , update_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM update_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , delete_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM delete_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , all_queues_sql AS (
  SELECT
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_data_select_columns,
      nrs.inner_data_select_columns||isbf.key_select_column||isbf.source_change_date_select,
      nrs.queue_table_aliased||nrs.base_join_sql,
      isbf.key_retrieval_sql,
      nrs.inner_global_where_sql||nrs.where_for_insert_sql,
      nrs.outer_global_where_sql) AS queue_insert_sql,
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_data_select_columns,
      nrs.inner_data_select_columns||usbf.key_select_column||usbf.source_change_date_select,
      nrs.queue_table_aliased||nrs.base_join_sql,
      usbf.key_retrieval_sql,
      nrs.inner_global_where_sql||nrs.where_for_update_sql,
      nrs.outer_global_where_sql) AS queue_update_sql,
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_data_select_columns,
      nrs.inner_data_select_columns||dsbf.delete_key_select_column||usbf.source_change_date_select,
      nrs.queue_table_aliased,
      dsbf.delete_key_retrieval_sql,
      nrs.inner_global_where_sql||nrs.where_for_delete_sql,
      nrs.outer_global_where_sql) AS queue_delete_sql,
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_metadata_select_columns,
      nrs.inner_metadata_select_columns,
      nrs.queue_table_aliased,
      nrs.inner_global_where_sql,
      nrs.outer_global_where_sql) AS queue_ids_sql
  FROM non_recursive_sql nrs
  INNER JOIN insert_sql_builder_final isbf ON isbf.queue_table_dep_id = nrs.queue_table_dep_id
  INNER JOIN update_sql_builder_final usbf ON usbf.queue_table_dep_id = nrs.queue_table_dep_id
  INNER JOIN delete_sql_builder_final dsbf ON dsbf.queue_table_dep_id = nrs.queue_table_dep_id
  )
  
  , final_queue_sql AS
  (SELECT string_agg(
    /****
    This first UNION is to union together INSERT, UPDATE, and DELETE events for a single queue table
     */
    format($$
    %s
    UNION ALL
    %s
    UNION ALL
    %s
    $$,
      queue_insert_sql,
      queue_update_sql,
      queue_delete_sql)
    /****
    This second UNION as the second arg of string_agg is the union together ALL queue tables for this fact table
     */
    , E'\nUNION ALL\n') AS event_sql,
    string_agg(queue_ids_sql, E'\nUNION ALL\n') AS raw_queued_changes_sql_out
  FROM all_queues_sql)
  
  , final_outputs AS (
  SELECT raw_queued_changes_sql_out,
  $$
  WITH all_changes AS (
  ($$||event_sql||$$)
  ORDER BY changed_at)
  
  , base_execution_groups AS
  (SELECT fact_table_id,
      queue_table_dep_id,
      queue_table_id_field,
      operation,
      changed_at,
      source_change_date,
      insert_merge_proid,
      update_merge_proid,
      delete_merge_proid,
      maximum_cutoff_time,
      key,
      CASE WHEN operation = 'I' THEN insert_merge_proid
      WHEN operation = 'U' THEN update_merge_proid
      WHEN operation = 'D' THEN delete_merge_proid
          END AS proid,
    RANK() OVER (
      PARTITION BY
        CASE
          WHEN operation = 'I' THEN insert_merge_proid
          WHEN operation = 'U' THEN update_merge_proid
          WHEN operation = 'D' THEN delete_merge_proid
        END
    ) AS execution_group
    FROM all_changes
    WHERE key IS NOT NULL)
  
  SELECT fact_table_id, proid, key, source_change_date
  FROM base_execution_groups beg
  WHERE proid IS NOT NULL
  GROUP BY execution_group, fact_table_id, proid, key, source_change_date
  /****
  This ordering is particularly important for date-range history tables
  where order of inserts is critical and usually expected to follow a pattern
  ***/
  ORDER BY execution_group, MIN(changed_at), MIN(queue_table_id_field);
  $$ AS gathered_queued_changes_sql_out
    ,
  
  $$
  DROP TABLE IF EXISTS process_queue;
  CREATE TEMP TABLE process_queue
  (process_queue_id serial,
   fact_table_id int,
   proid regproc,
   key_value text,
   source_change_date date);
  
  INSERT INTO process_queue
  (fact_table_id, proid, key_value, source_change_date)
  $$ AS process_queue_snippet,
  
  $$
  WITH all_ids AS
  ($$||raw_queued_changes_sql_out||$$)
  
  , new_metadata AS
  (SELECT MAX(fact_loader_batch_id) AS last_cutoff_id,
    maximum_cutoff_time,
    queue_table_dep_id
  FROM all_ids
  --Exclude dependent fact tables from updates directly to queue_table_deps
  WHERE fact_table_dep_id IS NULL
  GROUP BY queue_table_dep_id, maximum_cutoff_time)
  
  /****
  The dependent fact table uses the same queue_table_id_field as last_cutoff
  We are going to update fact_table_deps metadata instead of queue_table_deps
  ****/
  , new_metadata_fact_dep AS
  (SELECT MAX(fact_loader_batch_id) AS last_cutoff_id,
    maximum_cutoff_time,
    fact_table_dep_queue_table_dep_id
  FROM all_ids
  --Include dependent fact tables only
  WHERE fact_table_dep_id IS NOT NULL
  GROUP BY fact_table_dep_queue_table_dep_id, maximum_cutoff_time)
  
  , update_key AS (
  SELECT qdwr.queue_table_dep_id,
    --Cutoff the id to that newly found, otherwise default to last value
    COALESCE(mu.last_cutoff_id, qdwr.last_cutoff_id) AS last_cutoff_id,
    --This cutoff time must always be the same for all queue tables for given fact table.
    --Even if there are no new records, we move this forward to wherever the stream is at
    qdwr.maximum_cutoff_time AS last_cutoff_source_time
  FROM fact_loader.queue_deps_all qdwr
  LEFT JOIN new_metadata mu ON mu.queue_table_dep_id = qdwr.queue_table_dep_id
  WHERE qdwr.fact_table_id = $$||p_fact_table_id||$$
  --Exclude dependent fact tables from updates directly to queue_table_deps
    AND qdwr.fact_table_dep_id IS NULL
  )
  
  /****
  This SQL also nearly matches that for the queue_table_deps but would be a little ugly to try to DRY up
  ****/
  , update_key_fact_dep AS (
  SELECT qdwr.fact_table_dep_queue_table_dep_id,
    qdwr.fact_table_id,
    COALESCE(mu.last_cutoff_id, qdwr.last_cutoff_id) AS last_cutoff_id,
    qdwr.maximum_cutoff_time AS last_cutoff_source_time
  FROM fact_loader.queue_deps_all qdwr
  LEFT JOIN new_metadata_fact_dep mu ON mu.fact_table_dep_queue_table_dep_id = qdwr.fact_table_dep_queue_table_dep_id
  WHERE qdwr.fact_table_id = $$||p_fact_table_id||$$
  --Include dependent fact tables only
    AND qdwr.fact_table_dep_id IS NOT NULL
  )
  
  , updated_queue_table_deps AS (
  UPDATE fact_loader.queue_table_deps qtd
  SET last_cutoff_id = uk.last_cutoff_id,
    last_cutoff_source_time = uk.last_cutoff_source_time
  FROM update_key uk
  WHERE qtd.queue_table_dep_id = uk.queue_table_dep_id
  RETURNING qtd.*)
  
  , updated_fact_table_deps AS (
  UPDATE fact_loader.fact_table_dep_queue_table_deps ftd
  SET last_cutoff_id = uk.last_cutoff_id,
    last_cutoff_source_time = uk.last_cutoff_source_time
  FROM update_key_fact_dep uk
  WHERE ftd.fact_table_dep_queue_table_dep_id = uk.fact_table_dep_queue_table_dep_id
  RETURNING uk.*)
  
  UPDATE fact_loader.fact_tables ft
  SET last_refresh_source_cutoff = uqtd.last_cutoff_source_time,
    last_refresh_attempted_at = now(),
    last_refresh_succeeded = TRUE
  FROM
  (SELECT fact_table_id, last_cutoff_source_time
  FROM updated_queue_table_deps
  --Must use UNION to get only distinct values
  UNION
  SELECT fact_table_id, last_cutoff_source_time
  FROM updated_fact_table_deps) uqtd
  WHERE uqtd.fact_table_id = ft.fact_table_id;
  $$ AS metadata_update_sql_out
  FROM final_queue_sql)
  
  SELECT raw_queued_changes_sql_out,
   gathered_queued_changes_sql_out
    ,
   format($$
    %s
    %s$$, process_queue_snippet, gathered_queued_changes_sql_out) AS process_queue_sql_out,
   metadata_update_sql_out
  FROM final_outputs;
  
  $BODY$
  LANGUAGE SQL;
  
  
  CREATE OR REPLACE FUNCTION fact_loader.raw_queued_changes(p_fact_table_id INT)
  RETURNS TABLE (fact_table_id INT,
      queue_table_dep_id INT,
      fact_table_dep_id INT,
      fact_table_dep_queue_table_dep_id INT,
      queue_table_id_field BIGINT,
      fact_loader_batch_id BIGINT,
      maximum_cutoff_time TIMESTAMPTZ,
      min_missed_id BIGINT,
      queue_table_id INT
  ) AS
  $BODY$
  DECLARE
      v_raw_sql text;
  BEGIN
  
  SELECT raw_queued_changes_sql
  INTO v_raw_sql
  FROM fact_loader.sql_builder(p_fact_table_id);
  
  RETURN QUERY EXECUTE v_raw_sql;
  
  END;
  $BODY$
  LANGUAGE plpgsql;
  
  
  COMMENT ON VIEW fact_loader.queue_deps_all IS 'A view which gathers all fact table data in order to process queued changes and update it, including nested dependencies.';
  
  
  /* pg_fact_loader--1.4--1.5.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  CREATE OR REPLACE FUNCTION fact_loader.sql_builder(p_fact_table_id INT)
    RETURNS TABLE(raw_queued_changes_sql text,
                  gathered_queued_changes_sql text,
                  process_queue_sql text,
                  metadata_update_sql text) AS
  $BODY$
  
  /****
  The recursive part of this CTE are only the sql_builder parts.
  In Postgres, if any of your CTEs are recursive, you only use the RECURSIVE keyword on the first of a set.
  
  The retrieval info may be the same for all 3 events (insert, update, delete), in which case filter_scope is null
  Otherwise, they must be specified separately.
   */
  WITH RECURSIVE queue_deps_with_insert_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'I' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  , queue_deps_with_update_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'U' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  , queue_deps_with_delete_retrieval AS (
    SELECT *
    FROM fact_loader.queue_deps_all_with_retrieval
    WHERE (filter_scope = 'D' OR filter_scope IS NULL)
      AND fact_table_id = p_fact_table_id
  )
  
  /****
  Recursively build the SQL for any INSERT events found in the queues.
  
  The recursive part ONLY applies to cases where multiple joins have to be made to get at the source data,
  in which case there are multiple levels of key_retrieval_sequences for a given queue_table_dep_id. For an
  example of this, see the test cases involving the test.order_product_promos table.
   */
  , insert_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', b.'||return_columns[1]||'::TEXT AS key'
              ELSE ', unnest(array[b.'||array_to_string(return_columns, ',b.')||'])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||return_columns_from_join[1]||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level)||'])::TEXT AS key'
            END
          ELSE ''
      END AS key_select_column,
      CASE
        WHEN is_fact_key
          THEN ''
          ELSE 'INNER JOIN '||join_to_relation::TEXT||' j'||level||
               E'\n  ON b.'||quote_ident(return_columns[1])||' = j'||level||'.'||quote_ident(join_to_column)
      END AS key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_insert_retrieval c
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.'||return_columns[1]||'::TEXT AS key'
            ELSE ', unnest(b.'||array_to_string(return_columns,',j'||r.level)||')::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||return_columns_from_join[1]||'::TEXT AS key'
            ELSE ', unnest(j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS key_select_column,
      key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE E'\nINNER JOIN '||join_to_relation::TEXT||' j'||c.level||
             E'\n  ON j'||r.level||'.'||quote_ident(return_columns[1])||' = j'||c.level||'.'||quote_ident(join_to_column) END AS key_retrieval_sql,
      r.source_change_date_select
    FROM insert_sql_builder r
    INNER JOIN queue_deps_with_insert_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , update_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', b.'||return_columns[1]||'::TEXT AS key'
              ELSE ', unnest(array[b.'||array_to_string(return_columns, ',b.')||'])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||return_columns_from_join[1]||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level)||'])::TEXT AS key'
            END
          ELSE ''
      END AS key_select_column,
      CASE
        WHEN is_fact_key
          THEN ''
          ELSE 'INNER JOIN '||join_to_relation::TEXT||' j'||level||
               E'\n  ON b.'||quote_ident(return_columns[1])||' = j'||level||'.'||quote_ident(join_to_column)
      END AS key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_update_retrieval c
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.'||return_columns[1]||'::TEXT AS key'
            ELSE ', unnest(b.'||array_to_string(return_columns,',j'||r.level)||')::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||return_columns_from_join[1]||'::TEXT AS key'
            ELSE ', unnest(j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS key_select_column,
      key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE E'\nINNER JOIN '||join_to_relation::TEXT||' j'||c.level||
             E'\n  ON j'||r.level||'.'||quote_ident(return_columns[1])||' = j'||c.level||'.'||quote_ident(join_to_column) END AS key_retrieval_sql,
      r.source_change_date_select
    FROM update_sql_builder r
    INNER JOIN queue_deps_with_update_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , delete_sql_builder AS (
    SELECT queue_table_dep_id,
      level,
      --For deletes, same pattern as key_select_column but instead, we may be selecting from the audit tables instead
      CASE
        WHEN is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', q.before_change->>'||quote_literal(return_columns[1])||'::TEXT AS key'
              ELSE ', unnest(array[before_change->>'''||array_to_string(return_columns, ''', before_change->>''')||'''])::TEXT AS key'
            END
        WHEN join_return_is_fact_key
          THEN
            CASE
              WHEN array_length(return_columns, 1) = 1
              THEN ', j'||level||'.'||quote_ident(return_columns_from_join[1])||'::TEXT AS key'
              ELSE ', unnest(array[j'||level||'.'||array_to_string(return_columns_from_join, ',j'||level||'.')||'])::TEXT AS key'
            END
          ELSE ''
      END AS delete_key_select_column,
      CASE
      WHEN is_fact_key
        THEN ''
        ELSE format($$
          --Join to either the base table, or the audit table, one of which
          --will be missing the key in a delete case
            INNER JOIN LATERAL (
             SELECT %s FROM %s jb
             WHERE %s = %s
             UNION ALL
             SELECT %s FROM %s jq
             WHERE operation = 'D'
              AND %s = %s) %s ON TRUE
             $$, quote_ident(return_columns_from_join[1]),
              join_to_relation::TEXT,
              (CASE WHEN level = 1 THEN '(q'||'.before_change->>'||quote_literal(return_columns[1])||')::'||join_column_type ELSE 'j'||level||'.'||quote_ident(return_columns[1]) END),
              'jb.'||quote_ident(join_to_column),
              '(before_change->>'||quote_literal(return_columns_from_join[1])||')::'||return_columns_from_join_type,
              join_to_relation_queue::TEXT,
              (CASE WHEN level = 1 THEN '(q'||'.before_change->>'||quote_literal(return_columns[1])||')::'||join_column_type ELSE 'j'||level||'.'||quote_ident(return_columns[1]) END),
              '(jq.before_change->>'||quote_literal(join_to_column)||')::'||join_column_type,
              /****
              We use the higher level here just to be consistent with aliases from insert/update key retrieval
               */
              'j'||level
              )
      END
        AS delete_key_retrieval_sql,
      source_change_date_select
    FROM queue_deps_with_delete_retrieval
    WHERE level = 1
      AND fact_table_id = p_fact_table_id
    UNION ALL
    SELECT c.queue_table_dep_id,
      c.level,
      delete_key_select_column||CASE
      WHEN c.is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||r.level||'.before_change->>'||quote_literal(return_columns[1])||'::TEXT AS key'
            ELSE ', unnest(array[before_change->>'''||array_to_string(return_columns,',j'||r.level||'.before_change->>''')||'''])::TEXT AS key'
          END
      WHEN join_return_is_fact_key
        THEN
          CASE
            WHEN array_length(return_columns, 1) = 1
            THEN ', j'||c.level||'.'||quote_ident(return_columns_from_join[1])||'::TEXT AS key'
            ELSE ', unnest(array[j'||c.level||'.'||array_to_string(return_columns_from_join,',j'||c.level)||')::TEXT AS key'
          END
        ELSE ''
      END AS delete_key_select_column,
      delete_key_retrieval_sql||CASE
      WHEN is_fact_key
        THEN ''
        ELSE format($$
          --Join to either the base table, or the audit table, one of which
          --will be missing the key in a delete case
            INNER JOIN LATERAL (
             SELECT %s FROM %s jb
             WHERE %s = %s
             UNION ALL
             SELECT %s FROM %s jq
             WHERE operation = 'D'
              AND %s = %s) %s ON TRUE
             $$, quote_ident(return_columns_from_join[1]),
              join_to_relation::TEXT,
              'j'||r.level||'.'||quote_ident(return_columns[1]),
              'jb.'||quote_ident(join_to_column),
              '(before_change->>'||quote_literal(return_columns_from_join[1])||')::'||return_columns_from_join_type,
              join_to_relation_queue::TEXT,
              'j'||r.level||'.'||quote_ident(return_columns[1]),
              '(jq.before_change->>'||quote_literal(join_to_column)||')::'||join_column_type,
              /****
              We use the higher level here just to be consistent with aliases from insert/update key retrieval
               */
              'j'||c.level
              )
      END
        AS delete_key_retrieval_sql,
      r.source_change_date_select
    FROM delete_sql_builder r
    INNER JOIN queue_deps_with_delete_retrieval c USING (queue_table_dep_id)
    WHERE c.level = r.level + 1
  )
  
  , field_vars AS (
    SELECT
      *,
      format($$
        %s AS fact_table_id,
        %s AS queue_table_dep_id,
        %s::INT AS fact_table_dep_id,
        %s::INT AS fact_table_dep_queue_table_dep_id,
        %s AS queue_table_id_field,
        q.fact_loader_batch_id,
        %s::TIMESTAMPTZ AS maximum_cutoff_time,
      -- We must not ignore ids which are above maximum_cutoff_time
      -- but below the highest id which is below maximum_cutoff_time
        MIN(q.fact_loader_batch_id)
        FILTER (
        WHERE %s %s > %s::TIMESTAMPTZ)
        OVER() AS min_missed_id
      $$,
        fact_table_id,
        queue_table_dep_id,
        (CASE WHEN fact_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_id::TEXT END),
        (CASE WHEN fact_table_dep_queue_table_dep_id IS NULL THEN 'NULL'::TEXT ELSE fact_table_dep_queue_table_dep_id::TEXT END),
        'q.'||quote_ident(queue_table_id_field),
        quote_literal(maximum_cutoff_time),
        'q.'||quote_ident(queue_table_timestamp),
        CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END,
        quote_literal(maximum_cutoff_time)
      )
        AS inner_shared_select_columns,
      $$
        fact_table_id,
        queue_table_dep_id,
        fact_table_dep_id,
        fact_table_dep_queue_table_dep_id,
        queue_table_id_field,
        fact_loader_batch_id,
        maximum_cutoff_time,
        min_missed_id
      $$
        AS outer_shared_select_columns,
      CASE WHEN queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(queue_table_tz) END
        AS changed_at_tz_correction
    FROM fact_loader.queue_deps_all c
    WHERE c.fact_table_id = p_fact_table_id
  )
  
  , non_recursive_sql AS (
    SELECT
    /****
    Separate select list for:
      - raw queue_ids from queue tables
      - gathered data from joining queue_ids to source tables to get actual keys to update in fact tables
     */
    -- gathering all queue_ids from queue tables
      queue_table_dep_id,
      outer_shared_select_columns,
      format($$
        %s,
        %s %s AS changed_at,
        %s AS queue_table_id
      $$,
        inner_shared_select_columns,
        'q.'||quote_ident(queue_table_timestamp),
        changed_at_tz_correction,
        queue_table_id
      )
        AS inner_metadata_select_columns,
      format($$
        %s,
        queue_table_id
      $$,
        outer_shared_select_columns
      )
        AS outer_metadata_select_columns,
  
    -- gathering actual keys to update in fact tables by joining from queue_ids to source tables
      format($$
        %s,
        %s AS operation,
        %s %s AS changed_at,
        %s::REGPROC AS insert_merge_proid,
        %s::REGPROC AS update_merge_proid,
        %s::REGPROC AS delete_merge_proid
      $$,
        inner_shared_select_columns,
        'q.'||quote_ident(queue_table_op),
        'q.'||quote_ident(queue_table_timestamp),
        changed_at_tz_correction,
        CASE WHEN insert_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(insert_merge_proid) END,
        CASE WHEN update_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(update_merge_proid) END,
        CASE WHEN delete_merge_proid IS NULL THEN 'NULL' ELSE quote_literal(delete_merge_proid) END
      )
        AS inner_data_select_columns,
      format($$
        %s,
        operation,
        changed_at,
        insert_merge_proid,
        update_merge_proid,
        delete_merge_proid,
        key,
        source_change_date
      $$,
        outer_shared_select_columns
      )
        AS outer_data_select_columns,
  
    -- This is simply the queue table aliased as q
      format('%s q', queue_table_relid::TEXT) AS queue_table_aliased,
  
    -- This is the SQL to join from the queue table to the base table
      format($$
        INNER JOIN %s b
          ON q.%s::%s = b.%s
      $$,
        queue_of_base_table_relid::TEXT,
        quote_ident(queue_table_key),
        queue_of_base_table_key_type,
        quote_ident(queue_of_base_table_key))
             AS base_join_sql,
  
    -- This is a WHERE statement to be added to ALL gathering of new queue_ids to process.
    -- There is a further filter based on the window min_missed_id after this subquery
      format($$ %s
          $$,
          CASE
            WHEN last_cutoff_id IS NOT NULL
            THEN 'q.fact_loader_batch_id > '||last_cutoff_id
            ELSE
              'TRUE'
          END)
             AS inner_global_where_sql,
      format($$
          -- changed_at is guaranteed now to be in timestamptz - any time zone casting is only in subquery
          changed_at < %s
          AND (min_missed_id IS NULL OR (fact_loader_batch_id < min_missed_id))
          $$,
          quote_literal(c.maximum_cutoff_time)
          )
             AS outer_global_where_sql,
      format($$
        AND q.%s = 'I'
        $$,
          queue_table_op)
             AS where_for_insert_sql,
      format($$
        AND (q.%s = 'U' AND %s)
        $$,
          queue_table_op,
          CASE
            WHEN relevant_change_columns IS NULL
              THEN 'TRUE'
            ELSE
              format($$q.%s ?| '{%s}'$$, queue_table_change, array_to_string(relevant_change_columns,','))
          END)
             AS where_for_update_sql,
      format($$
        AND q.%s = 'D'
        $$,
          queue_table_op)
             AS where_for_delete_sql
    FROM field_vars c
  )
  
  , insert_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM insert_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , update_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM update_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , delete_sql_builder_final AS
  (SELECT DISTINCT ON (queue_table_dep_id)
    *
  FROM delete_sql_builder
  ORDER BY queue_table_dep_id, level DESC
  )
  
  , all_queues_sql AS (
  SELECT
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_data_select_columns,
      nrs.inner_data_select_columns||isbf.key_select_column||isbf.source_change_date_select,
      nrs.queue_table_aliased||nrs.base_join_sql,
      isbf.key_retrieval_sql,
      nrs.inner_global_where_sql||nrs.where_for_insert_sql,
      nrs.outer_global_where_sql) AS queue_insert_sql,
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_data_select_columns,
      nrs.inner_data_select_columns||usbf.key_select_column||usbf.source_change_date_select,
      nrs.queue_table_aliased||nrs.base_join_sql,
      usbf.key_retrieval_sql,
      nrs.inner_global_where_sql||nrs.where_for_update_sql,
      nrs.outer_global_where_sql) AS queue_update_sql,
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_data_select_columns,
      nrs.inner_data_select_columns||dsbf.delete_key_select_column||usbf.source_change_date_select,
      nrs.queue_table_aliased,
      dsbf.delete_key_retrieval_sql,
      nrs.inner_global_where_sql||nrs.where_for_delete_sql,
      nrs.outer_global_where_sql) AS queue_delete_sql,
    format($$
    SELECT %s
    FROM (
    SELECT %s
    FROM %s
    WHERE %s ) sub
    WHERE %s
    $$,
      nrs.outer_metadata_select_columns,
      nrs.inner_metadata_select_columns,
      nrs.queue_table_aliased,
      nrs.inner_global_where_sql,
      nrs.outer_global_where_sql) AS queue_ids_sql
  FROM non_recursive_sql nrs
  INNER JOIN insert_sql_builder_final isbf ON isbf.queue_table_dep_id = nrs.queue_table_dep_id
  INNER JOIN update_sql_builder_final usbf ON usbf.queue_table_dep_id = nrs.queue_table_dep_id
  INNER JOIN delete_sql_builder_final dsbf ON dsbf.queue_table_dep_id = nrs.queue_table_dep_id
  )
  
  , final_queue_sql AS
  (SELECT string_agg(
    /****
    This first UNION is to union together INSERT, UPDATE, and DELETE events for a single queue table
     */
    format($$
    %s
    UNION ALL
    %s
    UNION ALL
    %s
    $$,
      queue_insert_sql,
      queue_update_sql,
      queue_delete_sql)
    /****
    This second UNION as the second arg of string_agg is the union together ALL queue tables for this fact table
     */
    , E'\nUNION ALL\n') AS event_sql,
    string_agg(queue_ids_sql, E'\nUNION ALL\n') AS raw_queued_changes_sql_out
  FROM all_queues_sql)
  
  , final_outputs AS (
  SELECT raw_queued_changes_sql_out,
  $$
  WITH all_changes AS (
  ($$||event_sql||$$)
  ORDER BY changed_at)
  
  , base_execution_groups AS
  (SELECT fact_table_id,
      queue_table_dep_id,
      queue_table_id_field,
      operation,
      changed_at,
      source_change_date,
      insert_merge_proid,
      update_merge_proid,
      delete_merge_proid,
      maximum_cutoff_time,
      key,
      CASE WHEN operation = 'I' THEN insert_merge_proid
      WHEN operation = 'U' THEN update_merge_proid
      WHEN operation = 'D' THEN delete_merge_proid
          END AS proid,
    RANK() OVER (
      PARTITION BY
        CASE
          WHEN operation = 'I' THEN insert_merge_proid
          WHEN operation = 'U' THEN update_merge_proid
          WHEN operation = 'D' THEN delete_merge_proid
        END
    ) AS execution_group
    FROM all_changes
    WHERE key IS NOT NULL)
  
  SELECT fact_table_id, proid, key, source_change_date
  FROM base_execution_groups beg
  WHERE proid IS NOT NULL
  GROUP BY execution_group, fact_table_id, proid, key, source_change_date
  /****
  This ordering is particularly important for date-range history tables
  where order of inserts is critical and usually expected to follow a pattern
  ***/
  ORDER BY execution_group, MIN(changed_at), MIN(queue_table_id_field);
  $$ AS gathered_queued_changes_sql_out
    ,
  
  $$
  DROP TABLE IF EXISTS process_queue;
  CREATE TEMP TABLE process_queue
  (process_queue_id serial,
   fact_table_id int,
   proid regproc,
   key_value text,
   source_change_date date);
  
  INSERT INTO process_queue
  (fact_table_id, proid, key_value, source_change_date)
  $$ AS process_queue_snippet,
  
  $$
  WITH all_ids AS
  ($$||raw_queued_changes_sql_out||$$)
  
  , new_metadata AS
  (SELECT MAX(fact_loader_batch_id) AS last_cutoff_id,
    maximum_cutoff_time,
    queue_table_dep_id
  FROM all_ids
  --Exclude dependent fact tables from updates directly to queue_table_deps
  WHERE fact_table_dep_id IS NULL
  GROUP BY queue_table_dep_id, maximum_cutoff_time)
  
  /****
  The dependent fact table uses the same queue_table_id_field as last_cutoff
  We are going to update fact_table_deps metadata instead of queue_table_deps
  ****/
  , new_metadata_fact_dep AS
  (SELECT MAX(fact_loader_batch_id) AS last_cutoff_id,
    maximum_cutoff_time,
    fact_table_dep_queue_table_dep_id
  FROM all_ids
  --Include dependent fact tables only
  WHERE fact_table_dep_id IS NOT NULL
  GROUP BY fact_table_dep_queue_table_dep_id, maximum_cutoff_time)
  
  , update_key AS (
  SELECT qdwr.queue_table_dep_id,
    --Cutoff the id to that newly found, otherwise default to last value
    COALESCE(mu.last_cutoff_id, qdwr.last_cutoff_id) AS last_cutoff_id,
    --This cutoff time must always be the same for all queue tables for given fact table.
    --Even if there are no new records, we move this forward to wherever the stream is at
    qdwr.maximum_cutoff_time AS last_cutoff_source_time
  FROM fact_loader.queue_deps_all qdwr
  LEFT JOIN new_metadata mu ON mu.queue_table_dep_id = qdwr.queue_table_dep_id
  WHERE qdwr.fact_table_id = $$||p_fact_table_id||$$
  --Exclude dependent fact tables from updates directly to queue_table_deps
    AND qdwr.fact_table_dep_id IS NULL
  )
  
  /****
  This SQL also nearly matches that for the queue_table_deps but would be a little ugly to try to DRY up
  ****/
  , update_key_fact_dep AS (
  SELECT qdwr.fact_table_dep_queue_table_dep_id,
    qdwr.fact_table_id,
    COALESCE(mu.last_cutoff_id, qdwr.last_cutoff_id) AS last_cutoff_id,
    qdwr.maximum_cutoff_time AS last_cutoff_source_time
  FROM fact_loader.queue_deps_all qdwr
  LEFT JOIN new_metadata_fact_dep mu ON mu.fact_table_dep_queue_table_dep_id = qdwr.fact_table_dep_queue_table_dep_id
  WHERE qdwr.fact_table_id = $$||p_fact_table_id||$$
  --Include dependent fact tables only
    AND qdwr.fact_table_dep_id IS NOT NULL
  )
  
  , updated_queue_table_deps AS (
  UPDATE fact_loader.queue_table_deps qtd
  SET last_cutoff_id = uk.last_cutoff_id,
    last_cutoff_source_time = uk.last_cutoff_source_time
  FROM update_key uk
  WHERE qtd.queue_table_dep_id = uk.queue_table_dep_id
  RETURNING qtd.*)
  
  , updated_fact_table_deps AS (
  UPDATE fact_loader.fact_table_dep_queue_table_deps ftd
  SET last_cutoff_id = uk.last_cutoff_id,
    last_cutoff_source_time = uk.last_cutoff_source_time
  FROM update_key_fact_dep uk
  WHERE ftd.fact_table_dep_queue_table_dep_id = uk.fact_table_dep_queue_table_dep_id
  RETURNING uk.*)
  
  UPDATE fact_loader.fact_tables ft
  SET last_refresh_source_cutoff = uqtd.last_cutoff_source_time,
    last_refresh_attempted_at = now(),
    last_refresh_succeeded = TRUE
  FROM
  (SELECT fact_table_id, last_cutoff_source_time
  FROM updated_queue_table_deps
  --Must use UNION to get only distinct values
  UNION
  SELECT fact_table_id, last_cutoff_source_time
  FROM updated_fact_table_deps) uqtd
  WHERE uqtd.fact_table_id = ft.fact_table_id;
  $$ AS metadata_update_sql_out
  FROM final_queue_sql)
  
  SELECT raw_queued_changes_sql_out,
   gathered_queued_changes_sql_out
    ,
   format($$
    %s
    %s$$, process_queue_snippet, gathered_queued_changes_sql_out) AS process_queue_sql_out,
   metadata_update_sql_out
  FROM final_outputs;
  
  $BODY$
  LANGUAGE SQL;
  
  
  /* pg_fact_loader--1.5--1.6.sql */
  
  -- complain if script is sourced in psql, rather than via CREATE EXTENSION
  \echo Use "CREATE EXTENSION pg_fact_loader" to load this file. \quit
  
  DROP VIEW fact_loader.queue_deps_all_with_retrieval;
  DROP VIEW fact_loader.queue_deps_all;
  DROP VIEW fact_loader.prioritized_jobs;
  
  -- Must ensure we have the fully schema-qualified regprod before converting to text
  SET search_path TO '';
  ALTER TABLE fact_loader.debug_process_queue ALTER COLUMN proid TYPE TEXT;
  ALTER TABLE fact_loader.debug_process_queue ADD CONSTRAINT check_proid CHECK (COALESCE(proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps ALTER COLUMN delete_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps ADD CONSTRAINT check_delete_merge_proid CHECK (COALESCE(delete_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps ALTER COLUMN insert_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps ADD CONSTRAINT check_insert_merge_proid CHECK (COALESCE(insert_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps ALTER COLUMN update_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_table_dep_queue_table_deps ADD CONSTRAINT check_update_merge_proid CHECK (COALESCE(update_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_table_deps ALTER COLUMN default_delete_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_table_deps ADD CONSTRAINT check_default_delete_merge_proid CHECK (COALESCE(default_delete_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_table_deps ALTER COLUMN default_insert_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_table_deps ADD CONSTRAINT check_default_insert_merge_proid CHECK (COALESCE(default_insert_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_table_deps ALTER COLUMN default_update_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_table_deps ADD CONSTRAINT check_default_update_merge_proid CHECK (COALESCE(default_update_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_tables ALTER COLUMN daily_scheduled_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_tables ADD CONSTRAINT check_daily_scheduled_proid CHECK (COALESCE(daily_scheduled_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.fact_tables ALTER COLUMN fact_table_agg_proid TYPE TEXT;
  ALTER TABLE fact_loader.fact_tables ADD CONSTRAINT check_fact_table_agg_proid CHECK (COALESCE(fact_table_agg_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.queue_table_deps ALTER COLUMN delete_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.queue_table_deps ADD CONSTRAINT check_delete_merge_proid CHECK (COALESCE(delete_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.queue_table_deps ALTER COLUMN insert_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.queue_table_deps ADD CONSTRAINT check_insert_merge_proid CHECK (COALESCE(insert_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  ALTER TABLE fact_loader.queue_table_deps ALTER COLUMN update_merge_proid TYPE TEXT;
  ALTER TABLE fact_loader.queue_table_deps ADD CONSTRAINT check_update_merge_proid CHECK (COALESCE(update_merge_proid::REGPROC, 'boolin') IS NOT NULL);
  RESET search_path;
  
  
  CREATE OR REPLACE VIEW fact_loader.prioritized_jobs AS
  WITH jobs_with_daily_variables AS (
  SELECT
     ft.*,
  /***
  Keep all this logic of daily jobs as variables to ease visualization of logic in the next cte below!!
   */
      (--If this is the first run of a scheduled job, it is eligible
              ft.last_refresh_attempted_at IS NULL
            OR (
                 --If it was last attempted successfully prior to this scheduled time only - meaning yesterday, it is eligible
                  (
                   ft.last_refresh_succeeded AND
                   ft.last_refresh_attempted_at::DATE <
                    -- Timezone taken from daily_scheduled_tz if base job, otherwise look up the timezone of the base job if this is dependent
                      (now() AT TIME ZONE COALESCE(
                                            ft.daily_scheduled_tz,
                                            base.daily_scheduled_tz
                                            )
                      )::DATE
                  )
                OR
                 --If a job has failed and been re-enabled, it is eligible again even though it has been attempted at or after the scheduled time
                   NOT ft.last_refresh_succeeded
               )
       ) AS daily_not_attempted_today,
  
    (now() AT TIME ZONE ft.daily_scheduled_tz)::TIME
        BETWEEN daily_scheduled_time AND '23:59:59.999999'::TIME AS daily_scheduled_time_passed,
  
    base.use_daily_schedule
    AND base.last_refresh_succeeded
    AND base.last_refresh_attempted_at :: DATE = (now() AT TIME ZONE base.daily_scheduled_tz) :: DATE
      AS daily_base_job_finished,
  
    ft.depends_on_base_daily_job_id = ft.depends_on_parent_daily_job_id AS daily_has_only_one_parent,
  
    -- This should only be used in combination with daily_has_only_one_parent
    parent.use_daily_schedule
    AND parent.last_refresh_succeeded
    AND parent.last_refresh_attempted_at :: DATE = (now() AT TIME ZONE COALESCE(parent.daily_scheduled_tz, base.daily_scheduled_tz)) :: DATE
      AS parent_job_finished
  FROM fact_loader.fact_tables ft
  LEFT JOIN LATERAL
    (SELECT ftb.use_daily_schedule,
      ftb.last_refresh_succeeded,
      ftb.last_refresh_attempted_at,
      ftb.daily_scheduled_tz
    FROM fact_loader.fact_tables ftb
    WHERE ftb.fact_table_id = ft.depends_on_base_daily_job_id) base ON TRUE
  LEFT JOIN LATERAL
    (SELECT ftp.use_daily_schedule,
      ftp.last_refresh_succeeded,
      ftp.last_refresh_attempted_at,
      ftp.daily_scheduled_tz
    FROM fact_loader.fact_tables ftp
    WHERE ftp.fact_table_id = ft.depends_on_parent_daily_job_id) parent ON TRUE
  WHERE enabled
  )
  
  , jobs_with_daily_schedule_eligibility AS (
  SELECT
     *,
     --Only run this job according to the same day of the daily_scheduled_time
     --according to configured timezone
     (use_daily_schedule AND daily_not_attempted_today
       AND
      (
        daily_scheduled_time_passed
        OR
       (daily_base_job_finished AND (daily_has_only_one_parent OR parent_job_finished))
      )
     ) AS daily_schedule_eligible
  FROM jobs_with_daily_variables)
  
  SELECT *
  FROM jobs_with_daily_schedule_eligibility
  WHERE NOT use_daily_schedule OR daily_schedule_eligible
  ORDER BY
    CASE WHEN force_worker_priority THEN 0 ELSE 1 END,
    --If a job has a daily schedule, once the time has come for the next refresh,
    --prioritize it first
    CASE
      WHEN daily_schedule_eligible
      THEN (now() AT TIME ZONE daily_scheduled_tz)::TIME
      ELSE NULL
    END NULLS LAST,
    --This may be improved in the future but is a good start
    last_refresh_attempted_at NULLS FIRST,
    priority
  ;
  
  
  CREATE OR REPLACE VIEW fact_loader.queue_deps_all AS
  WITH RECURSIVE fact_table_dep_cutoffs AS
  (SELECT
      1 AS level
      , qtd.queue_table_dep_id
      , ftdqc.fact_table_dep_id
      , ftdqc.fact_table_dep_queue_table_dep_id
      --This dep_maximum_cutoff_time is being taken from the queue_table_deps, because we cannot go past when the
      --fact table has been updated
      , qtd.last_cutoff_id                        AS dep_maximum_cutoff_id
      , qtd.last_cutoff_source_time               AS dep_maximum_cutoff_time
      , ftd.parent_id                                 AS parent_fact_table_id
      , ftd.child_id                                  AS child_fact_table_id
      , ftd.child_id                                  AS base_fact_table_id
      , queue_table_id
      , relevant_change_columns
      , ftdqc.last_cutoff_id
      , ftdqc.last_cutoff_source_time
      , ftdqc.insert_merge_proid
      , ftdqc.update_merge_proid
      , ftdqc.delete_merge_proid
    FROM fact_loader.queue_table_deps qtd
    INNER JOIN fact_loader.fact_table_dep_queue_table_deps ftdqc ON ftdqc.queue_table_dep_id = qtd.queue_table_dep_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.fact_table_dep_id = ftdqc.fact_table_dep_id
    UNION ALL
    /****
    In this recursive part, we walk UP the chain to the base level in order to get the
    last_cutoff_id and last_cutoff_source_time of parent_ids because children must never surpass those.
  
    The ONLY difference between this recursive part and the non-recursive part are the dep_maximum_cutoffs.
    That means we can get our resultant data below by simply selecting distinct ON the right fields and order
    by dep_maximum_cutoffs to get the most conservative cutoff window, that is, the minimum cutoff amongst
    the queue tables and any PARENT fact table cutoffs.
  
    That means if, for example,
      - IF a queue table has been cutoff up until 11:00:00
      - AND IF a level 1 fact table dependent on that queue table was last cutoff at 10:55:00
      - THEN a level 2 fact table dependent on level 1 fact table must not go past 10:55:00 when it is processed.
     */
    SELECT
      ftdc.level + 1 AS level
      , ftdc.queue_table_dep_id
      , ftdc.fact_table_dep_id
      , ftdc.fact_table_dep_queue_table_dep_id
      --This dep_maximum_cutoff_time is being taken from the queue_table_deps, because we cannot go past when the
      --fact table has been updated
      , ftdqc.last_cutoff_id                        AS dep_maximum_cutoff_id
      , ftdqc.last_cutoff_source_time               AS dep_maximum_cutoff_time
      , ftd.parent_id                                 AS parent_fact_table_id
      , ftd.child_id                                  AS child_fact_table_id
      , ftdc.base_fact_table_id
      , ftdc.queue_table_id
      , ftdc.relevant_change_columns
      , ftdc.last_cutoff_id
      , ftdc.last_cutoff_source_time
      , ftdc.insert_merge_proid
      , ftdc.update_merge_proid
      , ftdc.delete_merge_proid
    FROM fact_loader.queue_table_deps qtd
    INNER JOIN fact_loader.fact_table_dep_queue_table_deps ftdqc ON ftdqc.queue_table_dep_id = qtd.queue_table_dep_id
    INNER JOIN fact_loader.fact_table_deps ftd ON ftd.fact_table_dep_id = ftdqc.fact_table_dep_id
    INNER JOIN fact_table_dep_cutoffs ftdc ON ftdc.parent_fact_table_id = ftd.child_id
  )
  
  , adjusted_fact_table_deps AS (
  /****
  The reason we look at distinct queue_table_dep_id and not simply queue_table_id
  is because two parent fact tables could have differing logic for retrieving changes
  for the same base queue_tables.
   */
  SELECT DISTINCT ON(base_fact_table_id, queue_table_dep_id)
  *
  FROM fact_table_dep_cutoffs
  ORDER BY base_fact_table_id, queue_table_dep_id, dep_maximum_cutoff_time
  )
  
  , queue_table_info AS (
      SELECT * FROM fact_loader.queue_table_delay_info()
  )
  
  /****
  For fact tables that depend on other fact tables, we join the child fact table to the queue_table_deps of the parent
  fact table, and just reuse this exactly, with these distinctions:
    - From the fact_table_dep table, we do use the proids, and the last_cutoff_id
    - We use the parent last_cutoff_source_time as the maximum_cutoff, because we can only update those records already updated on the parent
    - We pass the information of which table for which to update metadata in the end
   */
  , queue_table_deps_with_nested AS (
    /****
    This part of the union is for the base level of queue_table_deps - for fact tables with no other dependent fact tables
     */
    SELECT
      queue_table_dep_id
      , NULL :: INT                                AS fact_table_dep_id
      , NULL :: INT                                AS fact_table_dep_queue_table_dep_id
      , NULL :: BIGINT                             AS dep_maximum_cutoff_id
      , NULL :: TIMESTAMPTZ                        AS dep_maximum_cutoff_time
      , fact_table_id
      , queue_table_id
      , relevant_change_columns
      , last_cutoff_id
      , last_cutoff_source_time
      , insert_merge_proid
      , update_merge_proid
      , delete_merge_proid
    FROM fact_loader.queue_table_deps
    UNION ALL
    /****
    This part of the union is for fact tables with other dependent fact tables
     */
    SELECT
      queue_table_dep_id
      , fact_table_dep_id
      , fact_table_dep_queue_table_dep_id
      , aftd.dep_maximum_cutoff_id
      , aftd.dep_maximum_cutoff_time
      , base_fact_table_id                                  AS fact_table_id
      , queue_table_id
      , relevant_change_columns
      , aftd.last_cutoff_id
      , aftd.last_cutoff_source_time
      , aftd.insert_merge_proid
      , aftd.update_merge_proid
      , aftd.delete_merge_proid
    FROM adjusted_fact_table_deps aftd
  )
  
  SELECT
    ft.fact_table_id,
    ft.fact_table_relid,
    ft.fact_table_agg_proid,
    qt.queue_table_id,
    qt.queue_table_relid,
    qt.queue_of_base_table_relid,
    qtd.relevant_change_columns,
    qtd.last_cutoff_id,
    qtd.last_cutoff_source_time,
    rt.if_name AS provider_name,
    rt.replication_set_name,
    qtd.dep_maximum_cutoff_id,  --Not used yet - TODO - think about if it needs to be used to filter as cutoff MAX in addition to the time filter
    LEAST(
        MIN(qtd.dep_maximum_cutoff_time)
        OVER (
          PARTITION BY qtd.fact_table_id ),
        MIN(rt.source_time)
        OVER (
          PARTITION BY qtd.fact_table_id )
    ) AS maximum_cutoff_time,
    aqt.queue_table_id_field,
    'primary_key'::name AS queue_table_key,
    'operation'::name AS queue_table_op,
    'change'::name AS queue_table_change,
    'changed_at'::name AS queue_table_timestamp,
    qt.queue_table_tz,
    aqbt.queue_of_base_table_key,
    aqbt.queue_of_base_table_key_type,
    queue_table_dep_id,
    fact_table_dep_id,
    fact_table_dep_queue_table_dep_id,
    insert_merge_proid,
    update_merge_proid,
    delete_merge_proid,
    qt.purge
  FROM queue_table_deps_with_nested qtd
  INNER JOIN fact_loader.fact_tables ft ON ft.fact_table_id = qtd.fact_table_id
  INNER JOIN fact_loader.queue_tables qt ON qt.queue_table_id = qtd.queue_table_id
  INNER JOIN queue_table_info rt ON rt.queue_of_base_table_relid = qt.queue_of_base_table_relid
  INNER JOIN LATERAL
    (SELECT a.attname AS queue_of_base_table_key, format_type(atttypid, atttypmod) AS queue_of_base_table_key_type
    FROM (SELECT
            i.indrelid
            , unnest(indkey) AS ik
            , row_number()
              OVER ()        AS rn
          FROM pg_index i
          WHERE i.indrelid = qt.queue_of_base_table_relid AND i.indisprimary) pk
    INNER JOIN pg_attribute a
      ON a.attrelid = pk.indrelid AND a.attnum = pk.ik) aqbt ON TRUE
  INNER JOIN LATERAL
    (SELECT a.attname AS queue_table_id_field
    FROM (SELECT
            i.indrelid
            , unnest(indkey) AS ik
            , row_number()
              OVER ()        AS rn
          FROM pg_index i
          WHERE i.indrelid = qt.queue_table_relid AND i.indisprimary) pk
    INNER JOIN pg_attribute a
      ON a.attrelid = pk.indrelid AND a.attnum = pk.ik) aqt ON TRUE
  ORDER BY ft.fact_table_relid;
  
  
  CREATE OR REPLACE VIEW fact_loader.queue_deps_all_with_retrieval AS
  SELECT
    qtd.*,
    krs.filter_scope,
    krs.level,
    krs.return_columns, --we need not get the type separately.  It must match queue_of_base_table_key_type
    krs.is_fact_key,
    krs.join_to_relation,
    qtk.queue_table_relid AS join_to_relation_queue,
    krs.join_to_column,
    ctypes.join_column_type,
    krs.return_columns_from_join,
    ctypes.return_columns_from_join_type,
    krs.join_return_is_fact_key,
    /***
    We include this in this view def to be easily shared by all events (I, U, D) in sql_builder,
    as those may be different in terms of passing source_change_date.
     */
    format(', %s::DATE AS source_change_date',
      CASE
        WHEN krs.pass_queue_table_change_date_at_tz IS NOT NULL
          /***
          For casting queue_table_timestamp to a date, we first ensure we have it as timestamptz (objective UTC time).
          Then, we cast it to the timezone of interest on which the date should be based.
          For example, 02:00:00 UTC time on 2018-05-02 is actually 2018-05-01 in America/Chicago time.
          Thus, any date-based fact table must decide in what time zone to consider the date.
           */
          THEN format('(%s %s AT TIME ZONE %s)',
                      'q.'||quote_ident(qtd.queue_table_timestamp),
                      CASE WHEN qtd.queue_table_tz IS NULL THEN '' ELSE 'AT TIME ZONE '||quote_literal(qtd.queue_table_tz) END,
                      quote_literal(krs.pass_queue_table_change_date_at_tz))
          ELSE 'NULL'
        END) AS source_change_date_select
  FROM fact_loader.queue_deps_all qtd
  INNER JOIN fact_loader.key_retrieval_sequences krs ON qtd.queue_table_dep_id = krs.queue_table_dep_id
  LEFT JOIN fact_loader.queue_tables qtk ON qtk.queue_of_base_table_relid = krs.join_to_relation
  LEFT JOIN LATERAL
    (SELECT MAX(CASE WHEN attname = krs.join_to_column THEN format_type(atttypid, atttypmod) ELSE NULL END) AS join_column_type,
      MAX(CASE WHEN attname = krs.return_columns_from_join[1] THEN format_type(atttypid, atttypmod) ELSE NULL END) AS return_columns_from_join_type
    FROM pg_attribute a
    WHERE a.attrelid IN(krs.join_to_relation)
      /****
      We stubbornly assume that if there are multiple columns in return_columns_from_join, they all have the same type.
      Undue complexity would ensue if we did away with that rule.
       */
      AND a.attname IN(krs.join_to_column,krs.return_columns_from_join[1])) ctypes ON TRUE;
  
  
  