10.1.7

Edit

Data Purge overview

IA Cloud Enterprise generates a considerable amount of data. Much of the data results from finished Business Processes (BP) or internal maintenance. Accumulation of obsolete data can lead to the following issues:

  • Low disk space. The database can run short of free disk space, resulting in unexpected BP errors.
  • Low performance. The increasing number of strings in the database affects their processing speed.
  • Security flaws. Sometimes, corporate security policies require removing data older than two or four weeks.

For the above reasons, it is strongly recommended to remove unwanted data from Control Tower (CT) regularly, using the Data Purge feature. The feature purges the data automatically according to a preset schedule.

Additionally, separate purge procedures are available for the following components:

Best practices

As you start working with a BP that writes data to a disk, make sure to enable automatic purging for Control Tower data. If you fail to do this before you start, here is what you must do:

  1. Once you see that the free space on the database server is quickly diminishing, and the data occupies about 70-75% of the disk, enable automatic purging for 20 hours with minimal delay.
  2. If it is too late for Step 1, run forced purge. Remember, the procedure irrevocably removes all data before a specific date.
  3. As a last resort, you can do the following:
    1. Mount another disk with enough available space.
    2. Copy the data to it.
    3. Clean up the AwsHitQuestion table. By default, AwsHitQuestion occupies about 70% of the available disk space. After cleaning it up, you can free another 30-40%.
    4. Execute shrink_db to restore all freed extends to the system.
    5. Restore the copied data and enable automatic purging.

It is also recommended to monitor the Analytics component regularly and clean it up monthly.

Purge CT data

The standard CT purge procedure cleans up the data of completed BPs. It also verifies that the process is finished correctly with due account for all links and the data integrity inside the database is preserved.

Out-of-the-box purge

The out-of-the-box scheduler in Control Tower is used for purging data from the following main tables:

SchemaTableDescription
ctAwsHitQuestionContains JSON with the answer processing output for a record.
ctAwsHitAssignmentAnswerContains raw answers given by a worker to a record.
ctHitSubmissionDataItemContains JSON with the input data for a record.
ctHitSubmissionDataItemHistoryContains snapshots of the existing record input data if it is modified by step transition (for example, two records are merged).
ctHitDataItemLogContains a log of life cycle events for a record, for example, posting, submission, approval.
ctFILEContains a file description, for example, name and type.
ctDATA_STOREContains the binary content of files.
ctEVENT_TRACKINGContains a text description of an event, for example, a message or an exception stack trace.
ctEVENT_OBJECTContains a reference to an object-induced event, for example, a BP step or a definition.

The entity-relationship diagram is as follows:

To set a purge schedule in Control Tower, do as follows:

  1. Go to System Settings > System Preferences > Schedule Settings.
  2. Specify the period and duration of the scheduled purge and other required parameters.

For more information on setting the purge schedule, see Edit Data Purge settings.

Forced purge

Forced purge is applied manually. Unlike in case with the standard procedure, while doing the type of cleansing, specify the date after which the data must be preserved. All data before this date is cleaned by id.

Data Purge produces a high load on a disk. Before running the procedure, make sure there are no non-completed BPs started before the purge date and you understand what you are going to do. Otherwise, a forced purge may lead to unpredictable consequences.

caution

Forced purge uses run.createDate to get runs for purging, which can lead to losing some data of the runs closed after the last date.

important

Forced purge is not intended for consistent data removal. For some cases, data after the last date can be affected.

Forced purge can clean up the following tables:

SchemaTableDescription
ctAuditDataContains audit information stored by Control Tower objects.
ctAwsHitAssignmentContains job execution information per single process task.
ctAwsHitQuestionContains JSON with the answer processing output for a record.
ctCustomQualificationContains qualifications that are different from the ones integrated into a Crowds.
ctEndpointTaskAppears as a clone of AwsHit after AwsHit is submitted successfully to a Crowd.
ctEndPointTaskAssignmentIt is a temporal entity. Contains a number of assignments accepted by workers and appears when a worker accepts a task.
ctHitRecordSourceShows the origin of HSDI in AwsHit: gold, marked, ordinary.
ctPackageContains the description of created packages in Control Tower.
ctWorkerActivityLogContains worker or Bot assignment statistics.
ctWorkerDayActivityContains daily worker or Bot statistics.
dsdata_store_auditContains a log of activities related to Data Stores.
wsaud_qualificationContains an audit of qualification changes in WorkSpace.
wsmessageContains WorkSpace messages.
wshit_submissionContains workers' responses in the key-value format.

To force purge these tables, do as follows:

  1. Via console, connect to the workfusion MS SQL database.

  2. Create a purgeData_forceById generic procedure. The process removes all rows where IDs are lower than the i_lastId value.

    MS SQL purgeData_forceById (with progress tracking) Prerequisites:

    -- as db_owner (sys.admin):
create table ct.force_purge_log
( id bigint     identity(1,1)   not null primary key
, log_date      datetime2       not null default getUTCDate()
, table_name    nvarchar(128)   not null
, n_rows        bigint          not null
);

    Stored procedure:

    if object_id('[ct].[purgeData_forceById]') is not null drop procedure [ct].[purgeData_forceById];
go
create procedure [ct].[purgeData_forceById]
( @i_tableName  nvarchar(63)
, @i_idColumn   nvarchar(63)
, @i_lastId     bigint
, @i_batchSize  bigint       = 30000
, @i_sleepTime  nvarchar(12) = '00:00:00.01'
) as
begin
--set nocount on;
set xact_abort on;
    /***
    PRE-REQUESITES:
        create table ct.force_purge_log
        ( id bigint     identity(1,1)   not null primary key
        , log_date      datetime2       not null default getUTCDate()
        , table_name    nvarchar(128)   not null
        , n_rows        bigint          not null
        );
    CHANGELOG: 10.1-rev-5
        2021-01-22 apouckatch   added: logging, @i_sleepTime
        2020-11-11 ksakharchuk  changed to delete top() and removed transaction
        2020-10-02 apouckatch   commit every batch (instead of once per table)
        2020-02-01 onovak       adopt for MSSQL
        2018-08-31 apouckatch   lower a transaction level to READ COMMITTED
        2018-08-21 apouckatch   created
    ***/
    declare @v_tabCount  bigint;
    declare @v_rowCount  bigint;
    declare @v_rowTotal  bigint;
    declare @batchSize   bigint = @i_batchSize;
    declare @batchDeleteSQL nvarchar(max);
        
    -- check the table name in DB dictionary (prevent an SQL injection):
    select @v_tabCount = count(*)
    from information_schema.columns as c
    where c.table_schema = 'ct'
      and c.table_name   = @i_tableName
      and c.column_name  = @i_idColumn;   
   
if @v_tabCount >= 1
begin
try
    set @v_rowTotal = 0;
   
    set @batchDeleteSQL  = N'
        declare @row_count bigint;
        while (1=1)
        begin
                delete top ('+cast(@batchSize as nvarchar(10))+')
                from ct.['+ @i_tableName + ']
                where [' + @i_idColumn + '] <= ' + cast(@i_lastId as nvarchar(100)) +';
   
                set @row_count = @@rowcount;
                if @row_count = 0
                    break;
                set @v_rowTotal =@v_rowTotal + @row_count;
 
                insert into ct.force_purge_log( table_name, n_rows)
                values( ''' + @i_tableName + ''', @row_count);
                waitfor delay @sleep;
        end;';
        
    exec sp_executeSql @batchDeleteSQL, N'@sleep nvarchar(12), @v_rowTotal bigint output', @sleep = @i_sleepTime, @v_rowTotal = @v_rowTotal output;
   
end try
begin catch
    declare @errormessage   nvarchar(4000)  = error_message();
    declare @errorseverity  int             = error_severity();
    declare @errorstate     int             = error_state();
    raiserror(@errormessage, @errorseverity, @errorstate);
end catch;
   
select @i_tableName  as Table_Name
     , @v_rowTotal   as Deleted_Rows;
   
end;
go

    MS SQL purgeData_forceById (without progress tracking, obsolete) 

    if object_id('[ct].[purgeData_forceById]') is not null drop procedure [ct].[purgeData_forceById];
go
create procedure [ct].[purgeData_forceById]
( @i_tableName  nvarchar(63)
, @i_idColumn   nvarchar(63)
, @i_lastId     bigint
, @i_batchSize  bigint = 30000
) as
begin
--set nocount on;
set xact_abort on;
    /*** CHANGELOG: 10.1-rev-3
    2020-11-11 ksakharchuk  changed to delete top() and removed transaction
    2020-10-02 apouckatch   commit every batch (instead of once per table)
    2020-02-01 onovak       adopt for MSSQL
    2018-08-31 apouckatch   lower a transaction level to READ COMMITTED
    2018-08-21 apouckatch   created
    ***/
    declare @v_tabCount  bigint;
    declare @v_rowCount  bigint;
    declare @v_rowTotal  bigint;
    declare @batchSize   bigint = @i_batchSize;
    declare @batchDeleteSQL nvarchar(max);
    declare @sleep_time  nvarchar(12) = '00:00:01';

    -- check the table name in DB dictionary (prevent an SQL injection):
    select @v_tabCount = count(*)
    from information_schema.columns as c
    where c.table_schema = 'ct'
      and c.table_name   = @i_tableName
      and c.column_name  = @i_idColumn;     

if @v_tabCount >= 1
begin try
    set @v_rowTotal = 0;

    set @batchDeleteSQL  = N'
        declare @row_count int = 1;
        while (@row_count > 0)
        begin
                delete top ('+cast(@batchSize as nvarchar(10))+') 
                from ct.['+ @i_tableName + ']
                where [' + @i_idColumn + '] <= ' + cast(@i_lastId as nvarchar(100)) +';

                set @row_count = @@rowcount;
                set @v_rowTotal =@v_rowTotal + @row_count;
                waitfor delay @sleep;
        end;';
  
    --set transaction isolation level read committed;
    --print @batchDeleteSQL
    exec sp_executesql @batchDeleteSQL, N'@sleep nvarchar(12), @v_rowTotal bigint output', @sleep = @sleep_time, @v_rowTotal = @v_rowTotal output;

end try
begin catch
    declare @errormessage nvarchar(4000) = error_message();
    declare @errorseverity int = error_severity();
    declare @errorstate    int = error_state();
    raiserror(@errormessage, @errorseverity, @errorstate);
end catch;

select @i_tableName  as Table_Name
     , @v_rowTotal   as Deleted_Rows;

end;
go

  3. Edit the value of last date: set @last_date = '<insert-your-date-here>';. All data before this date will be removed.

  4. Copy and save the following script. Remember to redefine @last_date.

    MS SQL script 

    declare @last_date          datetime = '2020-12-31'
    ,   @preserve_active    bit      = 1            -- keep any active RUNs and their BPs
    ,   @preserve_draft     bit      = 1            -- keep 'DRAFT' BPs
    ,   @batch_size         int      = 3000
    ,   @last_run_id        bigint
    ;
with cte_min_active_run as
(   select  min(r_min.id) as run_id
    from ct.run as r_min
    where r_min.status not in
        ( 'COMPLETED'
        , 'DELETED'
        , case when @preserve_draft = 0 then 'DRAFT' else '-dummy-' end
        )
)
, cte_first_active_root_run as
(   select first_act_root_run.id as run_id
    from ct.run             as first_act_run_full
    join cte_min_active_run as first_act_run_1    on first_act_run_full.id   = first_act_run_1.run_id
    join ct.run             as first_act_root_run on first_act_root_run.uuid = first_act_run_full.rootRunUUID
)
, cte_upper_limit_run as
(   select coalesce
        (   (select run_id from cte_first_active_root_run where @preserve_active = 1)
        ,   (select max(id) + 1 from ct.run)
        ,   1
        ) as run_id
)
select @last_run_id = max(id)
from ct.run as r
where r.id < (select run_id from cte_upper_limit_run)
    and r.startDate <= @last_date
;
/*** previous version without preserving of runs: ***
declare @last_run_id  bigint = (select isnull( max(id) , 0) as last_run_id  from ct.Run                    as r  where r.startDate     <  @last_date  );
***/
declare @last_item_id bigint = (select isnull( max(id) , 0) as last_item_id from ct.HitSubmissionDataItem  as i  where i.run_id        <= @last_run_id);
declare @last_aud_id  bigint = (select isnull( max(id) , 0) as last_aud_id  from ct.AuditData              as a  where a.createdDate   <  @last_date  );
declare @last_hit_id  bigint = (select isnull( max(id) , 0) as last_hit_id  from ct.AwsHit                 as h  where h.run_id        <= @last_run_id);
declare @last_trk_id  bigint = (select isnull( max(tracker_id), 0) as tr_id from ct.DataItemTrackerHitLink as tr where tr.data_item_id <= @last_item_id);
declare @last_aha_id  bigint = (select isnull( max(id) , 0) as last_aha_id  from ct.AwsHitAssignment       as ha where ha.hit_id       <= @last_hit_id);
declare @last_et_id   bigint = (select isnull( max(id) , 0) as last_et_id   from ct.EndpointTask           as et where et.run_id       <= @last_run_id);
declare @last_ev_id   bigint = (select isnull( max(id) , 0) as last_ev_id   from ct.event_tracking         as et where et.created_time <  @last_date  );
declare @last_file_id bigint = (select isnull( max(id) , 0) as last_file_id from ct.[file]                 as f  where f.runid         <= @last_run_id);
declare @last_ds_id   bigint =
(   select coalesce
        (   (   -- case 1: skip Data_Store that has the [file] not purged this time
                select min(f.data_store_id) - 1 as last_ds_id
                from ct.[file] f
                where f.runid <= @last_run_id
                and exists
                    (   select 1
                        from ct.[file] fr
                        where fr.data_store_id = f.data_store_id
                        and fr.id > (select max(id) from ct.[file] where runid <= @last_run_id)
                    )
            )
        ,   (   -- case 2: just fetch Data_Store for the last purged [file]
                select max(f.data_store_id) as last_ds_id
                from ct.[file] f
                where f.runid <= @last_run_id
            )
        ,   -- case 3: dummy id
            0
        ) as last_ds_id
);
 
-- output interim results:
select
      @last_run_id  as last_run_id
    , @last_item_id as last_item_id
    , @last_aud_id  as last_aud_id
    , @last_hit_id  as last_hit_id
    , @last_trk_id  as last_trk_id
    , @last_aha_id  as last_aha_id
    , @last_et_id   as last_et_id
    , @last_ev_id   as last_ev_id
    , @last_file_id as last_file_id
    , @last_ds_id   as last_ds_id
;
  
begin try
    exec ct.purgeData_forceById 'AwsHitQuestion'                  , 'item_id'          , @last_item_id, @batch_size;
    exec ct.purgeData_forceById 'HitDataItemLog'                  , 'item_id'          , @last_item_id, @batch_size;
    exec ct.purgeData_forceById 'HitSubmissionDataItemHistory'    , 'item_id'          , @last_item_id, @batch_size;
    exec ct.purgeData_forceById 'PluginExecutionLog'              , 'data_item_id'     , @last_item_id, @batch_size;
    exec ct.purgeData_forceById 'HitRecordSource'                 , 'runId'            , @last_run_id , @batch_size;
    exec ct.purgeData_forceById 'AuditData'                       , 'id'               , @last_aud_id , @batch_size;
    exec ct.purgeData_forceById 'HitDataItemLog'                  , 'assignment_id'    , @last_aha_id , @batch_size;
    exec ct.purgeData_forceById 'DataItemTrackerHitLink'          , 'tracker_id'       , @last_trk_id , @batch_size;
    exec ct.purgeData_forceById 'DataItemTrackerLog'              , 'tracker_id'       , @last_trk_id , @batch_size;
    exec ct.purgeData_forceById 'DataItemTracker'                 , 'id'               , @last_trk_id , @batch_size;
    exec ct.purgeData_forceById 'AwsHitAssignmentAnswer'          , 'assignment_id'    , @last_aha_id , @batch_size;
    exec ct.purgeData_forceById 'WorkerFitnessHistoryAssignment'  , 'assignmentId'     , @last_aha_id , @batch_size;
    exec ct.purgeData_forceById 'AwsHitAssignment'                , 'hit_id'           , @last_hit_id , @batch_size;
    exec ct.purgeData_forceById 'HitSubmissionDataItem'           , 'id'               , @last_item_id, @batch_size;
    exec ct.purgeData_forceById 'WorkerActivityLog'               , 'run_id'           , @last_run_id , @batch_size;
    exec ct.purgeData_forceById 'WorkerDayActivity'               , 'runId'            , @last_run_id , @batch_size;
    exec ct.purgeData_forceById 'Package'                         , 'fileId'           , @last_file_id, @batch_size;
    exec ct.purgeData_forceById 'FILE'                            , 'id'               , @last_file_id, @batch_size;
    delete from ct.WorkerQualification where qualification_id in (select id from ct.CustomQualification cq where cq.TEST_STORE_ID <= @last_ds_id);
    exec ct.purgeData_forceById 'CustomQualification'             , 'TEST_STORE_ID'    , @last_ds_id  , @batch_size;
    exec ct.purgeData_forceById 'DATA_STORE'                      , 'ID'               , @last_ds_id  , @batch_size;
    exec ct.purgeData_forceById 'EndPointTaskAssignment'          , 'task_id'          , @last_et_id  , @batch_size;
    exec ct.purgeData_forceById 'EndpointTask'                    , 'id'               , @last_et_id  , @batch_size;
    exec ct.purgeData_forceById 'event_object'                    , 'event_tracking_id', @last_ev_id  , @batch_size;
    exec ct.purgeData_forceById 'event_tracking'                  , 'id'               , @last_ev_id  , @batch_size;
end try
begin catch
    throw;
end catch;

During maintenance

To reclaim free space, run the dbcc shrinkdatabase workfusion command.

Transactional purge

Out-of-the-box purge can also remove data related to a specific record or business transaction. Such an action can be requested by various regulators to remove the ID of a document or transaction from all systems.

Before you launch a transactional purge, make sure that the BP contains a step with activated tracking.

The operation finds out all hits or items by tracker_id and then clears all related tables in the ct and ws schemas.

Root tables:

  • ct.AwsHit
  • ws.hit (ws.hit.unique_request_token = ct.AwsHit.uuid)

Initial Key: AwsHit.uuid

CT tables to purge:

TableColumn
assignmentforhitjobhitid
awshitid, hitid
awshitassignmenthitid
awshiteventhit_id
awshitquestionhitid
awshitviewhistoryhitid
botrecordexecutionattempthitid
endpointtaskawshitid
event_trackinghitid
external_task_submissionhitid
hitdataitemloghit_id
hitdisabledforworkerhitid
hitrecordsourcehitid
hitsubmissiondataitemhitid
run_assignmentshitid
run_aws_hitshit_id
hit_uuid
submissionawshit_id
worker_messagehitid
workeractivityloghitid
workercommunicationhitid
workerfeedbackhitid
workerfitnesshistoryhitid

WorkSpace (WS) tables to purge:

TableColumn
assignmenthit_id
hitunique_request_token
hit_audithit_id
hit_historyhit_id
hit_type_preview_datahit_id
notification_eventhit_vid
notification_receptorhit_type_id
qualification_requirementhit_type_id

To run a transactional purge, use the following procedure:

View procedure 

CREATE PROCEDURE dbo.purge_tx_hit_sp
  @i_hit_uuid   nvarchar(36)
, @o_rec_cnt    bigint output
, @i_estimate   bit = 0
, @i_verbose    bit = 0
AS
begin
    set nocount on;
/***
    CHANGELOG: 10.1-rev-7
        2019-12-16 apouckatch reverse order
        2019-12-13 apouckatch created

    SAMPLE:
        declare @uuid   nvarchar(36) = '7641e603-6bac-450e-ac21-82998bf5a12f';
        declare @cnt    int;
        exec purge_tx_hit_sp @i_hit_uuid = @uuid, @o_rec_cnt = @cnt output;
        print formatMessage( 'CNT:%s', cast( @cnt as nvarchar(10))

***/
    declare cur_tables_to_purge cursor read_only forward_only for
    select
          schema_name
        , table_name
        , key_col_name
        , case sql_where_in
            when 'CT_HIT_ID'    then 'select ct_hit_id   from #tx_hits'
            when 'CT_HIT_UUID'  then 'select ct_hit_uuid from #tx_hits'
            when 'CT_ITEM_ID'   then 'select ct_item_id  from #tx_hits'
            when 'CT_ASMT_ID'   then 'select a.id        from ct.AwsHitAssignment a where a.hit_id in (select ct_hit_id from #tx_hits)'
            when 'CT_TRK_ID'    then 'select tracker_id  from #tx_tracker'
            when 'WS_HIT_ID'    then 'select ws_hit_id   from #tx_hits'
            when 'WS_ASMT_ID'   then 'select a.id        from ws.assignment a where a.hit_id in (select ws_hit_id from #tx_hits)'
            else sql_where_in
            end             as sql_where_id
    from ( values
              -- Control Tower:
                (      0, 'ct' , 'AwsHit'                       , 'id'              , 'CT_HIT_ID'  )
              , (      5, 'ct' , 'DataItemTracker'              , 'id'              , 'CT_TRK_ID'  )
              , (     10, 'ct' , 'AwsHitEvent'                  , 'hit_id'          , 'CT_HIT_ID'  )
              , (     20, 'ct' , 'AwsHitAssignment'             , 'hit_id'          , 'CT_HIT_ID'  )
              , (     30, 'ct' , 'AwsHitViewHistory'            , 'hitid'           , 'CT_HIT_ID'  )
              , (     40, 'ct' , 'AwsHitAssignmentAnswer'       , 'assignment_id'   , 'CT_ASMT_ID' )
              , (     41, 'ct' , 'AwsHitAssignmentAnswer'       , 'item_id'         , 'CT_ITEM_ID' )
              , (     50, 'ct' , 'HitSubmissionDataItem'        , 'hit_id'          , 'CT_HIT_ID'  )
              , (     60, 'ct' , 'HitSubmissionDataItemHistory' , 'item_id'         , 'select i.id from ct.HitSubmissionDataItem i where i.hit_id in (select ct_hit_id from #tx_hits)')
              , (     70, 'ct' , 'AwsHitQuestion'               , 'hit_id'          , 'CT_HIT_ID'  )
              , (     71, 'ct' , 'AwsHitQuestion'               , 'item_id'         , 'CT_ITEM_ID' )
              , (     80, 'ct' , 'HitDataItemLog'               , 'hit_id'          , 'CT_HIT_ID'  )
              , (     81, 'ct' , 'HitDataItemLog'               , 'item_id'         , 'CT_ITEM_ID' )
              , (     82, 'ct' , 'HitDataItemLog'               , 'assignment_id'   , 'CT_ASMT_ID' )
              , (     90, 'ct' , 'DataItemTrackerLog'           , 'data_item_id'    , 'CT_ITEM_ID' )
              , (     91, 'ct' , 'DataItemTrackerLog'           , 'tracker_id'      , 'CT_TRK_ID'  )
              , (    100, 'ct' , 'DataItemTrackerHitLink'       , 'data_item_id'    , 'CT_ITEM_ID' )
              , (    101, 'ct' , 'DataItemTrackerHitLink'       , 'tracker_id'      , 'CT_TRK_ID'  )
              , (    110, 'ct' , 'HitRecordSource'              , 'hitid'           , 'CT_HIT_ID'  )
              , (    120, 'ct' , 'WorkerActivityLog'            , 'hitid'           , 'CT_HIT_UUID')
              , (    130, 'ct' , 'run_aws_hits'                 , 'hit_id'          , 'CT_HIT_ID'  )
              -- WorkSpace:
              , (    500, 'ws' , 'hit'                          , 'id'              , 'WS_HIT_ID'  )
              , (    510, 'ws' , 'assignment'                   , 'hit_id'          , 'WS_HIT_ID'  )
              , (    520, 'ws' , 'hit_submission'               , 'assignment_id'   , 'WS_ASMT_ID' )
              , (    530, 'ws' , 'assignment_audit'             , 'assignment_id'   , 'WS_ASMT_ID' )
              , (    540, 'ws' , 'hit_history'                  , 'hit_id'          , 'WS_HIT_ID'  )
              , (    550, 'ws' , 'hit_type_preview_data'        , 'hit_id'          , 'WS_HIT_ID'  )
              , (    560, 'ws' , 'hit_audit'                    , 'hit_id'          , 'WS_HIT_ID'  )
         ) as t ( seq_id, schema_name, table_name               , key_col_name      , sql_where_in )
    order by seq_id desc   -- to track dependencies from details to master

    if @i_verbose = 1
        print 'Purging [hit_uuid:' + @i_hit_uuid + ']'

    set @o_rec_cnt = 0

    -- prepare a tracker:
    select trk.tracker_id
    into #tx_tracker
    from ct.awsHit                  as hit
    join ct.hitSubmissionDataItem   as item on hit.id  = item.hit_id
    join ct.dataItemTrackerHitLink  as trk  on item.id = trk.data_item_id
    where hit.uuid = @i_hit_uuid

    -- prepare a set of hits:
    select
          tx_ct_hits.id     as ct_hit_id
        , tx_ct_hits.uuid   as ct_hit_uuid
        , tx_items.id       as ct_item_id
        , tx_ws_hits.id     as ws_hit_id
    into #tx_hits
    from #tx_tracker                as trk
    join ct.dataItemTrackerHitLink  as tx_trk       on tx_trk.tracker_id = trk.tracker_id
    join ct.hitSubmissionDataItem   as tx_items     on tx_items.id       = tx_trk.data_item_id
    join ct.awsHit                  as tx_ct_hits   on tx_ct_hits.id     = tx_items.hit_id
    left join ws.hit                as tx_ws_hits   on tx_ct_hits.uuid   = tx_ws_hits.unique_request_token

    create nonclustered index idx_tx_hits_ct_hit_id   on #tx_hits( ct_hit_id  )
    create nonclustered index idx_tx_hits_ct_hit_uuid on #tx_hits( ct_hit_uuid)
    create nonclustered index idx_tx_hits_ct_item_id  on #tx_hits( ct_item_id )
    create nonclustered index idx_tx_hits_ws_hit_id   on #tx_hits( ws_hit_id  )

    -- loop tables:
    open cur_tables_to_purge

    while (1=1)
    begin
        declare @schema_name    nvarchar(128)
            ,   @table_name     nvarchar(128)
            ,   @key_col_name   nvarchar(128)
            ,   @sql_where_in   nvarchar(4000)
            ,   @sql_from_where nvarchar(1000)
            ,   @sql_execute    nvarchar(1000)
            ,   @start_date     datetime2
            ,   @end_date       datetime2
            ,   @exec_ms        bigint
            ,   @rec_cnt        bigint

        fetch cur_tables_to_purge
        into  @schema_name  , @table_name
            , @key_col_name , @sql_where_in

        if @@fetch_status != 0
            break

        set @start_date = getDate()

        set @sql_from_where = formatMessage
            ( 'from [%s].[%s] where [%s] in (%s)'
            , @schema_name, @table_name, @key_col_name, @sql_where_in
            )

        if @i_estimate = 1
        begin
            set @sql_execute = formatMessage( 'select @cnt = count(*) %s', @sql_from_where)
            exec sp_executeSql @sql_execute
                , N'@cnt int out'
                , @cnt = @rec_cnt output
        end
        else
        begin
            set @sql_execute = formatMessage('delete %s', @sql_from_where)
            exec sp_executeSql @sql_execute
            set @rec_cnt = @@rowcount
        end

        set @end_date = getDate()

        set @exec_ms = dateDiff( ms, @start_date, @end_date)

        set @o_rec_cnt = @o_rec_cnt + @rec_cnt

        if @i_verbose = 1
            print formatMessage
                ( 'DBG: [%3s].[%-30s][cnt: %7s][ms: %7s][sql:%s]'
                , @schema_name
                , @table_name
                , cast( @rec_cnt as nvarchar(7))
                , cast( @exec_ms as nvarchar(7))
                , @sql_execute
                )

    end -- while(1=1)

    close      cur_tables_to_purge
    deallocate cur_tables_to_purge

end

Purge Data Stores

Purging of Data Stores is required to avoid performance issues when executing Bot Tasks. To clean up Data Stores, create a separate Business Process and specify objects and conditions to trigger the procedure. It is recommended to include the cleanup step at the end of each Business Process.

For more information, see the Purge Data Stores guide.

Purge S3 buckets

You can purge the data in S3 buckets with MinIO configured on the installation server (APP or INT depending on your deployment). The operation uses the default Linux find / exec rm commands to delete the data. You can also specify patterns for deletion, for example, time of modification or creation, file extension, and so on.

To schedule Data Purge for S3 buckets, follow the steps below:

note

In the following instruction, /opt/workfusion/ is used as the default INSTALL_DIR installation directory.

  1. On the main installation server, in /opt/workfusion/supervisord/cron/cron.d/, create the minio-cleanup.sh Bash script with the following content:

    #!/bin/bash
# Set INSTALL_DIR according your installation
INSTALL_DIR=/opt/workfusion
BKT_LIST=`awk '{print $1}' $INSTALL_DIR/supervisord/conf/cron.d/minio_buckets.txt`

# Set desired execution time in HH:MM format. Change "00:00" accordingly. For the HA installation, set 1 hour interval between each Master host like 00:00, 01:00, 02:00
if [[ "$(date +%H:%M)" == "00:00" ]]; then
for val in $BKT_LIST
do

# Date print for log
echo "Cleaning $val at `date`" >> $INSTALL_DIR/supervisord/log/minio_cleanup.log

# Delete all PDF files older than 7 days
find $INSTALL_DIR/shared/minio/data/$val -iname "*.pdf" -mtime +7 -exec ls -lh {} >> $INSTALL_DIR/supervisord/log/minio_cleanup.log \; -exec rm -rf {} \;

# Delete all HTML files older than 7 days
find $INSTALL_DIR/shared/minio/data/$val -iname "*.html" -mtime +7 -exec ls -lh {} >> $INSTALL_DIR/supervisord/log/minio_cleanup.log \; -exec rm -rf {} \;
done
fi

  2. Set execution permissions for the script:

    chmod +x /opt/workfusion/supervisord/cron/cron.d/minio-cleanup.sh

  3. In /opt/workfusion/supervisord/conf/cron.d/, create the minio_buckets.txt file. In the file, specify the list of buckets to be cleared. Remember to add each bucket's name as a separate line, for example:

    doc-upload
custom-bucket-1
custom-bucket-2

  4. In /opt/workfusion/logrotate/logrotate.d/, create the minio-cleanup.conf logrotate configuration for the minio_cleanup.log file.

    # Ansible managed

/opt/workfusion/logs/minio_cleanup.log

{
    rotate 7
    copytruncate
    delaycompress
    compress
    notifempty
    missingok
    maxsize 200M
}

After that, the WorkFusion cron utility runs the MinIO purge automatically according to the specified schedule.

Purge Analytics

The Analytics component features a customizable Data Purge procedure that cleans up data from Control Tower, Elasticsearch, RPA, Data Stores, and other components that you can add. Unlike the CT scheduler, the procedure is run manually. As you run the purge, it the delete process handles data table by table using specific settings (dp_config) for each of them.

warning

Running Data Purge can lead to an increase in the transaction log size and low disk space.

Before starting the Analytics Data Purge, make sure you have stopped all ETL processes:

  1. Stop the ETL scheduler:

    1. On the Analytics server, go to the project that has the same name as the site (in this example, test).

    2. Open the AA_ETL_Scheduler workbook.

    3. Go to the Refresh Schedules tab. The tab contains the names of all schedules if they exist.

    4. In the list, select AA_ETL_Scheduler, and go to the Details tab.

    5. Set the status to Disabled:

    After the Data Purge, remember to enable the schedule for data upgrade.

  2. Run the ETL process manually using SSMS and ensure it finishes successfully. The script must run without errors.

    exec [etl#process_sp]

    The following script returns 0 rows if there are no errors in the last ETL execution:

    select id, duration_sec , source_name, target_name, rows_number, notes
from etl_log
where notes <> 'ok' and
group_date =  (select max(group_date) from etl_log)

Levels

Analytics data contains some levels according to the solution architecture:

  • source: pm.* tables
  • ods: the middle level between the source one and wh storages
  • wh: the dashboard creation level

Data Purge cleans up all these levels at a time.

Tables

dp_config

This table stores the settings for data purging per table. It should have at least one row for Data Purge to run correctly with the is_taken = 1.

SchemaTableDescription
pmmetricSource table with system metrics, such as CPU, disk, and memory usage, for all components
pmbep_worker_metricsSource table with the processing results of Bot Execution Platform (BEP) Agent tasks
pmbep_worker_resourcesSource table with information about BEP Agent resource consumption
dmpm_bep_worker_metricTransitional table (between the source and data warehouse) with the processing results of BEP Agent tasks
dmpm_bep_worker_resourceTransitional table with information about BEP Agent resource consumption
dmct_task_instanceTransitional table with information about tasks per each Business Process step
dmct_task_executionTransitional table with information about each assignment in the task for workers
dmct_task_itemTransitional table with information about each input record
dmwh_bep_agentData warehouse table with information about BEP Agent resource consumption
dmwh_bep_metricData warehouse table with the processing results of BEP Agent tasks
dmwh_component_metricData warehouse table with system metrics, such as CPU, disk, and memory usage for all components
dmwh_transaction_itemData warehouse table with information about each input record or document in a BP
dmwh_transaction_e2eData warehouse table with aggregated statistics about each input record or document in a BP
  • id: increment bigint.
  • sp_name: the nvarchar(255) name of the stored procedure to delete rows in a table.
  • table_name: the nvarchar(255) name of the table.
  • date_field: the nvarchar(255) name of the datetime field in the table_name table. Used when data deletion is based on a time range.
  • id_field: the name of the identity column in the temp table. It is used when data deletion is based on IDs.
  • group_name: the name of the group to which the data belongs (pm, rpa, ct, ml, src_pm).
  • layer: ods, wh, or src.
  • is_taken: the flag to define whether the table is taken next time.
  • batch_size: the count of rows to delete at a time. The default value is 100000. Limiting the rows is required for performance improvement. The count is to be investigated and depends on the size and load of your database.
  • sleep_time: delay time in milliseconds before the next step begins. The default value is 0.
  • prev_id: the ID of the previous step. If null, it's the first step in a sequence.

dp_log

The table stores the set of data for a purge run:

  • id: increment bigint.
  • sp_name: the nvarchar(255) name of the stored procedure.
  • table_name: the nvarchar(255) name of the table.
  • date_field: the nvarchar(255) name of the datetime field in the table_name table.
  • purge_run_id: the increment integer to count purge sets. Each execution of the master procedure adds 1 to previous purge_run_id.
  • rows_cnt: the number of rows deleted per table.
  • duration_sec: the duration in seconds of the Data Purge process per table.
  • start_date: the date and time when the process started.
  • end_date: the date and time when the process ended.
  • status: ok or an error message if something went wrong.

Procedure and parameters

The stored master purge_data_analytics_sp procedure is the primary one and starts the purge data process. The procedure is to be run manually in SSMS.

The default parameters are 30 days back from now.

For the procedure, set the following:

  1. Define the period for the data to be available in Analytics DB. You can set up the period type (day, hour, minute) as well as the period value or count.
  2. Define the batch size (count of rows to delete at a time) and sleep time between batch purges.
--example to run
declare @cnt int

exec @cnt = purge_data_analytics_sp 'day',30

[purge_data_analytics_sp] 

create or alter procedure purge_data_analytics_sp
@metric_type nvarchar(10) = 'day' -- day/hour/minute - type of range
, @metric_value int = 30 -- value determining how far to go back to delete rows, goes in pair with @metric_type
as
/********
SAMPLE:
exec purge_data_analytics_sp 'day',30

CHANGELOG: 10.1-rev
2019-12-12 added [hpath] to order set of tables
2019-08-02 created
********/

begin
set nocount on
; /*parameters that define the execution sql command string*/
declare @sp_name nvarchar(255) -- the SP name to purge the table
declare @tb_name nvarchar(255) -- the table name where rows will be deleted
declare @fl_name nvarchar(255) -- the name of the field (datetime2) used to calculate the range for deleting if needed
declare @id_name nvarchar(255) -- used for deleting in ct_* tables to identify a specific id/table
declare @group nvarchar(8) -- pm,rpa,ct
declare @batch_size int -- the size of the batch to delete (in rows)
declare @sleep_time float --the  sleep time to delay the next step
/*parameters that store information to put into the log table*/
declare @start_date datetime2 -- the start datetime of each deleting process
declare @end_date datetime2 -- the end datetime of each deleting process
declare @duration_sec bigint -- the duration in seconds, the difference between @start_date and @end_date
declare @rows_cnt bigint -- the count of rows that were deleted
declare @status nvarchar(max) -- 'ok' or error message as a result
declare @purge_run_id int -- the parameter to distinguish different execs of master SP
/*inner parameters*/
declare @sql nvarchar(max) -- the sql statement to run the next-level SP to delete rows
declare @date_from datetime2 -- the upper limit for deleting process_execution
declare @temp_table nvarchar(32) --

set @purge_run_id = (select coalesce(max(purge_run_id),0) + 1 from [dp_log]) -- each execution of master SP will be marked
;
if object_id('tempdb..##dp_ct_ids',N'U') is not null drop table ##dp_ct_ids
--table to collect all ids in ct_* tables to delete
create table ##dp_ct_ids
( process_execution_id bigint not null,
[start_date] datetime2(7),
task_id bigint not null,
task_instance_id bigint not null,
task_item_id bigint not null,
task_execution_id bigint null,
task_transaction_id bigint null,
plugin_id bigint null)
;
if object_id('tempdb..##dp_rpa_sessions_ids',N'U') is not null drop table ##dp_rpa_sessions_ids
create table ##dp_rpa_sessions_ids
(bot_session_id bigint not null)
;
if object_id('tempdb..##dp_rpa_status_ids',N'U') is not null drop table ##dp_rpa_status_ids
create table ##dp_rpa_status_ids
(bot_status_id bigint not null)
;
if object_id('tempdb..##dp_ml_stat_ids',N'U') is not null drop table ##dp_ml_stat_ids
create table ##dp_ml_stat_ids
(mls_id bigint not null)
;

-- get the date starting from which the process execution data is stored (or the upper limit for deleting)
set @date_from = case @metric_type when 'day' then dateadd(day,-@metric_value,getdate())
when 'hour' then dateadd(hour,-@metric_value,getdate())
when 'minute' then dateadd(minute,-@metric_value,getdate())
end
; -- collect all ids for ct_*,rpa_*,ml_* tables
exec full_dp_ids_sp @date_from

-- cursor inside which the processq is running on the [dp_config] table and deletes or purges data for a specific table with exact parameters - steps
declare cur_purge_sp cursor for
with D_SRC as
(
select id
, isnull(prev_id,0) prev_id
from dp_config
)
, D_TREE( id, prev_id, hpath) as
(
select s.id
, s.prev_id
, cast(s.id as nvarchar(4000)) as hpath
from D_SRC s
where s.prev_id = 0
union all
select s.id
, s.prev_id
, concat( t.hpath, '/', s.id) as hpath
from D_SRC s
inner join D_TREE t on s.prev_id = t.id
)
select
[sp_name]
, [table_name]
, [date_field]
, [id_field]
, case [group_name] when 'ct' then '##dp_ct_ids'
when 'ml' then '##dp_ml_stat_ids'

when 'rpa' then case [id_field] when 'bot_status_id' then '##dp_rpa_status_ids' else '##dp_rpa_sessions_ids' end
else ''
end as temp_table
, [group_name]
, [batch_size]
, [sleep_time]
from [dp_config] dp
join D_TREE dt on dp.id = dt.id
where [is_taken] = 1
order by dt.hpath

open cur_purge_sp

while (1=1) --while the last row is read
begin
fetch next from cur_purge_sp
into @sp_name, @tb_name, @fl_name, @id_name, @temp_table, @group, @batch_size, @sleep_time

if @@fetch_status != 0
break

set @start_date = getdate() --define the start of the step

begin try
begin tran

; -- form sql statement
set @sql = N'exec @row_cnt = ' + @sp_name + ' @tb_name, @fl_name,@id_name, @temp_table, @group, @date_from, @batch_size, @sleep_time '
;
exec sp_executesql @sql, N'@tb_name nvarchar(255), @fl_name nvarchar(255), @id_name nvarchar(255), @temp_table nvarchar(32), @group nvarchar(8), @date_from datetime2, @batch_size int, @sleep_time float, @row_cnt int output',
@tb_name = @tb_name
, @fl_name = @fl_name
, @date_from = @date_from
, @id_name = @id_name
, @temp_table = @temp_table
, @group = @group
, @batch_size = @batch_size
, @sleep_time = @sleep_time
, @row_cnt = @rows_cnt output

commit tran
set @status = 'ok'

end try
begin catch --if error
if @@trancount <> 0
rollback transaction
;
declare @errormessage nvarchar(4000) = error_message()
declare @errorseverity int = error_severity()
declare @errorstate int = error_state()
-- save status
set @status = formatmessage( '[fail] %i: %s', error_number(), @errormessage)
;
raiserror(@errormessage, @errorseverity, @errorstate)
;
end catch

set @end_date = getdate() --define the end of the step
set @duration_sec = datediff(s, @start_date, @end_date)
-- fill in the log table
insert into [dp_log] ([sp_name], [table_name], [date_field], [purge_run_id], [rows_cnt], [duration_sec], [start_date], [end_date], [status])
values (@sp_name, @tb_name, @fl_name, @purge_run_id, @rows_cnt, @duration_sec, @start_date, @end_date,@status)
;
end
-- close cursor
close cur_purge_sp
deallocate cur_purge_sp
-- clean up the temp table with ct_* ids
if object_id('tempdb..##dp_ct_ids',N'U') is not null drop table ##dp_ct_ids
if object_id('tempdb..##dp_rpa_sessions_ids',N'U') is not null drop table ##dp_rpa_sessions_ids
if object_id('tempdb..##dp_rpa_status_ids',N'U') is not null drop table ##dp_rpa_status_ids
if object_id('tempdb..##dp_ml_stat_ids',N'U') is not null drop table ##dp_ml_stat_ids

end

Additional reading

For information about other means to optimize Business Process data, refer to the article.

← PreviousConfigure and run Data Purge →