Posted on by Dengineer PC

The following custom stored proc (c_physical_delete_processing)
works by looking at a field named “dss_deleted_flag” in the  data vault, which equals “N”, and if it can’t find the record in the stage table, it adds the record from the vault table with a dss_deleted_flag = ‘Y’ to the stage table. The stage table would have this field as well, set to default as dss_deleted_flag = ‘N’,

So at the end of the stored proc, you would have the rows which do exist in the source table as “N” and the ones that do not exist as “Y”

 

This works for data vault satellites, and the naming convention we used for data vault stage tables is starting the name with “stg_dv_“,

so a satellite table for example would be s_test_table, and its stage would be stg_sv_s_test_table.

The stored proc would be placed into a a single job in between the stage and satellite tables. E.G the order of job would be

  1. stg_sv_s_test_table
  2. c_physical_delete_processing
  3. s_test_table

 

Code below

Transact-SQL
0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
 
CREATE PROCEDURE c_physical_delete_processing
  @p_sequence         integer
, @p_job_name         varchar(256)
, @p_task_name        varchar(256)
, @p_job_id           integer
, @p_task_id          integer
, @p_return_msg       varchar(256) OUTPUT
, @p_status           integer      OUTPUT
AS
  SET XACT_ABORT OFF  -- Turn off auto abort on errors
 
  SET NOCOUNT ON      -- Turn off row count messages
 
  --===============================================================
  -- Control variables used in most programs
  --===============================================================
  DECLARE
    @v_msgtext         varchar(256) -- Text for audit_trail
  , @v_step            integer      -- return code
  , @v_insert_count    integer      -- no of records inserted
  , @v_count           integer      -- General counter
  , @v_return_status   integer      -- Query result status
  , @v_db_code         varchar(10)  -- Database error code
  , @v_db_msg          varchar(100) -- Database error message
 
  DECLARE @v_work_tbl TABLE(
                        stg_tbl_name VARCHAR(64),
                        satellite_tbl_name VARCHAR(64),
                        SQL_str VARCHAR(MAX)
                      );
  DECLARE @v_work_tbl2 TABLE(
                        stg_tbl_name VARCHAR(64),
                        satellite_tbl_name VARCHAR(64),
                        SQL_str VARCHAR(MAX)
                      );
  DECLARE @v_7_tbl TABLE(
                        stg_tbl_name VARCHAR(64),
                        EDW_tbl_name VARCHAR(64),
str_1 VARCHAR(max),
str_2 VARCHAR(max),
str_3 VARCHAR(max)
                      );
 
 
  DECLARE
    @v_stg_tbl_name VARCHAR(64),
    @v_satellite_tbl_name VARCHAR(64),
    @v_SQL_str NVARCHAR(MAX);
  --===============================================================
  -- Main
  --===============================================================
  BEGIN TRY
 
    SET @v_step = 100;
    SET @v_insert_count = 0;
 
    SET @v_step = 200;
 
 
 
 
 
WITH
  v_1 AS (
    SELECT DISTINCT
      wTC_obj_key stg_tbl_obj_key,
      wTC_name stg_tbl_name,
      RIGHT(wTC_name, LEN(wTC_name) - LEN('stg_dv_')) EDW_tbl_name
      FROM WS_wrk_task_ctrl
      WHERE wTC_job_key = @p_job_id
        AND wTC_name LIKE 'stg\_dv\_%'
                       ESCAPE '\'
        AND EXISTS (
              SELECT NULL
                FROM WS_stage_col
                WHERE sC_obj_key = WS_wrk_task_ctrl.wTC_obj_key
                  AND sC_col_name = 'dss_deleted_flag'
            )
  ),
  v_2 AS (
    SELECT
      v_1.stg_tbl_name,
      v_1.EDW_tbl_name,
      WS_stage_col.sC_col_name column_name,
      ROW_NUMBER()
        OVER (
          PARTITION BY v_1.stg_tbl_name
          ORDER BY WS_stage_col.sC_order
        ) column_ordinal_num
      FROM
        v_1
        JOIN WS_stage_col
          ON v_1.stg_tbl_obj_key = WS_stage_col.sC_obj_key
         AND WS_stage_col.sC_col_name != 'dss_updated_date'
  ),
  v_3 AS (
    SELECT
      v_2.stg_tbl_name,
      v_2.EDW_tbl_name,
      v_2.column_name,
      v_2.column_ordinal_num,
      CASE WS_normal_col.nC_key_type
        WHEN 'A'
          THEN 'Y'
        ELSE 'N'
      END business_key_flag
      FROM
        v_2
        LEFT JOIN WS_normal_tab
          ON v_2.EDW_tbl_name = WS_normal_tab.nT_table_name
        LEFT JOIN WS_normal_col
          ON WS_normal_tab.nT_obj_key = WS_normal_col.nC_obj_key
         AND v_2.column_name = WS_normal_col.nC_col_name
         AND WS_normal_col.nC_key_type = 'A'
  ),
  v_7 AS (
    SELECT
      stg_tbl_name,
      EDW_tbl_name,
      (SELECT column_name + ', '
         FROM v_3 v_4
         WHERE stg_tbl_name = v_3.stg_tbl_name
         ORDER BY column_ordinal_num
         FOR XML PATH('')
      ) str_1,
      (SELECT CASE column_name
                WHEN 'cnt'
                  THEN '1/**/ cnt'
                WHEN 'row_num'
                  THEN '1/**/ row_num'
                WHEN 'dss_deleted_flag'
                  THEN '''Y'' dss_deleted_flag'
                ELSE column_name
              END + ', '
         FROM v_3 v_5
         WHERE stg_tbl_name = v_3.stg_tbl_name
         ORDER BY column_ordinal_num
         FOR XML PATH('')
      ) str_2,
      (SELECT column_name + ' = ' + EDW_tbl_name + '.' + column_name + ' AND '
         FROM v_3 v_6
         WHERE stg_tbl_name = v_3.stg_tbl_name
           AND business_key_flag = 'Y'
         ORDER BY column_ordinal_num
         FOR XML PATH('')
      ) str_3
      FROM v_3
      GROUP BY stg_tbl_name, EDW_tbl_name
  )
 
-----------------------------------------------------------------
-- add tables to temporary table used in two sets of logic below.
 
  INSERT
    INTO @v_7_tbl
    SELECT
      stg_tbl_name,
      EDW_tbl_name,
  str_1,
  str_2,
  str_3
      FROM v_7
       WHERE CHARINDEX('ftcacc_transaction',EDW_tbl_name) = 0;
 
 
-----------------------------------------------------------------
-- Insert rows into stage table for rows no longer in load
 
 
  INSERT
    INTO @v_work_tbl
    SELECT
      stg_tbl_name,
      EDW_tbl_name,
      'INSERT ' +
        'INTO '+ stg_tbl_name + ' WITH (TABLOCK) (' +
                   str_1 +
                   'dss_updated_date' +
                ') ' +
        'SELECT ' +
           str_2 +
          'GETDATE() dss_updated_date ' +
          'FROM ' + EDW_tbl_name +
         ' WHERE dss_current_flag = ''Y'' ' +
            'AND dss_deleted_flag = ''N'' ' +
            'AND NOT EXISTS (' +
                  'SELECT NULL ' +
                    'FROM ' + COALESCE(
                                (SELECT sT_table_name
                                   FROM WS_stage_tab
                                   WHERE sT_table_name =
                                         stg_tbl_name + '_bk_only'
                                ),
                                stg_tbl_name
                              ) +
                   ' WHERE ' + LEFT(str_3, LEN(str_3) - LEN(' AND ')) +
                ')' SQL_str
      FROM @v_7_tbl;
 
    SET @v_step = 300;
 
    WHILE (SELECT COUNT(*) cnt
             FROM @v_work_tbl
          ) > 0
    BEGIN
      SELECT TOP 1
        @v_stg_tbl_name = stg_tbl_name,
        @v_satellite_tbl_name = satellite_tbl_name,
        @v_SQL_str = SQL_str
        FROM @v_work_tbl
        ORDER BY 1;
      SET @v_msgtext = 'Inserting into ' + UPPER(@v_stg_tbl_name) + ' from ' +
                       UPPER(@v_satellite_tbl_name) + '...';
      EXEC @v_return_status = WSWrkAudit
                                'I',
                                @p_job_name,
                                @p_task_name,
                                @p_sequence,
                                @v_msgtext,
                                NULL,
                                NULL,
                                @p_task_id,
                                @p_job_id;
 
      EXEC(@v_SQL_str);
 
      SET @v_count = @@ROWCOUNT;
      SET @v_msgtext = '@V_COUNT: ' + CAST(@v_count AS VARCHAR);
      SET @v_insert_count = @v_insert_count + @v_count;
      EXEC @v_return_status = WSWrkAudit
                                'I',
                                @p_job_name,
                                @p_task_name,
                                @p_sequence,
                                @v_msgtext,
                                NULL,
                                NULL,
                                @p_task_id,
                                @p_job_id;
      DELETE @v_work_tbl
        WHERE stg_tbl_name = @v_stg_tbl_name;
    END
 
 
 
-----------------------------------------------------------------
-- Update dss_deleted_rows ino stage table where column called 'row' is null
 
 
 
INSERT
    INTO @v_work_tbl2
    SELECT
      stg_tbl_name,
      EDW_tbl_name,
  'update '+ stg_tbl_name + ' set row=1 where row is null AND dss_deleted_flag = ''Y'' '
      FROM @v_7_tbl temp
   JOIN  WS_stage_tab
on WS_stage_tab.st_table_name= temp.stg_tbl_name
JOIN WS_stage_col
  on sc_obj_key = st_obj_key
  and sc_col_name = 'row';
 
 
    SET @v_step = 400;
 
    WHILE (SELECT COUNT(*) cnt
             FROM @v_work_tbl2
          ) > 0
    BEGIN
      SELECT TOP 1
        @v_stg_tbl_name = stg_tbl_name,
        @v_satellite_tbl_name = satellite_tbl_name,
        @v_SQL_str = SQL_str
        FROM @v_work_tbl2
        ORDER BY 1;
      SET @v_msgtext = 'Updating ' + UPPER(@v_stg_tbl_name) + ' from ' +
                       UPPER(@v_satellite_tbl_name) + '...';
 
      EXEC(@v_SQL_str);
 
      SET @v_count = @@ROWCOUNT;
      SET @v_msgtext = '@V_COUNT: ' + CAST(@v_count AS VARCHAR);
      SET @v_insert_count = @v_insert_count + @v_count;
 
      DELETE @v_work_tbl2
        WHERE stg_tbl_name = @v_stg_tbl_name;
    END
 
 
 
 
 
  -----------------------------------------------------------------
-- Finish processing
 
    SET @v_step = 500;
    SET @p_return_msg = LEFT(
                          'C_PHYSICAL_DELETE_PROCESSING completed. ' +
                          'Rows inserted: ' + CAST(
                                                @v_insert_count
                                                AS VARCHAR
                                              ),
                          256
                        );
    SET @p_status = 1;
 
  END TRY
  BEGIN CATCH
    SET @v_DB_code = CAST(/**/ERROR_NUMBER() AS VARCHAR);
    SET @v_DB_msg = /**/ERROR_MESSAGE();
    SET @p_status =/**/-3;
    SET @p_return_msg = LEFT(
                          'C_PHYSICAL_DELETE_PROCESSING FAILED' +
                        '. Step num.: ' + CAST(@v_step AS VARCHAR) +
                        '. Err. num.: ' + @v_DB_code +
                        '. Err. msg.: ''' + @v_DB_msg + '''',
                          256
                        );
    EXEC @v_return_status = WSWrkError
                              'E',
                              @p_job_name,
                              @p_task_name,
                              @p_sequence,
                              @p_return_msg,
                              @v_DB_code,
                              @v_DB_msg,
                              @p_task_ID,
                              @p_job_ID,
                              NULL;
  END CATCH;
 
  RETURN 0
 

 

Share This Via: