Test

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')
    );

-- 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;