1
\$\begingroup\$

Here is a stored procedure which we use to port the data from a remote server to our data warehouse server. The SP uses a generic style where all the tables along with its schema and data will be loaded into the data warehouse.

The script is working fine, but now we have a requirement where few of the tables like customer_access (which I have filtered in the query - couple more tables are there) need to be refreshed within a short time (1-2 minutes).The data count of those tables is around 500K to 1000K, hence the porting takes around 7-8 minutes.

The logic used in the script is drop and recreate each and every time the table will be dropped and recreated.

To make it faster I thought of using the MERGE or UPSERT logic and I need your suggestions on implementing this in a most generic way which will work for all tables.

Scored Procedure Code:

CREATE OR REPLACE PROCEDURE public.load_foreign_schema_postgresql_stock(
    _server text,
    _remote_schema text,
    _dl_schema text DEFAULT NULL::text,
    _foreign_table_schema text DEFAULT NULL::text,
    check_ft boolean DEFAULT true,
    recreate_primary_keys boolean DEFAULT true,
    log_table_schema_name text DEFAULT NULL::text,
    log_table_name text DEFAULT NULL::text)
LANGUAGE 'plpgsql'

AS $BODY$
DECLARE
    tab                     RECORD;
    tab_indices             RECORD;

    start_time              TIMESTAMPTZ;

    pg_class                TEXT := 'pg_class_stock' || now()::TEXT;
    pg_namespace            TEXT := 'pg_namespace_stock' || now()::TEXT;
    pg_index                TEXT := 'pg_index_stock' || now()::TEXT;
    pg_attribute            TEXT := 'pg_attribute_stock' || now()::TEXT;

    foreign_server          TEXT;
    remote_schema           TEXT;
    foreign_schema          TEXT;
    dl_schema               TEXT;
    table_name              TEXT;
    nr_rows                 TEXT;
    duration_table_load     INTERVAL;
    duration_index_creation INTERVAL;
    duration_analyze        INTERVAL;
    recreate_primary_keys       BOOLEAN;

    index_query             TEXT;

BEGIN
    IF _foreign_table_schema ISNULL
    THEN
        _foreign_table_schema := 'ft_' || _remote_schema;
    END IF;

    IF _dl_schema ISNULL
    THEN
        _dl_schema := _remote_schema;
    END IF;

    foreign_server := _server;
    remote_schema := _remote_schema;
    dl_schema := _dl_schema;
    foreign_schema := _foreign_table_schema;
    recreate_primary_keys := load_foreign_schema_postgresql_stock.recreate_primary_keys;

    IF recreate_primary_keys
    THEN
        EXECUTE format('CREATE FOREIGN TABLE %1$I.%2$I
                    (
                    oid OID OPTIONS (COLUMN_NAME ''oid''),
                    relname NAME OPTIONS (COLUMN_NAME ''relname''),
                    relnamespace OID OPTIONS (COLUMN_NAME ''relnamespace''),
                    relkind CHAR OPTIONS (COLUMN_NAME ''relkind'')
                    )
                    SERVER %3$I
                    OPTIONS (SCHEMA_NAME ''pg_catalog'', TABLE_NAME ''pg_class'')
                  ;', _foreign_table_schema, pg_class, _server);
        EXECUTE format('CREATE FOREIGN TABLE %1$I.%2$I
                    (
                    oid OID OPTIONS (COLUMN_NAME ''oid''),
                    nspname TEXT OPTIONS (COLUMN_NAME ''nspname'')
                    )
                    SERVER %3$I
                    OPTIONS (SCHEMA_NAME ''pg_catalog'', TABLE_NAME ''pg_namespace'')
                  ;', _foreign_table_schema, pg_namespace, _server);
        EXECUTE format('CREATE FOREIGN TABLE %1$I.%2$I
                    (
                    oid OID OPTIONS (COLUMN_NAME ''oid''),
                    indexrelid OID OPTIONS (COLUMN_NAME ''indexrelid''),
                    indrelid OID OPTIONS (COLUMN_NAME ''indrelid''),
                    indisunique BOOLEAN OPTIONS (COLUMN_NAME ''indisunique''),
                    indisprimary BOOLEAN OPTIONS (COLUMN_NAME ''indisprimary''),
                    indkey INT2VECTOR OPTIONS (COLUMN_NAME ''indkey'')
                    )
                    SERVER %3$I
                    OPTIONS (SCHEMA_NAME ''pg_catalog'', TABLE_NAME ''pg_index'')
                  ;', _foreign_table_schema, pg_index, _server);
        EXECUTE format('CREATE FOREIGN TABLE %1$I.%2$I
                    (
                    attrelid OID OPTIONS (COLUMN_NAME ''attrelid''),
                    attname NAME OPTIONS (COLUMN_NAME ''attname''),
                    attnum SMALLINT OPTIONS (COLUMN_NAME ''attnum'')
                    )
                    SERVER %3$I
                    OPTIONS (SCHEMA_NAME ''pg_catalog'', TABLE_NAME ''pg_attribute'')
                  ;', _foreign_table_schema, pg_attribute, _server);
        COMMIT;
    END IF;

    FOR tab IN
        (
            SELECT foreign_table_schema,
                   foreign_table_name
            FROM information_schema.foreign_tables
            WHERE foreign_table_schema = _foreign_table_schema
              AND foreign_server_name = _server
              AND foreign_table_name NOT IN (pg_class, pg_attribute, pg_namespace, pg_index)
              AND foreign_table_name='customer_access'
            ORDER BY 1, 2
        )
        LOOP
            table_name := tab.foreign_table_name;

            RAISE NOTICE 'Loading table %.% into %.%', tab.foreign_table_schema, tab.foreign_table_name, _dl_schema, tab.foreign_table_name;
            EXECUTE format('DROP TABLE IF EXISTS %I.%I;', _dl_schema, tab.foreign_table_name);
            start_time := clock_timestamp();
            EXECUTE format('CREATE TABLE %1$I.%2$I AS
                    SELECT *
                    FROM %3$I.%2$I;', _dl_schema, tab.foreign_table_name, tab.foreign_table_schema);
            duration_table_load := clock_timestamp() - start_time;

            GET DIAGNOSTICS nr_rows := ROW_COUNT;
            RAISE NOTICE '% rows loaded into %.%', nr_rows, _dl_schema, tab.foreign_table_name;

            duration_index_creation := NULL;

            IF recreate_primary_keys
            THEN
                start_time := clock_timestamp();
                index_query := format('
                    WITH indexes AS (
                        SELECT n.nspname AS schemaname,
                               c.relname AS tablename,
                               i.relname AS indexname,
                               x.indisprimary,
                               indkey,
                               indrelid
                        FROM (((%1$I.%5$I x
                            JOIN %1$I.%2$I c ON ((c.oid = x.indrelid)))
                            JOIN %1$I.%2$I i ON ((i.oid = x.indexrelid)))
                             LEFT JOIN %1$I.%4$I n ON ((n.oid = c.relnamespace)))
                        WHERE ((c.relkind = ANY (ARRAY [''r''::"char", ''m''::"char"])) AND (i.relkind = ''i''::"char"))
                        AND n.nspname = %6$L
                                AND c.relname = %7$L
                    ),
                         indexes_cols_pkey AS (
                             SELECT *,
                                    indkey[a] AS ikey
                             FROM indexes,
                                  generate_subscripts(indkey, 1) t(a)
                             WHERE indisprimary
                         ), pkeys AS (
                        SELECT format(''ALTER TABLE %8$I.%%1$I ADD PRIMARY KEY (%%2$s)'', i.tablename,
                                      string_agg(format(''%%1$I'', attname::TEXT), '', '' ORDER BY a)) AS indexdef
                        FROM indexes_cols_pkey i
                             JOIN %1$I.%3$I a ON i.indrelid = a.attrelid AND a.attnum = ikey
                             GROUP BY i.schemaname,i.tablename
                    )
                    SELECT *
                    FROM pkeys
                    ;', tab.foreign_table_schema, pg_class, pg_attribute, pg_namespace, pg_index,
                                      remote_schema, table_name, dl_schema
                    );

                FOR tab_indices IN EXECUTE index_query
                    LOOP
                        RAISE NOTICE '%',tab_indices.indexdef;
                        EXECUTE tab_indices.indexdef;
                    END LOOP;
                duration_index_creation := clock_timestamp() - start_time;
            END IF;

            RAISE NOTICE 'Analyzing %.%', _dl_schema, tab.foreign_table_name;
            start_time := clock_timestamp();
            EXECUTE format('ANALYZE %I.%I', _dl_schema, tab.foreign_table_name);
            duration_analyze := clock_timestamp() - start_time;
            RAISE NOTICE 'Done analyzing %.%', _dl_schema, tab.foreign_table_name;

            COMMIT;

        END LOOP;
END
$BODY$;
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.