· 8 years ago · Dec 20, 2017, 11:44 AM
1-- CFG_DWH_IMPORT
2
3DECLARE @sql varchar(max);
4DECLARE @ColumnName as varchar(150)
5DECLARE @ColumnNameNew as varchar(150)
6DECLARE @IsPrimaryKey as BIT
7DECLARE @SourceSchema as varchar(150)
8DECLARE @SourceTable as varchar(150)
9DECLARE @TargetSchema as varchar(150)
10DECLARE @TargetTable as varchar(150)
11DECLARE @insertCount int
12DECLARE @updateCount int
13DECLARE @deleteCount int
14
15
16/*** Logeintrag ***/
17INSERT INTO CFG_LOG.dbo.LogTable
18 ([DateTime]
19 ,[Funktion]
20 ,[Text]
21 ,[Status])
22 SELECT SYSDATETIME() as ZEIT
23 ,'Merge'
24 ,'Der Merge wurde gestartet'
25 ,'INFO' as Status ;
26
27
28/*** Config Cursor erstellen **************************
29 Configtabelle aller Daten die ins DWH kommen
30*******************************************************/
31DECLARE cur_cfg_dwh_import CURSOR READ_ONLY FOR
32 SELECT SourceSchema
33 ,SourceTable
34 ,TargetSchema
35 ,TargetTable
36 FROM cfg_log.dbo.CFG_DWH_IMPORT WHERE upper([UPDATE]) = 'J'
37 OPEN cur_cfg_dwh_import
38 FETCH NEXT FROM cur_cfg_dwh_import INTO @SourceSchema
39 ,@SourceTable
40 ,@TargetSchema
41 ,@TargetTable
42
43
44WHILE (@@FETCH_STATUS <> -1)
45BEGIN
46 DECLARE @ViewDefinition varchar(max) = ''
47
48 select @ViewDefinition=rtrim(ltrim(m.definition))
49 from STA.sys.objects o
50 inner join STA.sys.sql_modules m on m.object_id = o.object_id
51 inner join STA.sys.views v on v.object_id = o.object_id
52 where v.name = @SourceTable
53
54 DECLARE @SEARCHPATTERN AS VARCHAR(MAX)
55 DECLARE @STARTPOS INT
56 DECLARE @LEN INT
57
58 -- --------------------------- Entfernen mehrzeiliger Kommentare ---------------------------
59 SET @SEARCHPATTERN='/*'
60 declare @ViewDefinitionNew as varchar(max) = @ViewDefinition
61 declare @Positions table (StartPos int )
62 Declare @pos int
63 Declare @oldpos int
64 Select @oldpos=0
65 select @pos=charindex(@SEARCHPATTERN,@ViewDefinition)
66 while @pos > 0 and @oldpos<>@pos
67 begin
68 insert into @Positions Values (@pos)
69 Select @oldpos=@pos
70 select @pos=CHARINDEX('*/',@ViewDefinition,@pos+1)
71
72 SET @ViewDefinitionNew = REPLACE(@ViewDefinitionNew, substring(@ViewDefinition, @oldpos, @pos), '')
73 end
74
75 SET @ViewDefinition = @ViewDefinitionNew
76
77 -- --------------------------- Entfernen einzeiliger Kommentare ---------------------------
78
79 DELETE FROM @Positions
80 SET @SEARCHPATTERN='--'
81 Select @oldpos=0
82 select @pos=charindex(@SEARCHPATTERN,@ViewDefinition)
83 while @pos > 0 and @oldpos<>@pos
84 begin
85 insert into @Positions Values (@pos)
86 Select @oldpos=@pos
87 select @pos=CHARINDEX(@SEARCHPATTERN,@ViewDefinition,@pos+1)
88 end
89
90 DECLARE @comment varchar(max)
91 DECLARE cur_replace_comments CURSOR READ_ONLY FOR
92 select SUBSTRING(@ViewDefinition, StartPos, CHARINDEX(CHAR(13), @ViewDefinition, StartPos)-StartPos) as ToBeRemoved from @Positions
93 OPEN cur_replace_comments
94 FETCH NEXT FROM cur_replace_comments INTO @comment
95
96 WHILE (@@FETCH_STATUS <> -1)
97 BEGIN
98 SET @ViewDefinition = REPLACE(@ViewDefinition, @comment, '')
99
100 FETCH NEXT FROM cur_replace_comments INTO @comment
101 END
102
103 CLOSE cur_replace_comments
104 DEALLOCATE cur_replace_comments
105
106 -- --------------------------- CLEAN STATEMENT ---------------------------
107 -- Replace the following occurencies
108 --NULL
109 --Horizontal Tab
110 --Line Feed
111 --Vertical Tab
112 --Form Feed
113 --Carriage Return
114 --Column Break
115 --Non-breaking space
116 --TAB
117 SET @ViewDefinition = RTRIM(LTRIM(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(@ViewDefinition, CHAR(0), ' '), CHAR(9), ' '), CHAR(10), ' '), CHAR(11), ' '), CHAR(12), ' '), CHAR(13), ' '), CHAR(14), ' '), CHAR(160), ' '), ' ', ' ')))
118
119
120 DECLARE @IgnoreCrossApply NVARCHAR(MAX) = @ViewDefinition
121 IF (CHARINDEX('CROSS APPLY', @IgnoreCrossApply) > 0)
122 BEGIN
123 SET @IgnoreCrossApply = SUBSTRING(@IgnoreCrossApply, 0, CHARINDEX('CROSS APPLY', @IgnoreCrossApply))
124 END
125
126 -- Rauslösen der StartPosition der Columns, welche mit AS benannt werden
127 DELETE FROM @Positions
128 SET @SEARCHPATTERN='% as %'
129 SET @LEN = Len(REPLACE(@SEARCHPATTERN, '%', ''))
130 Select @oldpos=0
131 select @pos=patindex(@SEARCHPATTERN,@IgnoreCrossApply)
132 while @pos > 0 and @oldpos<>@pos
133 begin
134 insert into @Positions Values (@pos+@LEN)
135 Select @oldpos=@pos
136 select @pos=patindex(@SEARCHPATTERN,Substring(@IgnoreCrossApply,@pos + 1,len(@IgnoreCrossApply))) + @pos
137 end
138
139 -- Speichern der Columns, welche mit AS benannt werden
140 SET @SEARCHPATTERN=' '
141
142 select StartPos, CHARINDEX(@SEARCHPATTERN,@ViewDefinition,STARTPOS+1) as EndPos, LTRIM(SUBSTRING(@ViewDefinition,STARTPOS,CHARINDEX(@SEARCHPATTERN,@ViewDefinition,STARTPOS+1)-STARTPOS)) as DesiredName,
143 -- Rauslösen und speichern der originalen Spaltennamen, welche mit ' ' oder ',' beginnen
144 -- Länge des originalen Feldes über "Reverse-String" und der Suche nach Blank finden +
145 -- entfernen von ,
146 -- entfernen von quotes
147 -- entfernen von whitespace
148 rtrim(ltrim((Replace(REPLACE(SUBSTRING(@ViewDefinition, StartPos-
149 CHARINDEX(' ', SUBSTRING(substring(REVERSE(@ViewDefinition), CHARINDEX(' ',REVERSE(@ViewDefinition),LEN(@ViewDefinition)-STARTPOS), STARTPOS), 5, LEN(substring(REVERSE(@ViewDefinition), CHARINDEX(' ',REVERSE(@ViewDefinition),LEN(@ViewDefinition)-STARTPOS), STARTPOS)) -5))-2
150 , CHARINDEX(' ', SUBSTRING(substring(REVERSE(@ViewDefinition), CHARINDEX(' ',REVERSE(@ViewDefinition),LEN(@ViewDefinition)-STARTPOS), STARTPOS), 5, LEN(substring(REVERSE(@ViewDefinition), CHARINDEX(' ',REVERSE(@ViewDefinition),LEN(@ViewDefinition)-STARTPOS), STARTPOS)) -5))), ',', ''), '''',''))))
151 as OriginalName
152 into #Parts
153 from @Positions
154 -- ACHTUNG: Folgendes Statement wird später nicht mehr benötigt
155 -- Ist nur zur Identifikation wie folgt: 'ID_Kundennummer' AS ID_Kundennummer, 'ID_Mandant' AS ID_Mandant
156 where LEFT(REPLACE(REPLACE(LTRIM(SUBSTRING(@ViewDefinition,STARTPOS,CHARINDEX(@SEARCHPATTERN,@ViewDefinition,STARTPOS+1)-STARTPOS)),'[',''),']',''),3) != 'ID_' and
157 LTRIM(SUBSTRING(@ViewDefinition,STARTPOS,CHARINDEX(@SEARCHPATTERN,@ViewDefinition,STARTPOS+1)-STARTPOS)) != '('
158
159 -- Daten bereinigen bei den umbenannten Spalten in den Views
160 -- Löschen von ,
161 update #Parts
162 set DesiredName=SUBSTRING(desiredName, 0, len(desiredName)-1)
163 where right(DesiredName,1)=','
164
165 delete from #Parts
166 where rtrim(ltrim(DesiredName)) ='' and RTRIM(Ltrim(originalName))=''
167
168 update #Parts
169 set DesiredName=REPLACE(REPLACE(DesiredName, '[', ''), ']', ''),
170 OriginalName=REPLACE(REPLACE(OriginalName, '[', ''), ']', '')
171
172 update #Parts
173 set OriginalName = substring(replace(replace(replace(ltrim(SUBSTRING(@ViewDefinition, startpos - charindex(',', SUBSTRING(reverse(@ViewDefinition), len(@ViewDefinition)-StartPos, EndPos))+2, StartPos)),',', ''),'[',''),']',''), 0, CHARINDEX(' as', replace(replace(replace(ltrim(SUBSTRING(@ViewDefinition, startpos - charindex(',', SUBSTRING(reverse(@ViewDefinition), len(@ViewDefinition)-StartPos, EndPos))+2, StartPos)),',', ''),'[',''),']','')))
174 where OriginalName = ''
175
176 --select * from #Parts
177
178 -- --------------------------- Einlesen der Tabellennamen ---------------------------
179
180 -- Suchen nach Join(s), From
181 DECLARE @Tabellennamen TABLE (name nvarchar(max))
182 DECLARE @Fields VARCHAR(MAX) = SUBSTRING(@ViewDefinition, 0, CHARINDEX('FROM ', @ViewDefinition))
183
184 -- Alles vor 'AS' wegschneiden
185 SET @SEARCHPATTERN='AS'
186 SET @STARTPOS = CHARINDEX(UPPER(@SEARCHPATTERN), UPPER(@Fields))
187 SELECT @Fields = SUBSTRING(@Fields, @STARTPOS, LEN(@Fields)-@STARTPOS+1)
188
189 -- Alles vor SELECT wegschneiden
190 SET @Fields = substring(@Fields,charindex(N'SELECT', UPPER(@Fields)),len(@Fields)-charindex(N'SELECT', UPPER(@Fields))+1)
191
192 SET @ViewDefinitionNew=@ViewDefinition
193 -- Verkeiner des Suchstrings um die Last zu senken
194 -- alles vor '... FROM ' wegschneiden
195 select @ViewDefinition=RTRIM(LTRIM(SUBSTRING(@ViewDefinition, CHARINDEX('FROM ', @ViewDefinition)+5, LEN(@ViewDefinition)-CHARINDEX('FROM ', @ViewDefinition)-4)))
196
197 -- FROM CLAUSE
198 INSERT INTO @Tabellennamen (name) VALUES (case when CHARINDEX('STA', case when CHARINDEX(' ', @ViewDefinition) = 0 then ltrim(replace(replace(@ViewDefinition, '[', ''), ']', '')) else ltrim(replace(replace(SUBSTRING(@ViewDefinition, 0, CHARINDEX(' ', @ViewDefinition)), '[', ''), ']', '')) end) = 1 then '' else 'STA.' end + case when CHARINDEX(' ', @ViewDefinition) = 0 then ltrim(replace(replace(@ViewDefinition, '[', ''), ']', '')) else ltrim(replace(replace(SUBSTRING(@ViewDefinition, 0, CHARINDEX(' ', @ViewDefinition)), '[', ''), ']', '')) end)
199
200 -- JOIN(s)
201 DELETE FROM @Positions
202 SET @SEARCHPATTERN='%JOIN%'
203 SET @LEN = Len(REPLACE(@SEARCHPATTERN, '%', ''))
204 Select @oldpos=0
205 select @pos=patindex(@SEARCHPATTERN,@ViewDefinition)
206 while @pos > 0 and @oldpos<>@pos
207 begin
208 insert into @Positions Values (@pos+@LEN)
209 Select @oldpos=@pos
210 select @pos=patindex(@SEARCHPATTERN,Substring(@ViewDefinition,@pos + 1,len(@ViewDefinition))) + @pos
211 end
212
213 Select REPLACE(t.name, ')','') as 'Name'
214 into
215 #Tabellen
216 from
217 (Select REPLACE(case when CHARINDEX('STA', replace(replace(ltrim(SUBSTRING(@ViewDefinition, StartPos, CHARINDEX(' ', @ViewDefinition, StartPos+1)-StartPos)), '[', ''), ']', '')) = 1 then '' else 'STA.' end + replace(replace(ltrim(SUBSTRING(@ViewDefinition, StartPos, CHARINDEX(' ', @ViewDefinition, StartPos+1)-StartPos)), '[', ''), ']', ''),')', '') as name
218 from @Positions
219 union select * from @Tabellennamen) t
220
221 -- --------------------------- Speichern der Column Names, welche wir mitnehmen wollen in das DWH ---------------------------
222
223
224 -- parsen aller .{COLUMNS} ... -> aller Spaltennamen -> nach dem "Punkt" suchen
225 -- alle, welche ein ' ' per Column enhalten e.g.
226 -- Col1 ,
227 -- Col2 , ...
228 DELETE FROM @Positions
229 SET @SEARCHPATTERN='% ,%'
230 SET @LEN = Len(REPLACE(@SEARCHPATTERN, '%', ''))
231 Select @oldpos=0
232 select @pos=patindex(@SEARCHPATTERN,@Fields)
233 while @pos > 0 and @oldpos<>@pos
234 begin
235 insert into @Positions Values (@pos+@LEN)
236 Select @oldpos=@pos
237 select @pos=patindex(@SEARCHPATTERN,Substring(@Fields,@pos + 1,len(@Fields))) + @pos
238 end
239
240
241 --select StartPos, rtrim(ltrim(case when SUBSTRING(@Fields, StartPos, CHARINDEX(' ', @Fields, StartPos)-StartPos) = '' then replace(replace(SUBSTRING(@Fields, StartPos, CHARINDEX(',', @Fields, StartPos)-StartPos), '[', ''), ']', '') else REPLACE(REPLACE(REPLACE(RTRIM(SUBSTRING(@Fields, StartPos, CHARINDEX(' ', @Fields, StartPos)-StartPos)),')',''),'[',''), ']','') end)) as Spaltenname
242 select StartPos, rtrim(ltrim(case when SUBSTRING(@Fields, StartPos, (CASE WHEN CHARINDEX(' ', @Fields, StartPos) = 0 THEN LEN(@Fields)+1 ELSE CHARINDEX(' ', @Fields, StartPos) end)-StartPos) = '' then replace(replace(SUBSTRING(@Fields, StartPos, CHARINDEX(',', @Fields, StartPos)-StartPos), '[', ''), ']', '') else REPLACE(REPLACE(REPLACE(RTRIM(SUBSTRING(@Fields, StartPos, (CASE WHEN CHARINDEX(' ', @Fields, StartPos) = 0 THEN LEN(@Fields)+1 ELSE CHARINDEX(' ', @Fields, StartPos) end)-StartPos)),')',''),'[',''), ']','') end)) as Spaltenname
243 into #Spaltenname
244 from @Positions
245 --union
246 select charindex(' ', @Fields),
247 ltrim(case when charindex('as', substring(@Fields, charindex(' ', @Fields), charindex(',', @Fields)-charindex(' ', @Fields)))>0 then substring(substring(@Fields, charindex(' ', @Fields), charindex(',', @Fields)-charindex(' ', @Fields)),0,charindex('as', substring(@Fields, charindex(' ', @Fields), charindex(',', @Fields)-charindex(' ', @Fields)))) else
248 substring(@Fields, charindex(' ', @Fields), charindex(',', @Fields)-charindex(' ', @Fields)) end)
249
250
251 -- Alle Spaltennamen, welche mit ID_ beginnen entfernen
252 -- dieser Teil kann später einmal auskommentiert werden
253 -- ist nur für den Übergang gedacht
254 delete from #Spaltenname
255 where Spaltenname like '''ID_%'
256
257 declare @skipTillPos int = 0
258 declare @initialized bit = 0
259
260 -- Parsen aller views, welche kein Blank pro Column haben
261 IF NOT EXISTS (select Spaltenname from #Spaltenname)
262 BEGIN
263 DELETE FROM @Positions
264 SET @SEARCHPATTERN=','
265 SET @LEN = Len(REPLACE(@SEARCHPATTERN, '%',''))
266 Select @oldpos=0
267 select @pos=charindex(@SEARCHPATTERN,@Fields)
268 while @pos > 0 and @oldpos<>@pos
269 begin
270 Select @oldpos=@pos
271
272 IF @initialized = 0
273 BEGIN
274 SET @initialized = 1
275 insert into @Positions Values (@pos+@LEN)
276 END
277
278 --select 'NORMAL: ' + @SourceTable
279 select @pos=charindex(@SEARCHPATTERN,Substring(@Fields,@pos + 1,len(@Fields))) + @pos
280
281 if @pos+1 >= (select CHARINDEX(',', @Fields, charindex('(', @Fields, @oldpos) -charindex(',', substring(reverse(@Fields), len(@Fields)-charindex('(', @Fields), charindex('(', @Fields))))+1)
282 and
283 @pos <=
284 (select CHARINDEX(',', @Fields, charindex(')', @Fields)-1))
285 and @pos +1 = (select CHARINDEX(',', @Fields, charindex('(', @Fields, @oldpos) -charindex(',', substring(reverse(@Fields), len(@Fields)-charindex('(', @Fields), charindex('(', @Fields))))+1)
286 begin
287 SET @skipTillPos = (select CHARINDEX(',', @Fields, charindex(')', @Fields)-1))
288 insert into @Positions Values ((select CHARINDEX(',', @Fields, charindex('(', @Fields, @oldpos) -charindex(',', substring(reverse(@Fields), len(@Fields)-charindex('(', @Fields), charindex('(', @Fields))))+1))
289 end
290 else
291 begin
292 if @pos >= @skipTillPos
293 SET @skipTillPos = 0
294
295 IF @skipTillPos = 0
296 BEGIN
297 IF NOT EXISTS (SELECT StartPos FROM @Positions WHERE @pos+@LEN = StartPos)
298 insert into @Positions Values (@pos+@LEN)
299 END
300 end
301 end
302
303 -- Rausparsen von Spaltennamen und Funktionen
304 insert into #Spaltenname
305 (StartPos, Spaltenname)
306 select StartPos,
307
308 -- ************************ wenn wir eine Funktion haben ************************
309 case when StartPos = (select CHARINDEX(',', @Fields, charindex('(', @Fields, StartPos) -charindex(',', substring(reverse(@Fields), len(@Fields)-charindex('(', @Fields), charindex('(', @Fields))))+1)
310 and (select charindex(')', @Fields, StartPos)) > 0 then
311 rtrim(ltrim(SUBSTRING(@Fields,
312 -- StartPos, String umdrehen und nach 'rückwerts' suchen
313 CHARINDEX(',', @Fields, charindex('(', @Fields, StartPos) -charindex(',', substring(reverse(@Fields), len(@Fields)-charindex('(', @Fields), charindex('(', @Fields))))+1,
314 -- EndPos
315 CHARINDEX(',', @Fields, charindex(')', @Fields)-1)
316 -- minus StartPos
317 -CHARINDEX(',', @Fields, charindex('(', @Fields, StartPos) -charindex(',', substring(reverse(@Fields), len(@Fields)-charindex('(', @Fields), charindex('(', @Fields))))-1
318 )))
319
320 -- ************************ wenn wir keine Funktion haben ************************
321 else
322 LTRIM(RTRIM(
323 case when
324 LTRIM(RTRIM(LEFT(case when substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) = '' then substring(@Fields, StartPos, len(@Fields)) else substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) end,charindex(' as ',
325 case when substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) = '' then substring(@Fields, StartPos, len(@Fields)) else substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) end ))))=''
326 then
327 case when substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) = '' then substring(@Fields, StartPos, len(@Fields)) else substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) end
328 else
329 LTRIM(RTRIM(LEFT(case when substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) = '' then substring(@Fields, StartPos, len(@Fields)) else substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) end,charindex(' as ',
330 case when substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) = '' then substring(@Fields, StartPos, len(@Fields)) else substring(substring(@Fields, StartPos, len(@Fields)-StartPos),0, charindex(',',substring(@Fields, StartPos, len(@Fields)-StartPos))) end ))))
331 end))
332 end as Spaltenname
333 from
334 @Positions
335 END
336
337 -- Nachschauen im Quellcode der View, welche Spalten und Tabellennamen wirklich verwendet werden.
338 -- Konkatinieren von Tablellenname + Columnname
339 select name + '.' + CASE WHEN CharIndex('.', Spaltenname) > 0 then SUBSTRING(REPLACE(Spaltenname, REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), ''), CharIndex('.', REPLACE(Spaltenname, REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), ''))+1, LEN(REPLACE(Spaltenname, REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), ''))-CharIndex('.', REPLACE(Spaltenname, REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), ''))) else REPLACE(Spaltenname, REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), '') end as Name
340 into #fields_in_view
341 from #Spaltenname, #Tabellen
342 where CHARINDEX(REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), @ViewDefinitionNew) > 0
343 and
344 CHARINDEX(REPLACE(Spaltenname, REPLACE(Name, 'STA.'+@SourceSchema+'.', ''), ''), @ViewDefinitionNew) > 0
345
346 -- --------------------------- Alles zusammensetzen ---------------------------
347
348 -- Enums
349 SELECT distinct *
350 into #Definitions
351 FROM
352 (
353 -- Umbennenung der Columns
354 SELECT rename.OriginalName, replace(rename.DesiredName,',','') as [SPALTENNAME_NEU], Definitions.* FROM
355 (
356 -- Nur columns, welche wir mitnehmen wollen (wie definiert in den Views), sollen auch im
357 -- DWH angelegt werden (#Spaltenname)
358 SELECT Definitions.* FROM
359 (
360 ---- Holen der Definitionen der Columns aus den Quell-Tabellen
361 select isc.TABLE_SCHEMA, isc.TABLE_NAME, primary_key.Name as [PRIMARY_KEY], isc.COLUMN_NAME, isc.IS_NULLABLE, isc.DATA_TYPE, isc.CHARACTER_MAXIMUM_LENGTH, isc.CHARACTER_OCTET_LENGTH, isc.NUMERIC_PRECISION, isc.NUMERIC_PRECISION_RADIX, isc.NUMERIC_SCALE, isc.DATETIME_PRECISION, isc.ORDINAL_POSITION from STA.INFORMATION_SCHEMA.COLUMNS isc
362 INNER JOIN #Tabellen desiredTable on REPLACE(desiredTable.Name, 'STA.', '') = isc.TABLE_SCHEMA + '.' + isc.TABLE_NAME
363 INNER JOIN STA.sys.schemas s on s.name = isc.TABLE_SCHEMA
364 INNER JOIN STA.sys.objects t on t.name = isc.TABLE_NAME
365 LEFT OUTER JOIN
366 (
367 -- Holen der PKs aufgrund der Tabellennamen
368 SELECT i.name as Name, c.name as ColumnName, sc.schema_id, o.name as TableName
369 FROM STA.sys.indexes i
370 INNER JOIN STA.sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id
371 INNER JOIN STA.sys.columns c ON ic.object_id = c.object_id and ic.column_id = c.column_id
372 INNER JOIN STA.sys.objects o ON i.object_id = o.object_id
373 INNER JOIN STA.sys.schemas sc ON o.schema_id = sc.schema_id
374 INNER JOIN #Tabellen tabellen on UPPER(tabellen.Name) = UPPER('STA.' + sc.name + '.' + o.name)
375 -- Die Client Nummer wird immer ignoriert, da diese im System nicht verwendet wird
376 WHERE i.is_primary_key = 1 and c.name <> 'client'
377 --ORDER BY o.Name, i.Name, ic.key_ordinal
378 ) primary_key on primary_key.TableName = isc.TABLE_NAME and primary_key.schema_id = s.schema_id and primary_key.ColumnName = isc.COLUMN_NAME
379 ) Definitions
380 LEFT OUTER JOIN #Spaltenname on Definitions.COLUMN_NAME = #Spaltenname.Spaltenname OR Definitions.TABLE_NAME + '.' + Definitions.COLUMN_NAME = #Spaltenname.Spaltenname
381 INNER JOIN #fields_in_view fiv on fiv.Name = 'STA.' + @SourceSchema + '.' + Definitions.TABLE_NAME + '.' + Definitions.COLUMN_NAME
382 ) Definitions
383 LEFT OUTER JOIN #Parts rename on rename.OriginalName like '%' + Definitions.TABLE_NAME + '.' + Definitions.COLUMN_NAME + '%' or rename.OriginalName = Definitions.COLUMN_NAME or rename.OriginalName = Definitions.TABLE_NAME + '.' + Definitions.COLUMN_NAME
384 ) Definitions
385 -- Collationkonflikt -> ist jedoch nicht relevant, da nur im CFG_LOG, weder in Source noch in Ziel
386 LEFT OUTER JOIN CFG_LOG.dbo.CFG_TRANSL_ENUM enums on enums.ColumnName = Definitions.COLUMN_NAME COLLATE Latin1_General_CI_AS and
387 enums.DBName = N'STA' COLLATE Latin1_General_CI_AS and
388 enums.TableName = Definitions.TABLE_NAME COLLATE Latin1_General_CI_AS
389 order by ORDINAL_POSITION asc
390
391 -- Spezialfall,
392 -- wenn wir ein Objekt hard-coded in einer View hinterlegen, so wird die spaltendefinition für jenes per Insert hinzugefügt
393 INSERT INTO #Definitions
394 ( OriginalName, SPALTENNAME_NEU, TABLE_SCHEMA, TABLE_NAME, PRIMARY_KEY, COLUMN_NAME, IS_NULLABLE, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, CHARACTER_OCTET_LENGTH, NUMERIC_PRECISION, NUMERIC_PRECISION_RADIX, NUMERIC_SCALE, DATETIME_PRECISION, ORDINAL_POSITION )
395 select NULL as OriginalName, NULL as SPALTENNAME_NEU, TABLE_SCHEMA, TABLE_NAME, NULL as PRIMARY_KEY, COLUMN_NAME, IS_NULLABLE, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, CHARACTER_OCTET_LENGTH, NUMERIC_PRECISION, NUMERIC_PRECISION_RADIX, NUMERIC_SCALE, DATETIME_PRECISION, ORDINAL_POSITION from STA.INFORMATION_SCHEMA.COLUMNS isc
396 --INNER JOIN #Tabellen desiredTable on REPLACE(desiredTable.Name, 'STA.', '') = isc.TABLE_SCHEMA + '.' + isc.TABLE_NAME
397 INNER JOIN STA.sys.schemas s on s.name = isc.TABLE_SCHEMA
398 INNER JOIN STA.sys.objects t on t.name = isc.TABLE_NAME
399 WHERE TABLE_NAME = @SourceTable and COLUMN_NAME not in (select CASE WHEN SPALTENNAME_NEU IS NOT NULL THEN SPALTENNAME_NEU ELSE COLUMN_NAME END as COLUMN_NAME from #Definitions) and LEFT(COLUMN_NAME,3) != 'ID_'
400
401 --select * from #Definitions
402 --select * from #Parts
403 --select * from #Spaltenname
404 --select * from #fields_in_view
405 --select * from #Definitions
406
407 -- --------------------------- CLEANUP---------------------------
408
409 -- Löschen der Temporären Objekte
410 drop table #Parts
411 drop table #Tabellen
412 drop table #Spaltenname
413 drop table #fields_in_view
414
415 -- --------------------------- Erstellen der Tabellen ---------------------------
416
417 -- Prüfen ob die Tabellen schon im DWH vorhanden sind
418 IF NOT EXISTS (select * from DWH.sys.objects o
419 inner join DWH.sys.schemas s on s.schema_id = o.schema_id
420 where @TargetTable = o.name and @TargetSchema = s.name)
421 BEGIN
422 -- Wenn nicht vorhanden, dann anlegen
423 PRINT 'Tablle ' + @TargetTable + ' noch nicht vorhanden, wird angelegt'
424
425 DECLARE @Query AS NVARCHAR(MAX)
426
427 USE DWH;
428 SET @Query = 'CREATE TABLE [' + @TargetSchema + '].[' + @TargetTable + '] ( '
429
430 DECLARE @COLUMN_DEFINITION AS NVARCHAR(MAX) = ''
431
432 -- ****************** ZUSAMMENBAUEN von Column Definition ******************
433
434 select @COLUMN_DEFINITION=@COLUMN_DEFINITION+
435
436 -- Menschlich leslicher (verweichlichter) Spaltenname oder Originalname
437 '['+ CASE WHEN SPALTENNAME_NEU IS NOT NULL then SPALTENNAME_NEU ELSE COLUMN_NAME END + '] ' +
438
439 -- Datentyp
440 CASE WHEN
441 -- ohne Precision oder scale
442 DATA_TYPE = 'bit' OR DATA_TYPE = 'int' OR DATA_TYPE = 'bigint' OR DATA_TYPE = 'money' OR DATA_TYPE = 'ntext' OR DATA_TYPE = 'text' OR DATA_TYPE = 'datetime' then '[' + DATA_TYPE + ']' ELSE
443
444 -- mit precision oder scale
445 CASE WHEN
446 -- (n)(var)char
447 DATA_TYPE = 'nvarchar' OR DATA_TYPE = 'varchar' OR DATA_TYPE = 'char' OR DATA_TYPE = 'nchar' THEN
448 '[' + DATA_TYPE + '] (' + cast(CHARACTER_MAXIMUM_LENGTH as varchar(MAX)) + ')'
449 ELSE
450 -- float
451 '[' + DATA_TYPE + '] (' + cast(NUMERIC_SCALE as varchar(MAX)) + ', ' + cast(NUMERIC_PRECISION as varchar(MAX)) + ')'
452 END
453 END +
454
455 -- PRIMARY KEY Ja/Nein
456 --CASE WHEN PRIMARY_KEY IS NOT NULL then ' PRIMARY KEY ' ELSE ' ' END +
457
458 -- NULL oder NOT NULL
459 CASE WHEN IS_NULLABLE = 'YES' then ' ' ELSE ' NOT NULL ' END +
460
461 ', ' from #Definitions
462
463 -- ****************** ZUSAMMESETZEN VON CREATE STATEMENT ******************
464
465 SET @Query = @Query + SUBSTRING(@COLUMN_DEFINITION,0,LEN(@COLUMN_DEFINITION)-1) + ' )'
466
467 print @Query
468 exec sp_sqlexec @Query
469 END
470 ELSE
471 BEGIN
472 -- Wenn vorhanden, dann Daten mergen
473 PRINT 'Tabelle ' + @TargetTable + ' bereits vorhanden'
474
475 BEGIN TRY
476 /*********************************************************************************/
477 /*** Kontrollieren ob im Source View unbekannte Tabellenfelder vorhanden sind ***/
478 /*********************************************************************************/
479 DECLARE cur_Check_Column CURSOR READ_ONLY FOR
480 SELECT TabSource.COLUMN_NAME
481 FROM STA.INFORMATION_SCHEMA.COLUMNS TabSource
482 WHERE TabSource.TABLE_SCHEMA = @SourceSchema
483 AND TabSource.TABLE_NAME = @SourceTable
484 AND TabSource.TABLE_CATALOG = 'STA'
485
486 -- Folgende Einschränkung in der Where-Clause wird später nicht mehr verwendet werden
487 -- Kann somit sobald umgestellt entfernet werden
488 AND TabSource.COLUMN_NAME NOT lIKE 'ID_%'
489 AND NOT EXISTS
490 (SELECT TabSource.COLUMN_NAME FROM DWH.INFORMATION_SCHEMA.COLUMNS TabTarget
491 WHERE TabTarget.TABLE_SCHEMA = @TargetSchema
492 AND TabTarget.TABLE_NAME = @TargetTable
493 AND TabTarget.TABLE_CATALOG = 'DWH'
494 AND ISNULL(TabSource.CHARACTER_MAXIMUM_LENGTH, '') = ISNULL(TabTarget.CHARACTER_MAXIMUM_LENGTH,'')
495 AND TabSource.DATA_TYPE = TabTarget.DATA_TYPE
496 AND TabTarget.COLUMN_NAME = TabSource.COLUMN_NAME)
497 END TRY
498
499 BEGIN CATCH
500 INSERT INTO CFG_LOG.dbo.LogTable
501 ([DateTime]
502 ,[Funktion]
503 ,[Text]
504 ,[Status])
505 SELECT SYSDATETIME() as ZEIT
506 ,'MergeErrorHandling'
507 ,ERROR_MESSAGE() + ' ErrorNumber' + CAST(ERROR_NUMBER()as VARCHAR(4)) as Error
508 ,'Error' as a;
509 END CATCH
510
511 OPEN cur_Check_Column
512 FETCH NEXT FROM cur_Check_Column INTO @ColumnName
513 /**** falls spalten fehlen Fehleerbehandlung ****/
514 WHILE (@@FETCH_STATUS <> -1)
515 BEGIN
516 INSERT INTO CFG_LOG.dbo.LogTable
517 ([DateTime]
518 ,[Funktion]
519 ,[Text]
520 ,[Status])
521 SELECT SYSDATETIME() as ZEIT
522 ,'MergeErrorHandling'
523 ,'Achtung in der Zieltabelle ' + @SourceSchema + '.' + @SourceTable + ' Fehlt die Spalte ' + @ColumnName
524 ,'Error' as a ;
525 PRINT (@ColumnName)
526 FETCH NEXT FROM cur_Check_Column INTO @ColumnName
527 END
528
529 CLOSE cur_Check_Column
530 DEALLOCATE cur_Check_Column
531 END
532
533 SELECT COLUMN_NAME, case when SPALTENNAME_NEU IS NOT NULL THEN SPALTENNAME_NEU else COLUMN_NAME end as Name, CASE WHEN PRIMARY_KEY IS NULL THEN 0 ELSE 1 END as PK
534 --SELECT COLUMN_NAME, SPALTENNAME_NEU, CASE WHEN PRIMARY_KEY IS NULL THEN 0 ELSE 1 END as PK
535 into #ColumnsAndPKs
536 FROM #Definitions
537
538 select * from #ColumnsAndPKs
539
540 /*************************************************************************/
541 /** Primary Keys auslesen ***/
542 /*************************************************************************/
543 DECLARE cur_Check_Column CURSOR READ_ONLY FOR
544 select distinct COLUMN_NAME, Name, PK from #ColumnsAndPKs
545
546 DECLARE @MergeUsing as varchar(max)
547 set @MergeUsing = ''
548 DECLARE @MergeOn as varchar(max)
549 set @MergeOn = ''
550 DECLARE @MergeInsert_Value as varchar(max)
551 set @MergeInsert_Value = ''
552 DECLARE @Update_Set as varchar(max)
553 set @Update_Set = ''
554 DECLARE @Delete_Set as varchar(max)
555 set @Delete_Set = ''
556
557 select * from #Definitions
558
559 OPEN cur_Check_Column
560 FETCH NEXT FROM cur_Check_Column INTO @ColumnName,@ColumnNameNew, @IsPrimaryKey
561 WHILE (@@FETCH_STATUS <> -1)
562 BEGIN
563 -- Wenn wir gewisse Columns nicht im DWH haben, skippen
564 IF exists(select ordinal_position from DWH.INFORMATION_SCHEMA.COLUMNS isc
565 where COLUMN_NAME = @ColumnNameNew)
566 begin
567
568 IF UPPER(@ColumnName) != UPPER(@ColumnNameNew)
569 begin
570 PRINT 'COLUMN_NAME: ' +@ColumnName + ' (' + @ColumnNameNew + ')'
571 end
572 ELSE
573 begin
574 PRINT 'COLUMN_NAME: ' +@ColumnName
575 end
576
577 /* Using */
578 IF @IsPrimaryKey = 1
579 BEGIN
580 SET @MergeUsing = ''
581 END
582
583 /* On */
584 IF @IsPrimaryKey = 1
585 BEGIN
586 SET @MergeOn = @MergeOn + 'a.' + @ColumnNameNew + ' = ' + 'b.'+ @ColumnNameNew + ' and '
587 END
588
589 /* @MergeInsert_Value */
590 --IF @IsPrimaryKey = 0
591 --BEGIN
592 SET @MergeInsert_Value = @MergeInsert_Value + @ColumnNameNew + ','
593 --END
594
595 /* On */
596 --IF @IsPrimaryKey = 0
597 --BEGIN
598 SET @Update_Set = @Update_Set + 'a.' + @ColumnNameNew + ' = ' + 'b.' + @ColumnNameNew + ' , '
599 --END
600 END
601 ELSE
602 BEGIN
603 IF UPPER(@ColumnName) != UPPER(@ColumnNameNew)
604 begin
605 PRINT 'COLUMN_NAME: ' +@ColumnName + ' (' + @ColumnNameNew + ') <- NOK'
606 end
607 ELSE
608 begin
609 PRINT 'COLUMN_NAME: ' +@ColumnName + ' <- NOK'
610 END
611 END
612
613 FETCH NEXT FROM cur_Check_Column INTO @ColumnName,@ColumnNameNew, @IsPrimaryKey
614 END
615
616 IF (@MergeOn != '')
617 SET @MergeOn = LEFT(@MergeOn , Len(@MergeOn) - 4)
618
619 IF (@MergeInsert_Value != '')
620 SET @MergeInsert_Value = LEFT(@MergeInsert_Value , Len(@MergeInsert_Value) - 1)
621
622 IF (@Update_Set != '')
623 SET @Update_Set = LEFT(@Update_Set , Len(@Update_Set) - 2)
624
625 --print ''
626 --print 'USING: ' + @MergeUsing
627 --print 'ON: ' + @MergeOn
628 --print 'INSERT: ' + @MergeInsert_Value
629 --print 'UPDATE: ' + @Update_Set
630
631 CLOSE cur_Check_Column
632 DEALLOCATE cur_Check_Column
633 drop table #ColumnsAndPKs
634
635 IF @MergeOn = ''
636 BEGIN
637 INSERT INTO CFG_LOG.dbo.LogTable
638 ([DateTime]
639 ,[Funktion]
640 ,[Text]
641 ,[Status])
642 SELECT SYSDATETIME() as ZEIT
643 ,'Merge'
644 ,'Merge bei Tabelle ' + @TargetTable + ' nicht möglich.'
645 ,'Error' as Status ;
646
647 PRINT 'ERROR: Merge bei Tabelle ' + @TargetTable + ' nicht möglich.'
648 END
649 ELSE
650 BEGIN
651 SET @sql = 'MERGE ' + 'DWH.' + @TargetSchema + '.' + @TargetTable + ' as a' +
652 ' USING ' +
653 ' (SELECT * FROM sta.' + @SourceSchema + '.' + @SourceTable + ' as c' +
654 -- ' where ' +
655 @MergeUsing + ' ) AS b' +
656 ' on ' +
657 @MergeOn +
658 ' WHEN NOT MATCHED BY TARGET THEN ' +
659 ' INSERT (' + @MergeInsert_Value + ')' +
660 ' VALUES (' + @MergeInsert_Value + ')' +
661 ' WHEN MATCHED THEN ' +
662 ' UPDATE SET ' +
663 @Update_Set +
664 ' WHEN NOT MATCHED BY SOURCE THEN ' +
665 ' DELETE ' +
666 ' OUTPUT $action INTO dwh.dbo.rowcounts ' +
667 ' ; '
668
669 print '<' + @TargetTable + '>'
670 print @sql
671 print '</' + @TargetTable + '>'
672
673 BEGIN TRY
674 BEGIN TRANSACTION
675 /* Löschen aller Einträge für das Log */
676 TRUNCATE TABLE dwh.dbo.rowcounts
677
678 --EXEC(@sql)
679
680 COMMIT TRANSACTION
681 END TRY
682 BEGIN CATCH
683 ROLLBACK TRANSACTION
684 INSERT INTO CFG_LOG.dbo.LogTable
685 ([DateTime]
686 ,[Funktion]
687 ,[Text]
688 ,[Status])
689 SELECT SYSDATETIME() as ZEIT
690 ,'Fehler beim Merge'
691 ,ERROR_MESSAGE() + ' ErrorNumber' + CAST(ERROR_NUMBER()as VARCHAR(4)) as Error
692 ,'Error' as a;
693 PRINT 'ERROR: Merge bei Tabelle ' + @TargetTable + ' nicht möglich.'
694 END CATCH
695
696
697 /************ Output abfrage wie viele Daten inserted/update/deletet/sind ************/
698 USE DWH
699
700 SELECT @insertcount = [INSERT]
701 ,@updatecount = [UPDATE]
702 ,@deletecount = [DELETE]
703 FROM
704 (SELECT mergeAction, 1 [rows]
705 FROM dwh.dbo.rowcounts) p
706 PIVOT (COUNT(rows) FOR mergeAction
707 IN ([INSERT]
708 ,[UPDATE]
709 ,[DELETE])
710 ) as pvt
711
712 /* eintrag funktioniert sonst nicht :-( BUG?? */
713 DECLARE @TargetTable1 as varchar(150)
714 SET @TargetTable1 = @TargetTable
715 /*----------------------------------------*/
716
717 INSERT INTO CFG_LOG.dbo.LogTable
718 ([DateTime]
719 ,[Funktion]
720 ,[Text]
721 ,[Status])
722 SELECT SYSDATETIME() as ZEIT
723 ,'Merge'
724 ,'Merge in Tabelle ' + cast(UPPER(@TargetTable1) as nvarchar(150)) + ' Erfolgreich ausgeführt. Es wurden ' + cast(@insertcount as varchar(10)) + ' eingefügt und ' + cast(@updatecount as varchar(10)) + ' upgedatet und ' + cast(@deletecount as varchar(10)) + ' gelöscht.' as b
725 ,'Success' as a
726 END
727
728 FETCH NEXT FROM cur_cfg_dwh_import INTO @SourceSchema, @SourceTable, @TargetSchema, @TargetTable
729
730 -- Temporäres Objekt wird nicht mehr benötigt
731 drop table #Definitions
732END
733
734CLOSE cur_cfg_dwh_import
735DEALLOCATE cur_cfg_dwh_import
736
737INSERT INTO CFG_LOG.dbo.LogTable
738 ([DateTime]
739 ,[Funktion]
740 ,[Text]
741 ,[Status])
742 SELECT SYSDATETIME() as ZEIT
743 ,'Merge'
744 ,'Der Merge wurde erfolgreich beendet'
745 ,'INFO' as Status ;