~~~BEGIN_MAIN_SECTION~~~
[PRODUCT]
procedure=ETL_MERGE_TABLE
~~~END_MAIN_SECTION~~~


~~~BEGIN_VERSION_SECTION~~~
[VERSION]
version=2026-01-07

[SQL_SCRIPT___DEFAULT]


CREATE OR REPLACE PROCEDURE etl_merge_table (
    p_table_name        IN VARCHAR2,
    p_schema            IN VARCHAR2 DEFAULT NULL,
    p_dry_run           IN BOOLEAN  DEFAULT FALSE,
    p_create_table_only IN BOOLEAN  DEFAULT FALSE,
    p_parallel          IN NUMBER   DEFAULT 4
)
AUTHID CURRENT_USER
IS
    v_schema        VARCHAR2(128) :=
        COALESCE(UPPER(p_schema),
                 SYS_CONTEXT('USERENV','CURRENT_SCHEMA'));

    v_target_table  VARCHAR2(256) :=
        v_schema || '.' || UPPER(p_table_name);

    v_source_name   VARCHAR2(128) :=
        '#' || UPPER(p_table_name);

    v_source_table  VARCHAR2(256) :=
        v_schema || '."' || v_source_name || '"';

    v_exists        NUMBER;

    v_pk_condition  CLOB := '';
    v_pk_columns    CLOB := ',';

    v_update_list   CLOB := '';
    v_insert_cols   CLOB := '';
    v_insert_vals   CLOB := '';

    v_create_sql    CLOB;
    v_merge_sql     CLOB;
    v_count_update  CLOB;
    v_count_insert  CLOB;

    l_update_cnt    NUMBER := 0;
    l_insert_cnt    NUMBER := 0;

BEGIN
    IF p_create_table_only THEN
        SELECT COUNT(*)
        INTO v_exists
        FROM all_tables
        WHERE table_name = v_source_name
          AND owner = v_schema;

        IF v_exists = 1 THEN
            DBMS_OUTPUT.PUT_LINE('Delete existing table ' || v_source_table);
            EXECUTE IMMEDIATE 'DROP TABLE ' || v_source_table;
        END IF;

        v_create_sql :=
            'CREATE TABLE ' || v_source_table ||
            ' AS SELECT * FROM ' || v_target_table || ' WHERE 1 = 0';

        DBMS_OUTPUT.PUT_LINE('Creating empty table ' || v_source_table);
        EXECUTE IMMEDIATE v_create_sql;

        DBMS_OUTPUT.PUT_LINE('create_table_only = TRUE, EXITING.');
        RETURN;
    END IF;

    SELECT COUNT(*)
    INTO v_exists
    FROM all_tables
    WHERE table_name = v_source_name
      AND owner = v_schema;

    IF v_exists = 0 THEN
        v_create_sql :=
            'CREATE TABLE ' || v_source_table ||
            ' AS SELECT * FROM ' || v_target_table || ' WHERE 1 = 0';

        DBMS_OUTPUT.PUT_LINE('Creating empty table ' || v_source_table);
        EXECUTE IMMEDIATE v_create_sql;
    END IF;


    /*SELECT COUNT(*) INTO v_exists FROM all_tables WHERE table_name = 'SYS_MERGE_LOG_TB' AND owner = v_schema;

    IF v_exists = 0 THEN
        v_create_sql := 'CREATE TABLE ' || v_schema || '."' || 'SYS_MERGE_LOG_TB' || '"' || '(log_id NUMBER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, log_date TIMESTAMP DEFAULT SYSTIMESTAMP, target_table VARCHAR2(128), user_name VARCHAR2(30), rows_updated NUMBER, rows_inserted NUMBER, misc VARCHAR2(256))';
        DBMS_OUTPUT.PUT_LINE('Creating empty LOG table ' || v_schema || '."' || 'SYS_MERGE_LOG_TB' || '"');
        EXECUTE IMMEDIATE v_create_sql;
    END IF;*/



    FOR pk IN (
        SELECT acc.column_name
        FROM all_constraints ac
        JOIN all_cons_columns acc
          ON ac.constraint_name = acc.constraint_name
         AND ac.owner = acc.owner
        WHERE ac.constraint_type = 'P'
          AND ac.table_name = UPPER(p_table_name)
          AND ac.owner = v_schema
        ORDER BY acc.position
    ) LOOP
        v_pk_condition :=
            v_pk_condition || 't.' || pk.column_name ||
            ' = s.' || pk.column_name || ' AND ';

        v_pk_columns := v_pk_columns || pk.column_name || ',';
    END LOOP;

    IF v_pk_condition IS NULL THEN
        RAISE_APPLICATION_ERROR(
            -20001,
            'Table ' || v_target_table || ' missing primary key'
        );
    END IF;

    v_pk_condition := RTRIM(v_pk_condition, ' AND ');

    FOR c IN (
        SELECT column_name
        FROM all_tab_columns
        WHERE table_name = UPPER(p_table_name)
          AND owner = v_schema
        ORDER BY column_id
    ) LOOP
        -- UPDATE list
        IF INSTR(v_pk_columns, ',' || c.column_name || ',') = 0 THEN
            v_update_list :=
                v_update_list || 't.' || c.column_name ||
                ' = s.' || c.column_name || ', ';
        END IF;

        -- INSERT list
        v_insert_cols := v_insert_cols || c.column_name || ', ';
        v_insert_vals := v_insert_vals || 's.' || c.column_name || ', ';
    END LOOP;

    v_update_list := RTRIM(v_update_list, ', ');
    v_insert_cols := RTRIM(v_insert_cols, ', ');
    v_insert_vals := RTRIM(v_insert_vals, ', ');

    /* Count UPDATE / INSERT */
    v_count_update :=
        'SELECT COUNT(*) FROM ' || v_target_table || ' t ' ||
        'JOIN ' || v_source_table || ' s ON (' || v_pk_condition || ')';

    v_count_insert :=
        'SELECT COUNT(*) FROM ' || v_source_table || ' s ' ||
        'WHERE NOT EXISTS (SELECT 1 FROM ' || v_target_table ||
        ' t WHERE ' || v_pk_condition || ')';

    EXECUTE IMMEDIATE v_count_update INTO l_update_cnt;
    EXECUTE IMMEDIATE v_count_insert INTO l_insert_cnt;

    /* MERGE – PARALLEL */
    v_merge_sql :=
        'MERGE /*+ PARALLEL(t,' || p_parallel ||
        ') PARALLEL(s,' || p_parallel || ') */ INTO ' ||
        v_target_table || ' t ' ||
        'USING ' || v_source_table || ' s ' ||
        'ON (' || v_pk_condition || ') ' ||
        'WHEN MATCHED THEN UPDATE SET ' || v_update_list || ' ' ||
        'WHEN NOT MATCHED THEN INSERT (' || v_insert_cols || ') ' ||
        'VALUES (' || v_insert_vals || ')';

    DBMS_OUTPUT.PUT_LINE('UPDATE rows: ' || l_update_cnt);
    DBMS_OUTPUT.PUT_LINE('INSERT rows: ' || l_insert_cnt);

    SAVEPOINT before_merge;

    IF p_dry_run THEN
        DBMS_OUTPUT.PUT_LINE('--- DRY RUN MERGE ---');
        DBMS_OUTPUT.PUT_LINE(v_merge_sql);
        ROLLBACK TO before_merge;
    ELSE
        EXECUTE IMMEDIATE v_merge_sql;
		COMMIT;
        EXECUTE IMMEDIATE 'DROP TABLE ' || v_source_table;
        COMMIT;
        DBMS_OUTPUT.PUT_LINE('MERGE + COMMIT OK, TABLE ' || v_source_table || ' deleted.');
        INSERT INTO SYS_MERGE_LOG_TB(target_table,user_name,rows_updated,rows_inserted) VALUES(UPPER(p_table_name),v_schema,l_update_cnt,l_insert_cnt);
        COMMIT;
    END IF;

EXCEPTION
    WHEN OTHERS THEN
        ROLLBACK TO before_merge;
        DBMS_OUTPUT.PUT_LINE('ERROR: ' || SQLERRM);
        RAISE;
END;
/

~~~END_VERSION_SECTION~~~