· 8 years ago · Dec 20, 2017, 01:50 PM
1-- CFG_DWH_IMPORT
2SET NOCOUNT ON
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
539
540 /*************************************************************************/
541 /** Primary Keys auslesen ***/
542 /*************************************************************************/
543 DECLARE @PREV_NAME NVARCHAR(MAX) = ''
544 DECLARE cur_Check_Column CURSOR READ_ONLY FOR
545 select distinct COLUMN_NAME, Name, PK from #ColumnsAndPKs order by Name
546
547 DECLARE @MergeUsing as varchar(max)
548 set @MergeUsing = ''
549 DECLARE @MergeOn as varchar(max)
550 set @MergeOn = ''
551 DECLARE @MergeInsert_Value as varchar(max)
552 set @MergeInsert_Value = ''
553 DECLARE @Update_Set as varchar(max)
554 set @Update_Set = ''
555 DECLARE @Delete_Set as varchar(max)
556 set @Delete_Set = ''
557
558 select * from #Definitions
559
560 OPEN cur_Check_Column
561 FETCH NEXT FROM cur_Check_Column INTO @ColumnName,@ColumnNameNew, @IsPrimaryKey
562 WHILE (@@FETCH_STATUS <> -1)
563 BEGIN
564 -- Wenn wir gewisse Columns nicht im DWH haben, skippen
565 IF exists(select ordinal_position from DWH.INFORMATION_SCHEMA.COLUMNS isc
566 where COLUMN_NAME = @ColumnNameNew)
567 begin
568 /* On */
569 IF @IsPrimaryKey = 1
570 BEGIN
571 SET @MergeOn = @MergeOn + 'a.' + @ColumnNameNew + ' = ' + 'b.'+ @ColumnNameNew + ' and '
572 END
573
574 IF (@PREV_NAME = '' or @PREV_NAME != @ColumnName)
575 BEGIN
576
577 IF UPPER(@ColumnName) != UPPER(@ColumnNameNew)
578 begin
579 PRINT 'COLUMN_NAME: ' +@ColumnName + ' (' + @ColumnNameNew + ')'
580 end
581 ELSE
582 begin
583 PRINT 'COLUMN_NAME: ' +@ColumnName
584 end
585
586 /* Using */
587 IF @IsPrimaryKey = 1
588 BEGIN
589 SET @MergeUsing = ''
590 END
591
592 /* @MergeInsert_Value */
593 --IF @IsPrimaryKey = 0
594 --BEGIN
595 SET @MergeInsert_Value = @MergeInsert_Value + @ColumnNameNew + ','
596 --END
597
598 /* On */
599 --IF @IsPrimaryKey = 0
600 --BEGIN
601 SET @Update_Set = @Update_Set + 'a.' + @ColumnNameNew + ' = ' + 'b.' + @ColumnNameNew + ' , '
602 --END
603
604 SET @PREV_NAME = @ColumnName
605 END
606 ELSE
607 PRINT 'Sorting out ' + @PREV_NAME
608 END
609 ELSE
610 BEGIN
611 IF UPPER(@ColumnName) != UPPER(@ColumnNameNew)
612 begin
613 PRINT 'COLUMN_NAME: ' +@ColumnName + ' (' + @ColumnNameNew + ') <- NOK'
614 end
615 ELSE
616 begin
617 PRINT 'COLUMN_NAME: ' +@ColumnName + ' <- NOK'
618 END
619 END
620
621 FETCH NEXT FROM cur_Check_Column INTO @ColumnName,@ColumnNameNew, @IsPrimaryKey
622 END
623
624 IF (@MergeOn != '')
625 SET @MergeOn = LEFT(@MergeOn , Len(@MergeOn) - 4)
626
627 IF (@MergeInsert_Value != '')
628 SET @MergeInsert_Value = LEFT(@MergeInsert_Value , Len(@MergeInsert_Value) - 1)
629
630 IF (@Update_Set != '')
631 SET @Update_Set = LEFT(@Update_Set , Len(@Update_Set) - 2)
632
633 --print ''
634 --print 'USING: ' + @MergeUsing
635 --print 'ON: ' + @MergeOn
636 --print 'INSERT: ' + @MergeInsert_Value
637 --print 'UPDATE: ' + @Update_Set
638
639 CLOSE cur_Check_Column
640 DEALLOCATE cur_Check_Column
641 drop table #ColumnsAndPKs
642
643 IF @MergeOn = ''
644 BEGIN
645 INSERT INTO CFG_LOG.dbo.LogTable
646 ([DateTime]
647 ,[Funktion]
648 ,[Text]
649 ,[Status])
650 SELECT SYSDATETIME() as ZEIT
651 ,'Merge'
652 ,'Merge bei Tabelle ' + @TargetTable + ' nicht möglich.'
653 ,'Error' as Status ;
654
655 PRINT 'ERROR: Es kann kein Primary Key bei Tabelle ' + @TargetTable + ' gefunden werden.'
656 END
657 ELSE
658 BEGIN
659 SET @sql = 'MERGE ' + 'DWH.' + @TargetSchema + '.' + @TargetTable + ' as a' +
660 ' USING ' +
661 ' (SELECT * FROM sta.' + @SourceSchema + '.' + @SourceTable + ' as c' +
662 -- ' where ' +
663 @MergeUsing + ' ) AS b' +
664 ' on ' +
665 @MergeOn +
666 ' WHEN NOT MATCHED BY TARGET THEN ' +
667 ' INSERT (' + @MergeInsert_Value + ')' +
668 ' VALUES (' + @MergeInsert_Value + ')' +
669 ' WHEN MATCHED THEN ' +
670 ' UPDATE SET ' +
671 @Update_Set +
672 ' WHEN NOT MATCHED BY SOURCE THEN ' +
673 ' DELETE ' +
674 ' OUTPUT $action INTO dwh.dbo.rowcounts ' +
675 ' ; '
676
677 print '<' + @TargetTable + '>'
678 print @sql
679 print '</' + @TargetTable + '>'
680
681 BEGIN TRY
682 BEGIN TRANSACTION
683 /* Löschen aller Einträge für das Log */
684 TRUNCATE TABLE dwh.dbo.rowcounts
685
686 SET NOCOUNT OFF
687 EXEC(@sql)
688 SET NOCOUNT ON
689
690 COMMIT TRANSACTION
691 END TRY
692 BEGIN CATCH
693 ROLLBACK TRANSACTION
694 INSERT INTO CFG_LOG.dbo.LogTable
695 ([DateTime]
696 ,[Funktion]
697 ,[Text]
698 ,[Status])
699 SELECT SYSDATETIME() as ZEIT
700 ,'Fehler beim Merge'
701 ,ERROR_MESSAGE() + ' ErrorNumber' + CAST(ERROR_NUMBER()as VARCHAR(4)) as Error
702 ,'Error' as a;
703 PRINT 'ERROR: Merge bei Tabelle ' + @TargetTable + ' nicht möglich.'
704 END CATCH
705
706
707 /************ Output abfrage wie viele Daten inserted/update/deletet/sind ************/
708 USE DWH
709
710 SELECT @insertcount = [INSERT]
711 ,@updatecount = [UPDATE]
712 ,@deletecount = [DELETE]
713 FROM
714 (SELECT mergeAction, 1 [rows]
715 FROM dwh.dbo.rowcounts) p
716 PIVOT (COUNT(rows) FOR mergeAction
717 IN ([INSERT]
718 ,[UPDATE]
719 ,[DELETE])
720 ) as pvt
721
722 /* eintrag funktioniert sonst nicht :-( BUG?? */
723 DECLARE @TargetTable1 as varchar(150)
724 SET @TargetTable1 = @TargetTable
725 /*----------------------------------------*/
726
727 INSERT INTO CFG_LOG.dbo.LogTable
728 ([DateTime]
729 ,[Funktion]
730 ,[Text]
731 ,[Status])
732 SELECT SYSDATETIME() as ZEIT
733 ,'Merge'
734 ,'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
735 ,'Success' as a
736 END
737
738 FETCH NEXT FROM cur_cfg_dwh_import INTO @SourceSchema, @SourceTable, @TargetSchema, @TargetTable
739
740 -- Temporäres Objekt wird nicht mehr benötigt
741 drop table #Definitions
742END
743
744CLOSE cur_cfg_dwh_import
745DEALLOCATE cur_cfg_dwh_import
746
747INSERT INTO CFG_LOG.dbo.LogTable
748 ([DateTime]
749 ,[Funktion]
750 ,[Text]
751 ,[Status])
752 SELECT SYSDATETIME() as ZEIT
753 ,'Merge'
754 ,'Der Merge wurde erfolgreich beendet'
755 ,'INFO' as Status ;