DEV Community: Arpit Tiwari

Updated

Arpit Tiwari — Wed, 21 May 2025 17:56:30 +0000


CREATE OR REPLACE PROCEDURE compare_tables_with_multi (
    table_names IN SYS.ODCIVARCHAR2LIST,
    l_batch_no IN NUMBER
) AS
    sql_stmt VARCHAR2(10000);
    merge_stmt CLOB;
    ip_merge_stmt_1 CLOB;
    ip_merge_stmt_2 CLOB;
    set_clause VARCHAR2(10000);
    insert_clause_dst VARCHAR2(10000);
    insert_clause_src VARCHAR2(10000);
    ip_tbl_last_update_col VARCHAR2(1000);
    ip_tbl_create_date_col VARCHAR2(1000);
    ip_tbl_primary_col VARCHAR2(1000);
    resultset tbl_nmbr;
    insert_rows_affected NUMBER(10);
    chunk_sql VARCHAR2(4000);
    task_name VARCHAR2(1000);
    l_try NUMBER(5);
    l_status NUMBER(5);
    chunk_size NUMBER(10) := 20000;
BEGIN
    FOR i IN 1..table_names.COUNT LOOP
        -- Validate table_name
        IF table_names(i) IS NULL OR NOT REGEXP_LIKE(table_names(i), '^[A-Za-z0-9_]+$') THEN
            RAISE_APPLICATION_ERROR(-20001, 'Invalid table_name: ' || table_names(i));
        END IF;

        -- Verify that table_name exists in temp_date_col_mapper
        SELECT COUNT(*)
        INTO l_status
        FROM temp_date_col_mapper
        WHERE table_name = table_names(i);

        IF l_status = 0 THEN
            RAISE_APPLICATION_ERROR(-20002, 'Table name ' || table_names(i) || ' not found in temp_date_col_mapper');
        END IF;

        -- Get table's metadata
        BEGIN
            SELECT last_update_date_col, create_date_col, primary_key, merge_stmt_1, merge_stmt_2
            INTO ip_tbl_last_update_col, ip_tbl_create_date_col, ip_tbl_primary_col, ip_merge_stmt_1, ip_merge_stmt_2
            FROM temp_date_col_mapper
            WHERE table_name = table_names(i)
            FETCH FIRST 1 ROW ONLY;
        EXCEPTION
            WHEN TOO_MANY_ROWS THEN
                RAISE_APPLICATION_ERROR(-20003, 'Multiple rows found in temp_date_col_mapper for table_name: ' || table_names(i));
            WHEN NO_DATA_FOUND THEN
                RAISE_APPLICATION_ERROR(-20004, 'No metadata found in temp_date_col_mapper for table_name: ' || table_names(i));
        END;

        -- Validate ip_tbl_primary_col
        IF NOT REGEXP_LIKE(ip_tbl_primary_col, '^[A-Za-z0-9_]+$') THEN
            RAISE_APPLICATION_ERROR(-20005, 'Invalid primary column name: ' || ip_tbl_primary_col);
        END IF;

        -- Define task name
        task_name := SUBSTR('COMPARE_' || table_names(i) || '_' || TO_CHAR(SYSDATE, 'YYYYMMDDHH24MISS'), 1, 128);

        -- Create the task
        DBMS_PARALLEL_EXECUTE.CREATE_TASK(task_name => task_name);

        -- Chunk the table based on ROW NUMBER()
        chunk_sql := 'SELECT rn AS start_id, rn + :chunk_size - 1 AS end_id ' ||
                     'FROM (SELECT ROW_NUMBER() OVER (ORDER BY ' || ip_tbl_primary_col || ') AS rn ' ||
                     'FROM cr_5_0_x_' || table_names(i) || ' a) ' ||
                     'WHERE MOD(rn - 1, :chunk_size) = 0';

        DBMS_PARALLEL_EXECUTE.CREATE_CHUNKS_BY_SQL(
            task_name => task_name,
            sql_stmt => chunk_sql,
            by_rowid => FALSE
        );

        -- Define MERGE statement
        merge_stmt := ip_merge_stmt_1 || ' WHERE rn BETWEEN :start_id AND :end_id ' || ip_merge_stmt_2;

        -- Run task in parallel
        DBMS_PARALLEL_EXECUTE.RUN_TASK(
            task_name => task_name,
            sql_stmt => merge_stmt,
            language_flag => DBMS_SQL.NATIVE,
            parallel_level => 8
        );

        -- Retry mechanism
        l_try := 0;
        l_status := DBMS_PARALLEL_EXECUTE.TASK_STATUS(task_name => task_name);
        WHILE (l_try < 2 AND l_status != DBMS_PARALLEL_EXECUTE.FINISHED) LOOP
            DBMS_OUTPUT.PUT_LINE('Something went wrong in parallel execute - retrying');
            l_try := l_try + 1;
            DBMS_PARALLEL_EXECUTE.RESUME_TASK(task_name => task_name);
            l_status := DBMS_PARALLEL_EXECUTE.TASK_STATUS(task_name => task_name);
        END LOOP;

        -- Check status and log if failed
        IF l_status != DBMS_PARALLEL_EXECUTE.FINISHED THEN
            INSERT INTO error_log (timestamp, procedure_name, error_message)
            VALUES (SYSDATE, 'compare_tables_with_multi', 'Parallel execution failed after retries for table: ' || table_names(i));
            COMMIT;
            RAISE_APPLICATION_ERROR(-20006, 'Parallel execution failed after retries for table: ' || table_names(i));
        END IF;

        -- Drop the task
        DBMS_PARALLEL_EXECUTE.DROP_TASK(task_name => task_name);
    END LOOP;

    COMMIT;

EXCEPTION
    WHEN OTHERS THEN
        INSERT INTO error_log (timestamp, procedure_name, error_message)
        VALUES (SYSDATE, 'compare_tables_with_multi', SQLERRM || ' Stack: ' || DBMS_UTILITY.FORMAT_ERROR_STACK);
        COMMIT;
        ROLLBACK;
        RAISE;
END compare_tables_with_multi;
/

Com

Arpit Tiwari — Wed, 21 May 2025 17:29:03 +0000


CREATE OR REPLACE PROCEDURE compare_tables_with_multi (
    TABLE_NAME IN VARCHAR2, l_batch_no IN NUMBER
) AS
    sql_stmt VARCHAR2(10000);
    clob CLOB;
    ip_merge_stmt_1 CLOB;
    ip_merge_stmt_2 CLOB;
    set_clause VARCHAR2(10000);
    insert_clause VARCHAR2(10000);
    insert_src VARCHAR2(10000);
    ip_tbl_last_update_col VARCHAR2(1000);
    ip_tbl_create_date_col VARCHAR2(1000);
    ip_tbl_primary_col VARCHAR2(1000);
    resultset tbl_nmbr;
    insert_rows_affected NUMBER(10);
    chunk_sql VARCHAR2(4000);
    task_name VARCHAR2(1000);
    l_try NUMBER(5);
    l_status NUMBER(5);
    chunk_size NUMBER(10) := 20000;
BEGIN
    -- Validate TABLE_NAME to prevent SQL injection and ensure it's not too long
    IF NOT REGEXP_LIKE(TABLE_NAME, '^[A-Za-z0-9_]+$') THEN
        RAISE_APPLICATION_ERROR(-20001, 'Invalid TABLE_NAME: ' || TABLE_NAME);
    END IF;

    -- Get table's meta data, ensuring only one row is returned
    BEGIN
        SELECT last_update_date_col, create_date_col, primary_key, merge_stmt_1, merge_stmt_2
        INTO ip_tbl_last_update_col, ip_tbl_create_date_col, ip_tbl_primary_col, ip_merge_stmt_1, ip_merge_stmt_2
        FROM temp_date_col_mapper
        WHERE table_name = TABLE_NAME
        FETCH FIRST 1 ROW ONLY;
    EXCEPTION
        WHEN TOO_MANY_ROWS THEN
            RAISE_APPLICATION_ERROR(-20004, 'Multiple rows found in temp_date_col_mapper for table_name: ' || TABLE_NAME);
        WHEN NO_DATA_FOUND THEN
            RAISE_APPLICATION_ERROR(-20005, 'No metadata found in temp_date_col_mapper for table_name: ' || TABLE_NAME);
    END;

    -- Validate ip_tbl_primary_col
    IF NOT REGEXP_LIKE(ip_tbl_primary_col, '^[A-Za-z0-9_]+$') THEN
        RAISE_APPLICATION_ERROR(-20002, 'Invalid primary column name: ' || ip_tbl_primary_col);
    END IF;

    -- Define task name, ensuring it doesn't exceed 128 characters
    task_name := SUBSTR('COMPARE_' || TABLE_NAME || '_' || TO_CHAR(SYSDATE, 'YYYYMMDDHH24MISS'), 1, 128);

    -- Create the task
    DBMS_PARALLEL_EXECUTE.CREATE_TASK(task_name => task_name);

    -- Chunk the table based on ROW NUMBER()
    chunk_sql := 'SELECT rn AS start_id, rn + :chunk_size - 1 AS end_id ' ||
                 'FROM (SELECT ROW_NUMBER() OVER (ORDER BY ' || ip_tbl_primary_col || ') AS rn ' ||
                 'FROM cr_5_0_x_' || TABLE_NAME || ' a) ' ||
                 'WHERE MOD(rn - 1, :chunk_size) = 0';

    DBMS_PARALLEL_EXECUTE.CREATE_CHUNKS_BY_SQL(
        task_name => task_name,
        sql_stmt => chunk_sql,
        by_rowid => FALSE
    );

    -- Define MERGE statement
    merge_stmt := ip_merge_stmt_1 || ' WHERE rn BETWEEN :start_id AND :end_id ' || ip_merge_stmt_2;

    -- Run task in parallel
    DBMS_PARALLEL_EXECUTE.RUN_TASK(
        task_name => task_name,
        sql_stmt => merge_stmt,
        language_flag => DBMS_SQL.NATIVE,
        parallel_level => 8
    );

    -- Retry mechanism
    l_try := 0;
    l_status := DBMS_PARALLEL_EXECUTE.TASK_STATUS(task_name => task_name);
    WHILE (l_try < 2 AND l_status != DBMS_PARALLEL_EXECUTE.FINISHED) LOOP
        DBMS_OUTPUT.PUT_LINE('Something went wrong in parallel execute - retrying');
        l_try := l_try + 1;
        DBMS_PARALLEL_EXECUTE.RESUME_TASK(task_name => task_name);
        l_status := DBMS_PARALLEL_EXECUTE.TASK_STATUS(task_name => task_name);
    END LOOP;

    -- Check status and log if failed
    IF l_status != DBMS_PARALLEL_EXECUTE.FINISHED THEN
        INSERT INTO error_log (timestamp, procedure_name, error_message)
        VALUES (SYSDATE, 'compare_tables_with_multi', 'Parallel execution failed after retries for table: ' || TABLE_NAME);
        COMMIT;
        RAISE_APPLICATION_ERROR(-20003, 'Parallel execution failed after retries for table: ' || TABLE_NAME);
    END IF;

    -- Drop the task
    DBMS_PARALLEL_EXECUTE.DROP_TASK(task_name => task_name);

    COMMIT;

EXCEPTION
    WHEN OTHERS THEN
        INSERT INTO error_log (timestamp, procedure_name, error_message)
        VALUES (SYSDATE, 'compare_tables_with_multi', SQLERRM || ' Stack: ' || DBMS_UTILITY.FORMAT_ERROR_STACK);
        COMMIT;
        ROLLBACK;
        RAISE;
END compare_tables_with_multi;
/

Draft drop

Arpit Tiwari — Mon, 12 May 2025 09:45:30 +0000


CREATE OR REPLACE PROCEDURE purge_worker_proc (
    p_start_id      IN ROWID,
    p_end_id        IN ROWID,
    p_table         IN VARCHAR2,
    p_pk            IN VARCHAR2,
    p_schema        IN VARCHAR2,
    p_purge_dt      IN DATE,
    LAST_UPDATE_COL IN VARCHAR2,
    CREATE_DATE_COL IN VARCHAR2,
    l_batch_no      IN NUMBER
)
AS
    TYPE t_pk_tab IS TABLE OF VARCHAR2(4000);
    TYPE t_rowid_tab IS TABLE OF ROWID;

    v_pks     t_pk_tab;
    v_rowids  t_rowid_tab;
    v_sql     CLOB;
    v_cursor  SYS_REFCURSOR;
BEGIN
    -- Build dynamic SQL to select PK and ROWID
    v_sql := 'SELECT TO_CHAR(' || p_pk || '), ROWID ' ||
             'FROM ' || p_schema || '.' || p_table || ' ' ||
             'WHERE NVL(' || LAST_UPDATE_COL || ', ' || CREATE_DATE_COL || ') < :purge_dt ' ||
             'AND ROWID BETWEEN :start_id AND :end_id';

    DBMS_OUTPUT.PUT_LINE('SQL to fetch rows: ' || v_sql);

    -- Fetch PKs and ROWIDs to delete
    OPEN v_cursor FOR v_sql USING p_purge_dt, p_start_id, p_end_id;
    FETCH v_cursor BULK COLLECT INTO v_pks, v_rowids;
    CLOSE v_cursor;

    IF v_pks.COUNT > 0 THEN
        SAVEPOINT before_insert;

        BEGIN
            -- Insert into deleted_records table
            FORALL i IN v_pks.FIRST .. v_pks.LAST
                INSERT INTO deleted_records (TABLE_NAME, TABLE_PK_VALUE, DELETION_STATUS, BATCH_NO)
                VALUES (p_table, v_pks(i), 'DEL', l_batch_no);

            -- Build delete SQL using ROWID
            v_sql := 'DELETE FROM ' || p_schema || '.' || p_table || ' WHERE ROWID = :1';

            -- Delete rows using FORALL
            FORALL i IN v_rowids.FIRST .. v_rowids.LAST
                EXECUTE IMMEDIATE v_sql USING v_rowids(i);

            COMMIT;
            DBMS_OUTPUT.PUT_LINE('Deleted ' || v_pks.COUNT || ' rows.');

        EXCEPTION
            WHEN OTHERS THEN
                ROLLBACK TO before_insert;
                DBMS_OUTPUT.PUT_LINE('Error occurred, rolled back: ' || SQLERRM);
        END;

    ELSE
        DBMS_OUTPUT.PUT_LINE('No rows found for deletion.');
    END IF;
END;
/

Draft

Arpit Tiwari — Mon, 12 May 2025 09:34:48 +0000


CREATE OR REPLACE PROCEDURE CR_ARCHIVE_REGISTRY_PURGE_WORKER_PROC (
    p_start_id      IN ROWID,
    p_end_id        IN ROWID,
    p_table         IN VARCHAR2,
    p_pk            IN VARCHAR2,
    p_schema        IN VARCHAR2,
    p_purge_dt      IN DATE,
    LAST_UPDATE_COL IN VARCHAR2,
    CREATE_DATE_COL IN VARCHAR2,
    l_batch_no      IN NUMBER
)
AS
    v_sql             CLOB;
    v_cursor          SYS_REFCURSOR;
    v_pk              VARCHAR2(4000);
    v_rowid           ROWID;
    v_deleted_pk      VARCHAR2(4000);
    v_total_deleted   NUMBER := 0;
BEGIN
    -- 1. Build and run the SELECT to get ROWIDs to purge
    v_sql := 'SELECT TO_CHAR(' || p_pk || '), ROWID ' ||
             'FROM ' || p_schema || '.' || p_table || ' ' ||
             'WHERE NVL(' || LAST_UPDATE_COL || ', ' || CREATE_DATE_COL || ') < :purge_dt ' ||
             'AND ROWID BETWEEN :start_id AND :end_id';

    -- Open cursor for row-by-row processing
    OPEN v_cursor FOR v_sql USING p_purge_dt, p_start_id, p_end_id;
    LOOP
        FETCH v_cursor INTO v_pk, v_rowid;
        EXIT WHEN v_cursor%NOTFOUND;

        -- 2. Delete the row and capture the primary key of the deleted row
        v_sql := 'DELETE FROM ' || p_schema || '.' || p_table || ' ' ||
                 'WHERE ROWID = :1 ' ||
                 'RETURNING TO_CHAR(' || p_pk || ') INTO :2';

        EXECUTE IMMEDIATE v_sql
            INTO v_deleted_pk
            USING v_rowid;

        -- 3. Log the successfully deleted record
        INSERT INTO t_deleted_records (TABLE_NAME, TABLE_PK_VALUE, DELETION_STATUS, BATCH_NO)
        VALUES (p_table, v_deleted_pk, 'SUCCESS', l_batch_no);

        v_total_deleted := v_total_deleted + 1;

        COMMIT; -- Commit each row deletion
    END LOOP;
    CLOSE v_cursor;

    -- Output the total number of deleted rows
    IF v_total_deleted > 0 THEN
        DBMS_OUTPUT.PUT_LINE('Deleted ' || v_total_deleted || ' rows.');
    ELSE
        DBMS_OUTPUT.PUT_LINE('No rows found for deletion.');
    END IF;
EXCEPTION
    WHEN OTHERS THEN
        DBMS_OUTPUT.PUT_LINE('Error: ' || SQLERRM);
        IF v_cursor%ISOPEN THEN
            CLOSE v_cursor;
        END IF;
        RAISE;
END;
/

Test del

Arpit Tiwari — Mon, 12 May 2025 08:34:52 +0000


CREATE OR REPLACE PROCEDURE purge_worker_proc (
    p_start_id      IN ROWID,
    p_end_id        IN ROWID,
    p_table         IN VARCHAR2,
    p_pk            IN VARCHAR2,
    p_schema        IN VARCHAR2,
    p_purge_dt      IN DATE,
    LAST_UPDATE_COL IN VARCHAR2,
    CREATE_DATE_COL IN VARCHAR2,
    l_batch_no      IN NUMBER
)
AS
    TYPE t_pk_tab IS TABLE OF VARCHAR2(4000);
    TYPE t_rowid_tab IS TABLE OF ROWID;

    v_pks     t_pk_tab;
    v_rowids  t_rowid_tab;
    v_sql     CLOB;
    v_cursor  SYS_REFCURSOR;
    v_pk_val  VARCHAR2(4000);
    v_sql_del VARCHAR2(1000);
BEGIN
    -- Build dynamic SQL to select PK and ROWID
    v_sql := 'SELECT TO_CHAR(' || p_pk || '), ROWID ' ||
             'FROM ' || p_schema || '.' || p_table ||
             ' WHERE NVL(' || LAST_UPDATE_COL || ', ' || CREATE_DATE_COL || ') < :purge_dt ' ||
             'AND ROWID BETWEEN :start_id AND :end_id';

    DBMS_OUTPUT.PUT_LINE('SQL to fetch rows: ' || v_sql);

    -- Fetch PKs and ROWIDs
    OPEN v_cursor FOR v_sql USING p_purge_dt, p_start_id, p_end_id;
    FETCH v_cursor BULK COLLECT INTO v_pks, v_rowids;
    CLOSE v_cursor;

    IF v_pks.COUNT > 0 THEN
        -- Loop row by row
        FOR i IN v_rowids.FIRST .. v_rowids.LAST LOOP
            BEGIN
                -- Delete with RETURNING INTO
                v_sql_del := 'DELETE FROM ' || p_schema || '.' || p_table || 
                             ' WHERE ROWID = :1 RETURNING ' || p_pk || ' INTO :2';

                EXECUTE IMMEDIATE v_sql_del USING v_rowids(i), OUT v_pk_val;

                -- Insert only if delete succeeded
                INSERT INTO deleted_records (TABLE_NAME, TABLE_PK_VALUE, DELETION_STATUS, BATCH_NO)
                VALUES (p_table, v_pk_val, 'DEL', l_batch_no);

            EXCEPTION
                WHEN OTHERS THEN
                    DBMS_OUTPUT.PUT_LINE('Skipping row due to error: ' || SQLERRM);
            END;
        END LOOP;

        COMMIT;
        DBMS_OUTPUT.PUT_LINE('Deleted ' || v_pks.COUNT || ' rows.');

    ELSE
        DBMS_OUTPUT.PUT_LINE('No rows found for deletion.');
    END IF;
END;
/

[Boost]

Arpit Tiwari — Thu, 08 May 2025 13:06:32 +0000

Test

Arpit Tiwari — Tue, 06 May 2025 10:31:50 +0000


CREATE OR REPLACE PROCEDURE compare_tables(cur_tables IN VARCHAR2) AS
    l_sp_name            VARCHAR2(50) := 'COMPARE_TABLES';
    sql_stmt             VARCHAR2(4000);
    merge_stmt           CLOB;
    ip_merge_stmt_1      CLOB;
    ip_merge_stmt_2      CLOB;
    ip_tbl_last_update_col VARCHAR2(1000);
    ip_tbl_create_date_col VARCHAR2(1000);
    ip_tbl_primary_col   VARCHAR2(1000);
    REC_TABLE            tbl_str; -- Custom type for tracking PKs
    batch_size           NUMBER := 500000;
    total_rows           NUMBER := 0;
    batch_start          NUMBER := 1;
    batch_end            NUMBER;
    batch_number         NUMBER := 1;
    l_error_message      VARCHAR2(3000);
BEGIN
    -- Get metadata from mapper
    SELECT last_update_date, create_date, primary_key, merge_stmt_1, merge_stmt_2
    INTO ip_tbl_last_update_col, ip_tbl_create_date_col, ip_tbl_primary_col, ip_merge_stmt_1, ip_merge_stmt_2
    FROM temp_date_col_mapper
    WHERE table_name = cur_tables;

    -- Get total row count
    sql_stmt := 'SELECT COUNT(*) FROM cr_5_0_x.' || cur_tables;
    EXECUTE IMMEDIATE sql_stmt INTO total_rows;

    -- Batch loop
    WHILE batch_start <= total_rows LOOP
        batch_end := batch_start + batch_size - 1;
        IF batch_end > total_rows THEN
            batch_end := total_rows;
        END IF;

        -------------------------------------------------------------------
        -- Reconciliation Logging: INSERT rows
        -------------------------------------------------------------------
        sql_stmt := 'INSERT INTO reconciliation_log_archive (table_name, primary_key_val, action_type, action_time)
                     SELECT ''' || cur_tables || ''', src.' || ip_tbl_primary_col || ', ''INSERT'', SYSTIMESTAMP
                     FROM (
                         SELECT a.*, ROWNUM AS rn FROM cr_5_0_x.' || cur_tables || ' a
                     ) src
                     WHERE src.rn BETWEEN ' || batch_start || ' AND ' || batch_end || '
                     AND NOT EXISTS (
                         SELECT 1 FROM ' || cur_tables || ' tgt
                         WHERE tgt.' || ip_tbl_primary_col || ' = src.' || ip_tbl_primary_col || '
                     )';
        EXECUTE IMMEDIATE sql_stmt;

        -------------------------------------------------------------------
        -- Reconciliation Logging: UPDATE rows
        -------------------------------------------------------------------
        sql_stmt := 'INSERT INTO reconciliation_log_archive (table_name, primary_key_val, action_type, action_time)
                     SELECT ''' || cur_tables || ''', src.' || ip_tbl_primary_col || ', ''UPDATE'', SYSTIMESTAMP
                     FROM (
                         SELECT a.*, ROWNUM AS rn FROM cr_5_0_x.' || cur_tables || ' a
                     ) src
                     JOIN ' || cur_tables || ' tgt
                     ON tgt.' || ip_tbl_primary_col || ' = src.' || ip_tbl_primary_col || '
                     WHERE src.rn BETWEEN ' || batch_start || ' AND ' || batch_end || '
                     AND (' || ip_merge_stmt_1 || ') != (' || ip_merge_stmt_2 || ')';
        EXECUTE IMMEDIATE sql_stmt;

        -------------------------------------------------------------------
        -- Build and execute your original MERGE
        -------------------------------------------------------------------
        merge_stmt := ip_merge_stmt_1 ||
                      TO_CLOB(' WHERE rn BETWEEN ') ||
                      TO_CLOB(batch_start) ||
                      TO_CLOB(' AND ') ||
                      TO_CLOB(batch_end) ||
                      TO_CLOB(') src ') ||
                      ip_merge_stmt_2;

        DBMS_OUTPUT.PUT_LINE('Batch from ' || batch_start || ' to ' || batch_end);
        DBMS_OUTPUT.PUT_LINE('MERGE STMT: ' || merge_stmt);

        EXECUTE IMMEDIATE merge_stmt;

        -- Optional: Returning primary keys
        -- EXECUTE IMMEDIATE merge_stmt RETURNING BULK COLLECT INTO REC_TABLE;

        batch_start := batch_end + 1;
        batch_number := batch_number + 1;
        COMMIT;
    END LOOP;

EXCEPTION
    WHEN OTHERS THEN
        dbms_output.put_line('Error occurred in batch: ' || batch_number);
        dbms_output.put_line('Stack: ' || DBMS_UTILITY.FORMAT_ERROR_STACK);
        dbms_output.put_line('Backtrace: ' || DBMS_UTILITY.FORMAT_ERROR_BACKTRACE);
        ROLLBACK;
        l_error_message := DBMS_UTILITY.format_error_backtrace || DBMS_UTILITY.format_error_stack;
        sp_ael_arch_error_log(l_sp_name, cur_tables, batch_number, batch_start, l_error_message);
END compare_tables;
/

Testing

Arpit Tiwari — Mon, 05 May 2025 07:16:32 +0000


CREATE OR REPLACE PROCEDURE CR_ARCHIVE_REGISTRY.COMPARE_TABLES(cur_tables IN VARCHAR2, l_batch_no IN NUMBER)
AS
    l_prod_schema VARCHAR2(30) := 'CR_5_0_X';
    l_arch_schema VARCHAR2(30) := 'CR_ARCHIVE';
    l_error_message VARCHAR2(3000);

    -- Variables for table names and statements
    sql_stmt VARCHAR2(4000);
    merge_stmt CLOB;
    ip_merge_stmt_1 CLOB;
    ip_merge_stmt_2 CLOB;

    ip_tbl_last_update_col VARCHAR2(1000);
    ip_tbl_create_date_col VARCHAR2(1000);
    ip_tbl_primary_col VARCHAR2(1000);

    -- Record to store affected primary keys (for logging)
    TYPE rec_table IS RECORD (
        rn NUMBER
    );
    TYPE t_rec_table IS TABLE OF rec_table;
    v_inserted_rows t_rec_table;
    v_updated_rows t_rec_table;

    batch_size NUMBER := 500000;
    total_rows NUMBER := 0;
    batch_start NUMBER := 1;
    batch_end NUMBER;
    batch_number NUMBER;

BEGIN
    -- Step 1: Create or ensure the logging table exists
    EXECUTE IMMEDIATE '
        CREATE TABLE IF NOT EXISTS CR_ARCHIVE_REGISTRY.merge_log (
            log_id NUMBER GENERATED ALWAYS AS IDENTITY,
            table_name VARCHAR2(50),
            record_id VARCHAR2(50),
            column_name VARCHAR2(50),
            old_value VARCHAR2(4000),
            new_value VARCHAR2(4000),
            action VARCHAR2(20),
            batch_number NUMBER,
            log_timestamp TIMESTAMP DEFAULT SYSTIMESTAMP
        )';

    -- Step 2: Get the table's important column names
    SELECT last_updated_column, create_date, primary_key, merge_stmt_1, merge_stmt_2
    INTO ip_tbl_last_update_col, ip_tbl_create_date_col, ip_tbl_primary_col, ip_merge_stmt_1, ip_merge_stmt_2
    FROM tmp_date_col_mapper
    WHERE table_name = cur_tables;

    -- Step 3: Get the total number of rows to process
    sql_stmt := 'SELECT COUNT(*) FROM cr_5_0_x.' || cur_tables;
    EXECUTE IMMEDIATE sql_stmt INTO total_rows;

    -- Batch processing loop
    batch_number := l_batch_no;
    WHILE batch_start <= total_rows LOOP
        batch_end := batch_start + batch_size - 1;
        IF batch_end > total_rows THEN
            batch_end := total_rows;
        END IF;

        -- Step 4: Construct the merge statement dynamically for each batch
        merge_stmt := ip_merge_stmt_1 || ' '
                   || ' TO CLOB(''WHERE rn BETWEEN ' || batch_start || ' AND ' || batch_end || ') '
                   || ' TO CLOB('') '
                   || ' TO CLOB('') '
                   || ip_merge_stmt_2 || ' '
                   || ' RETURNING ' || ip_tbl_primary_col || ' BULK COLLECT INTO v_inserted_rows, v_updated_rows';

        -- Step 5: Execute the merge statement
        BEGIN
            EXECUTE IMMEDIATE merge_stmt;

            -- Log inserted rows
            FOR i IN 1..v_inserted_rows.COUNT LOOP
                INSERT INTO CR_ARCHIVE_REGISTRY.merge_log (
                    table_name, record_id, column_name, old_value, new_value, action, batch_number
                )
                SELECT cur_tables, v_inserted_rows(i).rn, 'INSERT', NULL, NULL, 'INSERT', batch_number
                FROM DUAL;
            END LOOP;

            -- Log updated rows
            FOR i IN 1..v_updated_rows.COUNT LOOP
                INSERT INTO CR_ARCHIVE_REGISTRY.merge_log (
                    table_name, record_id, column_name, old_value, new_value, action, batch_number
                )
                SELECT cur_tables, v_updated_rows(i).rn, 'UPDATE', NULL, NULL, 'UPDATE', batch_number
                FROM DUAL;
            END LOOP;

            -- Output progress
            dbms_output.put_line('For Table: ' || cur_tables || ' with Batch from ' || batch_start || ' - ' || batch_end);
            dbms_output.put_line('Rows affected: ' || (v_inserted_rows.COUNT + v_updated_rows.COUNT));

            -- Step 6: Reconciliation step for inserts and updates
            sql_stmt := 'SELECT COUNT(*) FROM cr_5_0_x.' || cur_tables || ' WHERE rn BETWEEN ' || batch_start || ' AND ' || batch_end
                     || ' AND (SYSDATE-COALESCE(AUDIT_LAST_UPDATE_DATE, AUDIT_CREATE_DATE, SYSDATE)) = (SYSDATE-COALESCE(AUDIT_LAST_UPDATE_DATE, AUDIT_CREATE_DATE, SYSDATE))';
            EXECUTE IMMEDIATE sql_stmt INTO total_rows;
            dbms_output.put_line('Reconciliation - Rows updated in source table: ' || total_rows);

            sql_stmt := 'SELECT COUNT(*) FROM cr_archive.' || cur_tables || ' WHERE rn BETWEEN ' || batch_start || ' AND ' || batch_end
                     || ' AND AUDIT_CREATE_DATE IS NOT NULL';
            EXECUTE IMMEDIATE sql_stmt INTO total_rows;
            dbms_output.put_line('Reconciliation - Rows inserted in target table: ' || total_rows);

            -- Step 7: Log discrepancies if counts don't match
            IF v_inserted_rows.COUNT + v_updated_rows.COUNT != total_rows THEN
                INSERT INTO CR_ARCHIVE_REGISTRY.merge_log (
                    table_name, record_id, column_name, old_value, new_value, action, batch_number
                )
                VALUES (cur_tables, NULL, 'COUNT_MISMATCH', v_inserted_rows.COUNT + v_updated_rows.COUNT, total_rows, 'DISCREPANCY', batch_number);
            END IF;

            -- Commit after each batch
            COMMIT;

        EXCEPTION
            WHEN OTHERS THEN
                dbms_output.put_line('Error in processing tables: ' || cur_tables || ' in batch');
                dbms_output.put_line('Error raised in: ' || $$plsql_unit || ' at line ' || $$plsql_line || ' - ' || sqlerrm);
                dbms_output.put_line('ERROR_STACK: ' || DBMS_UTILITY.FORMAT_ERROR_STACK);
                dbms_output.put_line('ERROR_BACKTRACE: ' || DBMS_UTILITY.FORMAT_ERROR_BACKTRACE);

                -- Rollback the current batch, insert the error log, and continue with the next table
                ROLLBACK;
                l_error_message := dbms_utility.format_error_backtrace || ' ' || dbms_utility.format_error_stack;
                INSERT INTO sp_ael_arch_error_log (sp_name, cur_table, batch_number, batch_start, l_error_message)
                VALUES ($$plsql_unit, cur_tables, batch_number, batch_start, l_error_message);
                COMMIT;
                RAISE;
        END;

        -- Step 8: Move to the next batch
        batch_start := batch_end + 1;
        batch_number := batch_number + 1;
    END LOOP;

    -- Final commit
    COMMIT;

EXCEPTION
    WHEN OTHERS THEN
        dbms_output.put_line('Error in processing tables: ' || cur_tables || ' in batch');
        dbms_output.put_line('Error raised in: ' || $$plsql_unit || ' at line ' || $$plsql_line || ' - ' || sqlerrm);
        dbms_output.put_line('ERROR_STACK: ' || DBMS_UTILITY.FORMAT_ERROR_STACK);
        dbms_output.put_line('ERROR_BACKTRACE: ' || DBMS_UTILITY.FORMAT_ERROR_BACKTRACE);
        ROLLBACK;
        RAISE;
END;
/

CREATE OR REPLACE PROCEDURE CR_ARCHIVE_REGISTRY.COMPARE_TABLES(cur_tables IN VARCHAR2, l_batch_no IN NUMBER) AS l_prod_schema VARCHAR2(30) := 'CR_5_0_X'; l_arch_schema VARCHAR2(30) := 'CR_ARCHIVE'; l_error_message VARCHAR2(3000); -- Va

Arpit Tiwari — Mon, 05 May 2025 07:15:37 +0000

[Boost]

Arpit Tiwari — Mon, 05 May 2025 07:15:21 +0000

Test1

Arpit Tiwari — Wed, 23 Apr 2025 10:15:23 +0000


CREATE OR REPLACE PROCEDURE purge_worker_proc (
    p_start_id       IN ROWID,
    p_end_id         IN ROWID,
    p_table          IN VARCHAR2,
    p_pk             IN VARCHAR2,
    p_schema         IN VARCHAR2,
    p_purge_dt       IN DATE,
    LAST_UPDATE_COL  IN VARCHAR2,
    CREATE_DATE_COL  IN VARCHAR2
) AS
    TYPE t_pk_tab    IS TABLE OF VARCHAR2(4000) INDEX BY PLS_INTEGER;
    TYPE t_rowid_tab IS TABLE OF ROWID INDEX BY PLS_INTEGER;

    v_pks       t_pk_tab;
    v_rowids    t_rowid_tab;
    v_sql_sel   CLOB;
    v_sql_del   VARCHAR2(1000);
    v_cursor    SYS_REFCURSOR;
    l_batch_no  NUMBER := 1;

    v_successful_rowids t_rowid_tab;
    v_idx INTEGER := 0;
BEGIN
    -- Step 1: Build dynamic SQL for selection
    v_sql_sel := '
        SELECT TO_CHAR(' || p_pk || '), ROWID 
        FROM ' || p_schema || '.' || p_table || '
        WHERE NVL(' || LAST_UPDATE_COL || ', ' || CREATE_DATE_COL || ') < :purge_dt
        AND ROWID BETWEEN :start_id AND :end_id';

    -- Step 2: Fetch data
    OPEN v_cursor FOR v_sql_sel USING p_purge_dt, p_start_id, p_end_id;
    FETCH v_cursor BULK COLLECT INTO v_pks, v_rowids;
    CLOSE v_cursor;

    IF v_pks.COUNT > 0 THEN
        BEGIN
            -- Step 3: Log deletions using bulk insert with SAVE EXCEPTIONS
            FORALL i IN INDICES OF v_pks SAVE EXCEPTIONS
                INSERT INTO t_deleted_records (
                    TABLE_NAME, TABLE_PK_VALUE, DELETION_STATUS, BATCH_NO
                ) VALUES (
                    p_table, v_pks(i), 'DEL', l_batch_no
                );

            -- Step 4: All inserts succeeded; delete all
            v_sql_del := 'DELETE FROM ' || p_schema || '.' || p_table || ' WHERE ROWID = :1';

            FORALL i IN INDICES OF v_rowids
                EXECUTE IMMEDIATE v_sql_del USING v_rowids(i);

            COMMIT;
            DBMS_OUTPUT.PUT_LINE('All rows inserted and deleted successfully. Total: ' || v_pks.COUNT);

        EXCEPTION
            WHEN DML_ERRORS THEN
                -- Step 5: Log insert failures
                DBMS_OUTPUT.PUT_LINE('Some inserts failed. Logging errors...');
                FOR i IN 1 .. SQL%BULK_EXCEPTIONS.COUNT LOOP
                    DBMS_OUTPUT.PUT_LINE(
                        'Insert failed at index ' || SQL%BULK_EXCEPTIONS(i).ERROR_INDEX ||
                        ': ' || SQLERRM(-SQL%BULK_EXCEPTIONS(i).ERROR_CODE)
                    );
                END LOOP;

                -- Step 6: Filter only successful inserts
                FOR i IN v_pks.FIRST .. v_pks.LAST LOOP
                    IF NOT EXISTS (
                        SELECT 1
                        FROM DUAL
                        WHERE i IN (SELECT SQL%BULK_EXCEPTIONS(j).ERROR_INDEX FROM DUAL CONNECT BY LEVEL <= SQL%BULK_EXCEPTIONS.COUNT)
                    ) THEN
                        v_idx := v_idx + 1;
                        v_successful_rowids(v_idx) := v_rowids(i);
                    END IF;
                END LOOP;

                -- Step 7: Delete only successfully inserted rows
                v_sql_del := 'DELETE FROM ' || p_schema || '.' || p_table || ' WHERE ROWID = :1';
                FORALL i IN v_successful_rowids.FIRST .. v_successful_rowids.LAST
                    EXECUTE IMMEDIATE v_sql_del USING v_successful_rowids(i);

                COMMIT;
                DBMS_OUTPUT.PUT_LINE('Deleted ' || v_successful_rowids.COUNT || ' successfully inserted rows.');
        END;
    ELSE
        DBMS_OUTPUT.PUT_LINE('No rows found for deletion.');
    END IF;
END purge_worker_proc;
/

Test

Arpit Tiwari — Wed, 23 Apr 2025 10:08:46 +0000

CREATE OR REPLACE PROCEDURE compare_tables_with_multi AS
sql_stmt VARCHAR2(10000);
merge_stmt CLOB;
ip_merge_stmt_1 CLOB;
ip_merge_stmt_2 CLOB;
set_clause VARCHAR2(10000);
insert_clause_dst VARCHAR2(10000);
insert_clause_src VARCHAR2(10000);
ip_tbl_last_update_col VARCHAR2(1000);
ip_tbl_create_date_col VARCHAR2(1000);
ip_tbl_primary_col VARCHAR2(1000);
resultset tbl_nmbr;
insert_rows_affected NUMBER(10);
v_merge_count NUMBER := 0;

chunk_sql               VARCHAR2(4000);
task_name               VARCHAR2(100);
L_TRY                   NUMBER(6);
L_STATUS                NUMBER(6);
chunk_size              NUMBER(10) := 20000;

CURSOR cur_tables IS
    SELECT *
    FROM temp_date_col_mapper
    WHERE table_name IN (
        SELECT table_name
        FROM leaf_nodes
        WHERE table_name IN ('UD_MPI_MPG_ID')
    );

BEGIN
FOR table_rec IN cur_tables LOOP
BEGIN
-- Get table metadata
SELECT last_update_date, create_date, primary_key, merge_stmt_1, merge_stmt_2
INTO ip_tbl_last_update_col, ip_tbl_create_date_col, ip_tbl_primary_col, ip_merge_stmt_1, ip_merge_stmt_2
FROM temp_date_col_mapper
WHERE table_name = table_rec.table_name;

        -- Define task name
        task_name := 'COMPARE_' || table_rec.TABLE_NAME || '_' || TO_CHAR(SYSDATE, 'YYYYMMDDHH24MISS');

        -- Create task
        DBMS_PARALLEL_EXECUTE.CREATE_TASK(task_name => task_name);

        -- Chunk SQL based on ROW_NUMBER()
        chunk_sql := 'SELECT rn AS start_id, rn + ' || chunk_size || ' - 1 AS end_id ' ||
                     'FROM (SELECT ROW_NUMBER() OVER (ORDER BY ' || ip_tbl_primary_col || ') AS rn ' ||
                     'FROM cr_s_0_x.' || table_rec.TABLE_NAME || ') ' ||
                     'WHERE MOD(rn - 1, ' || chunk_size || ') = 0';

        -- Create chunks
        DBMS_PARALLEL_EXECUTE.CREATE_CHUNKS_BY_SQL(
            task_name => task_name,
            sql_stmt  => chunk_sql,
            by_rowid  => FALSE
        );

        -- Define MERGE dynamic block that prints merged count
        merge_stmt := '
        DECLARE
            v_cnt NUMBER;
        BEGIN
            ' || ip_merge_stmt_1 || ' WHERE rn BETWEEN :start_id AND :end_id; ' || '
            v_cnt := SQL%ROWCOUNT;
            DBMS_OUTPUT.PUT_LINE(''Task: ' || ''' || task_name || ''' || ', Chunk Start ID: '' || :start_id || '', Rows Merged: '' || v_cnt);
            COMMIT;
        END;';

        -- Run merge in parallel
        DBMS_PARALLEL_EXECUTE.RUN_TASK(
            task_name      => task_name,
            sql_stmt       => merge_stmt,
            language_flag  => DBMS_SQL.NATIVE,
            parallel_level => 8
        );

        -- Handle retries if needed
        L_TRY := 0;
        L_STATUS := DBMS_PARALLEL_EXECUTE.TASK_STATUS(task_name);

        WHILE L_TRY < 2 AND L_STATUS != DBMS_PARALLEL_EXECUTE.FINISHED LOOP
            DBMS_OUTPUT.PUT_LINE('Retrying task: ' || task_name);
            L_TRY := L_TRY + 1;
            DBMS_PARALLEL_EXECUTE.RESUME_TASK(task_name);
            L_STATUS := DBMS_PARALLEL_EXECUTE.TASK_STATUS(task_name);
        END LOOP;

        -- Clean up
        DBMS_PARALLEL_EXECUTE.DROP_TASK(task_name);

    EXCEPTION
        WHEN OTHERS THEN
            DBMS_OUTPUT.PUT_LINE('Error processing table: ' || table_rec.TABLE_NAME);
            DBMS_OUTPUT.PUT_LINE('Error: ' || SQLERRM);
    END;
END LOOP;

END;