Alter master views to account for quirk in PAF schema where mainfile address_key...
[postgresql/geographic_data] / paf_postgresql_import_postgis.plpgsql
1 DROP FUNCTION IF EXISTS public.import_pc_paf(varchar, varchar);
2 CREATE OR REPLACE FUNCTION public.import_pc_paf(in_edition varchar, in_data varchar)
3 RETURNS boolean AS
4 $BODY$
5
6 DECLARE
7         v_data_root varchar;
8         v_data_main varchar;
9         v_data_postzon varchar;
10         v_data_alias varchar;
11         v_processed integer;
12         v_main_footer varchar;
13         v_std_footer varchar;
14         v_sql varchar;
15         v_table_created boolean;
16 BEGIN
17         v_data_root := in_data || '/' || in_edition || '/';
18         v_data_main := v_data_root || 'PAF MAIN FILE/';
19         v_data_postzon := v_data_root || 'POSTZON 100M/';
20         v_data_alias := v_data_root || 'ALIAS/';
21         
22         v_main_footer = '       %';
23         v_std_footer = '99999999%';
24         
25         RAISE NOTICE '%: Import starting for edition % with data root %', clock_timestamp(), in_edition, v_data_root;
26         CREATE TEMPORARY TABLE data_stage (data text) ON COMMIT DROP;
27         
28         -- 1) Localaties
29         TRUNCATE TABLE data_stage;
30         
31         RAISE NOTICE '%: Begin staging localaties', clock_timestamp();
32
33         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'local.c01');
34         EXECUTE v_sql;
35         
36         DELETE FROM data_stage WHERE data like '%LOCALITY'  || in_edition ||  '%' OR data like v_std_footer;
37
38         RAISE NOTICE '%: Done staging localaties, importing', clock_timestamp();
39         
40         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_localities') THEN
41                 RAISE NOTICE 'Table "public"."pc_paf_localities" already exists';
42                 v_table_created := false;
43         ELSE
44                 RAISE NOTICE 'Creating table "public"."pc_paf_localities"';
45                 CREATE TABLE public.pc_paf_localities (
46                         locality_key integer NOT NULL,
47                         post_town varchar(30),
48                         dependent_locality varchar(35),
49                         double_dependent_locality  varchar(35)
50                 );
51                 v_table_created := true;
52         END IF;
53         
54         TRUNCATE TABLE public.pc_paf_localities;
55         INSERT INTO public.pc_paf_localities
56         SELECT substring(data,1,6)::integer,
57                 nullif(trim(substring(data,52,30)),''),
58                 nullif(trim(substring(data,82,35)),''),
59                 nullif(trim(substring(data,117,35)),'')
60         FROM data_stage;
61         
62         GET DIAGNOSTICS v_processed = ROW_COUNT;
63         
64         IF (v_table_created) THEN
65                 ALTER TABLE public.pc_paf_localities ADD PRIMARY KEY (locality_key);
66         END IF;
67         
68         RAISE NOTICE '%: Done importing localities (imported % records)', clock_timestamp(), v_processed;
69         
70         -- 2) Thoroughfares
71         TRUNCATE TABLE data_stage;
72                 
73         RAISE NOTICE '%: Begin staging thoroughfares', clock_timestamp();
74
75         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'thfare.c01');
76         EXECUTE v_sql;
77         
78         DELETE FROM data_stage WHERE data like '%THOROUGH'  || in_edition ||  '%' OR data like v_std_footer;                          
79
80         RAISE NOTICE '%: Done staging thoroughfares, importing', clock_timestamp();
81         
82         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_thoroughfares') THEN
83                 RAISE NOTICE 'Table "public"."pc_paf_thoroughfares" already exists';
84                 v_table_created := false;
85         ELSE
86                 RAISE NOTICE 'Creating table "public"."pc_paf_thoroughfares"';
87                 CREATE TABLE public.pc_paf_thoroughfares (
88                         thoroughfare_key integer NOT NULL,
89                         thoroughfare_name varchar(60)
90                 );
91                 v_table_created := true;
92         END IF;
93         
94         TRUNCATE TABLE public. pc_paf_thoroughfares;
95         INSERT INTO public. pc_paf_thoroughfares
96         SELECT substring(data,1,8)::integer,
97                 nullif(trim(substring(data,9,60)),'')
98         FROM data_stage;
99
100         GET DIAGNOSTICS v_processed = ROW_COUNT;
101         
102         IF (v_table_created) THEN
103                 ALTER TABLE public.pc_paf_thoroughfares ADD PRIMARY KEY (thoroughfare_key);
104         END IF;
105         
106         RAISE NOTICE '%: Done importing thoroughfares (imported % records)', clock_timestamp(), v_processed;
107         
108         -- 3) Thoroughfares Descriptor
109         TRUNCATE TABLE data_stage;
110         
111         RAISE NOTICE '%: Begin staging thoroughfares descriptor', clock_timestamp();
112
113         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'thdesc.c01');
114         EXECUTE v_sql;
115         
116         DELETE FROM data_stage WHERE data like '%THDESCRI'  || in_edition ||  '%' OR data like v_std_footer; 
117         
118         RAISE NOTICE '%: Done staging thoroughfares descriptor, importing', clock_timestamp();
119         
120         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_thoroughfare_descriptor') THEN
121                 RAISE NOTICE 'Table "public"."pc_paf_thoroughfare_descriptor" already exists';
122                 v_table_created := false;
123         ELSE
124                 RAISE NOTICE 'Creating table "public"."pc_paf_thoroughfare_descriptor"';
125                 CREATE TABLE public.pc_paf_thoroughfare_descriptor (
126                         thoroughfare_descriptor_key integer NOT NULL,
127                         thoroughfare_descriptor varchar(20),
128                         approved_abbreviation varchar(6)
129                 );
130                 v_table_created := true;
131         end if;
132         
133         TRUNCATE TABLE public.pc_paf_thoroughfare_descriptor;
134         INSERT INTO public.pc_paf_thoroughfare_descriptor
135         SELECT substring(data,1,4)::integer,
136                 nullif(trim(substring(data,5,20)),''),
137                 nullif(trim(substring(data,25,6)),'')
138         FROM data_stage;
139         
140         GET DIAGNOSTICS v_processed = ROW_COUNT;
141         
142         IF (v_table_created) THEN
143                 ALTER TABLE public.pc_paf_thoroughfare_descriptor ADD PRIMARY KEY (thoroughfare_descriptor_key);
144         END IF;
145         
146         RAISE NOTICE '%: Done importing thoroughfares descriptor (imported % records)', clock_timestamp(), v_processed; 
147         
148         -- 4) Building Names
149         TRUNCATE TABLE data_stage;
150         
151         RAISE NOTICE '%: Begin staging building names', clock_timestamp();
152         
153         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'bname.c01');
154         EXECUTE v_sql;
155
156         DELETE FROM data_stage WHERE data like '%BUILDING'  || in_edition ||  '%' OR data like v_std_footer; 
157
158         RAISE NOTICE '%: Done staging building names, importing', clock_timestamp();
159         
160         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_building_names') THEN
161                 RAISE NOTICE 'Table "public"."pc_paf_building_names" already exists';
162                 v_table_created := false;
163         ELSE
164                 RAISE NOTICE 'Creating table "public"."pc_paf_building_names"';
165                 CREATE TABLE public.pc_paf_building_names (
166                         building_name_key integer NOT NULL,
167                         building_name varchar(50)
168                 );
169                 v_table_created := true;
170         END IF;
171
172         TRUNCATE TABLE public.pc_paf_building_names;
173         INSERT INTO public.pc_paf_building_names
174         SELECT substring(data,1,8)::integer,
175                 nullif(trim(substring(data,9,50)),'')
176         FROM data_stage;
177         
178         GET DIAGNOSTICS v_processed = ROW_COUNT;
179         
180         IF (v_table_created) THEN
181                 ALTER TABLE public.pc_paf_building_names ADD PRIMARY KEY (building_name_key);
182         END IF;
183         
184         RAISE NOTICE '%: Done importing building names (imported % records)', clock_timestamp(), v_processed;   
185         
186         -- 5) Sub Building Names file (subbname.c01)
187         TRUNCATE TABLE data_stage;
188         
189         RAISE NOTICE '%: Begin staging sub building names', clock_timestamp();
190         
191         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'subbname.c01');
192         EXECUTE v_sql;
193
194         DELETE FROM data_stage WHERE data like '%SUBBUILD'  || in_edition ||  '%' OR data like v_std_footer; 
195
196         RAISE NOTICE '%: Done staging sub building names, importing', clock_timestamp();
197         
198         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_sub_building_names') THEN
199                 RAISE NOTICE 'Table "public"."pc_paf_sub_building_names" already exists';
200                 v_table_created := false;
201         ELSE
202                 RAISE NOTICE 'Creating table "public"."pc_paf_sub_building_names"';
203                 CREATE TABLE public.pc_paf_sub_building_names (
204                         sub_building_name_key integer NOT NULL,
205                         sub_building_name varchar(50)
206                 );
207                 v_table_created := true;
208         END IF;
209
210         TRUNCATE TABLE public.pc_paf_sub_building_names;
211         INSERT INTO public.pc_paf_sub_building_names
212         SELECT substring(data,1,8)::integer,
213                 nullif(trim(substring(data,9,30)),'')
214         FROM data_stage;
215         
216         GET DIAGNOSTICS v_processed = ROW_COUNT;
217         
218         IF (v_table_created) THEN
219                 ALTER TABLE public.pc_paf_sub_building_names ADD PRIMARY KEY (sub_building_name_key);
220         END IF;
221         
222         RAISE NOTICE '%: Done importing sub building names (imported % records)', clock_timestamp(), v_processed;       
223         
224         -- 6) Organisations
225         TRUNCATE TABLE data_stage;
226         
227         RAISE NOTICE '%: Begin staging organisations', clock_timestamp();
228         
229         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'org.c01');
230         EXECUTE v_sql;          
231
232         DELETE FROM data_stage WHERE data like '%ORGANISA'  || in_edition ||  '%' OR data like v_std_footer; 
233
234         RAISE NOTICE '%: Done staging organisations, importing', clock_timestamp();
235         
236         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_organisations') THEN
237                 RAISE NOTICE 'Table "public"."pc_paf_organisations" already exists';
238                 v_table_created := false;
239         ELSE
240                 RAISE NOTICE 'Creating table "public"."pc_paf_organisations"';
241                 CREATE TABLE public.pc_paf_organisations (
242                         organisation_key integer NOT NULL,
243                         postcode_type varchar(1) NOT NULL,
244                         organisation_name varchar(60),
245                         department_name varchar(60)
246                 );
247                 v_table_created := true;
248         END IF;
249
250         TRUNCATE TABLE public.pc_paf_organisations;
251         INSERT INTO public.pc_paf_organisations
252         SELECT substring(data,1,8)::integer,
253                 nullif(trim(substring(data,9,1)),''),
254                 nullif(trim(substring(data,10,60)),''),
255                 nullif(trim(substring(data,70,60)),'')
256         FROM data_stage;
257
258         GET DIAGNOSTICS v_processed = ROW_COUNT;
259         
260         IF (v_table_created) THEN
261                 ALTER TABLE public.pc_paf_organisations ADD PRIMARY KEY (organisation_key, postcode_type);
262                 COMMENT ON COLUMN public.pc_paf_organisations.organisation_key IS 'When postcode type is L organisation_key relates to address_key';
263         END IF;
264         
265         RAISE NOTICE '%: Done importing organisations (imported % records)', clock_timestamp(), v_processed;
266         
267         -- 7) Postzon with latlon generated using postgis
268         TRUNCATE TABLE data_stage;
269         
270         RAISE NOTICE '%: Begin staging postzon', clock_timestamp();
271         
272         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_postzon || 'pzone100.c01');
273         EXECUTE v_sql;  
274         
275         DELETE FROM data_stage WHERE data like '%PZONE100' || in_edition || '%' OR data like v_main_footer; 
276         
277         RAISE NOTICE '%: Done staging postzon, importing', clock_timestamp();
278         
279         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_postzon_100m') THEN
280                 RAISE NOTICE 'Table "public"."pc_paf_postzon_100m" already exists';
281                 v_table_created := false;
282         ELSE
283                 RAISE NOTICE 'Creating table "public"."pc_paf_postzon_100m"';
284                 CREATE TABLE public.pc_paf_postzon_100m (
285                         postzon_100m_key serial NOT NULL,
286                         outward_code varchar(4) NOT NULL,
287                         inward_code varchar(3) NOT NULL,
288                         introduction_date date,
289                         grid_reference_east integer,
290                         grid_reference_north integer,
291                         country_code varchar(9),
292                         area_code_county varchar(9),
293                         area_code_district varchar(9),
294                         ward_code varchar(9),
295                         nhs_region varchar(9),
296                         nhs_code varchar(9),
297                         user_type smallint,
298                         grid_status smallint,
299                         latitude double precision,
300                         longitude double precision
301                 );
302                 v_table_created := true;
303         END IF;
304         
305         TRUNCATE TABLE public.pc_paf_postzon_100m;
306         INSERT INTO public.pc_paf_postzon_100m(outward_code, inward_code, introduction_date, grid_reference_east, grid_reference_north,
307         country_code, area_code_county, area_code_district, ward_code, nhs_region, nhs_code, user_type, grid_status, latitude, longitude)
308         SELECT nullif(trim(substring(data,1,4)),'') AS outward_code,
309                 nullif(trim(substring(data,5,3)),'') AS inward_code,
310                 to_date(substring(data,8,6), 'YYYYMM') AS introduction_date,
311                 (nullif(trim(substring(data,14,5)),'') || '0')::integer AS grid_reference_east,         
312                 (CASE WHEN substring(nullif(trim(substring(data,19,5)),''),1,1) ~ '[\x4f-\x5a]' THEN '1' || translate(nullif(trim(substring(data,19,5)),''), 'POUTZY', '221100') ELSE nullif(trim(substring(data,19,5)),'') END || '0')::integer AS grid_reference_north,
313                 nullif(trim(substring(data,24,9)),'') AS country_code,
314                 nullif(trim(substring(data,33,9)),'') AS area_code_county,
315                 nullif(trim(substring(data,42,9)),'') AS area_code_district,
316                 nullif(trim(substring(data,51,9)),'') AS ward_code,
317                 nullif(trim(substring(data,60,9)),'') AS nhs_region,
318                 nullif(trim(substring(data,69,9)),'') AS nhs_code,
319                 nullif(trim(substring(data,78,1)),'')::smallint AS user_type,
320                 nullif(trim(substring(data,79,1)),'')::smallint AS grid_status,
321                 ST_y(ST_transform(ST_GeomFromText('POINT('||(nullif(trim(substring(data,14,5)),'') || '0')||' '||(CASE WHEN substring(nullif(trim(substring(data,19,5)),''),1,1) ~ '[\x4f-\x5a]' THEN '1' || translate(nullif(trim(substring(data,19,5)),''), 'POUTZY', '221100') ELSE nullif(trim(substring(data,19,5)),'') END || '0')||')',CASE WHEN nullif(trim(substring(data,1,4)),'') LIKE 'BT%' THEN 29903 ELSE 27700 END),4326))::numeric(8,6) AS latitude,
322                 ST_x(ST_transform(ST_GeomFromText('POINT('||(nullif(trim(substring(data,14,5)),'') || '0')||' '||(CASE WHEN substring(nullif(trim(substring(data,19,5)),''),1,1) ~ '[\x4f-\x5a]' THEN '1' || translate(nullif(trim(substring(data,19,5)),''), 'POUTZY', '221100') ELSE nullif(trim(substring(data,19,5)),'') END || '0')||')',CASE WHEN nullif(trim(substring(data,1,4)),'') LIKE 'BT%' THEN 29903 ELSE 27700 END),4326))::numeric(8,6) AS longitude
323         FROM data_stage;
324
325         GET DIAGNOSTICS v_processed = ROW_COUNT;
326         
327         IF (v_table_created) THEN
328                 ALTER TABLE public.pc_paf_postzon_100m ADD PRIMARY KEY (postzon_100m_key),
329                 ADD CONSTRAINT pc_paf_postzon_100m_unique UNIQUE (outward_code, inward_code);
330                 CREATE INDEX pc_paf_postzon_100m_longitude ON public.pc_paf_postzon_100m (longitude);
331                 CREATE INDEX pc_paf_postzon_100m_latitude ON public.pc_paf_postzon_100m (latitude);
332                 
333                 COMMENT ON TABLE public.pc_paf_postzon_100m IS 'Geographical data from the Royal Mail with accuracy of 100m';
334                 COMMENT ON COLUMN public.pc_paf_postzon_100m.grid_reference_east IS 'BT postcodes reflect the Irish Grid public (different origin and scale); subtract 17000 to approximate onto OS National Grid. For latlong do not approximate but do proper conversion';
335                 COMMENT ON COLUMN public.pc_paf_postzon_100m.grid_reference_north IS 'BT postcodes reflect the Irish Grid public (different origin and scale); add 13000 to approximate onto OS National Grid. For latlong do not approximate but do proper conversion';
336                 COMMENT ON COLUMN public.pc_paf_postzon_100m.user_type IS '0 Small User 1 Large User';
337                 COMMENT ON COLUMN public.pc_paf_postzon_100m.grid_status IS '0  Status not supplied by OS \n
338                         1  Within the building of the matched address closest to the Postcode mean. \n
339                         2  Co-ordinates allocated by GROS during Postcode boundary creation to the building nearest the centre of the populated part of the Postcode (Scotland only) \n
340                         3  Approximate to within 50m of true position \n
341                         4  Postcode unit mean (direct copy from ADDRESS-POINT (GB) and COMPAS (NI) - mean of matched addresses with the same Postcode) \n
342                         5  Postcode imputed by ONS to 1 metre resolution \n
343                         6  Postcode sector mean - mainly PO Boxes \n
344                         9  No co-ordinates available';
345         END IF;
346         
347         RAISE NOTICE '%: Done importing postzon (imported % records)', clock_timestamp(), v_processed;  
348         
349         -- 8) Mainfile
350         TRUNCATE TABLE data_stage;
351         RAISE NOTICE '%: Begin staging mainfile', clock_timestamp();
352         
353         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c02');
354         EXECUTE v_sql;
355         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c03');
356         EXECUTE v_sql;
357         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c04');
358         EXECUTE v_sql;
359         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c05');
360         EXECUTE v_sql;
361         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c06');
362         EXECUTE v_sql;
363         
364         DELETE FROM data_stage WHERE data like '%ADDRESS '  || in_edition ||  '%' OR data like v_main_footer;
365
366         RAISE NOTICE '%: Done staging mainfile, importing', clock_timestamp();
367         
368         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_mainfile') THEN
369            RAISE NOTICE 'Table "public"."pc_paf_mainfile" already exists';
370            v_table_created := false;
371         ELSE
372                 RAISE NOTICE 'Creating table "public"."pc_paf_mainfile"';
373                 CREATE TABLE public.pc_paf_mainfile (
374                         paf_record_key serial NOT NULL,
375                         outward_code varchar(4) NOT NULL,
376                         inward_code varchar(3) NOT NULL,
377                         address_key integer NOT NULL,
378                         locality_key integer,
379                         thoroughfare_key integer,
380                         thoroughfare_descriptor_key integer,
381                         dependent_thoroughfare_key integer,
382                         dependent_thoroughfare_descriptor_key integer,
383                         building_number integer,
384                         building_name_key integer,
385                         sub_building_name_key integer,
386                         number_of_households integer,
387                         organisation_key integer,
388                         postcode_type varchar(1),
389                         concatenation_indicator varchar(1),
390                         delivery_point_suffix varchar(2),
391                         small_user_organisation_indicator varchar(1),
392                         po_box_number varchar(6)
393                 );
394                 v_table_created := true;
395         END IF;
396         
397         TRUNCATE TABLE public.pc_paf_mainfile RESTART IDENTITY;
398         INSERT INTO public.pc_paf_mainfile(outward_code,inward_code,address_key,locality_key,thoroughfare_key,
399                 thoroughfare_descriptor_key,dependent_thoroughfare_key,dependent_thoroughfare_descriptor_key,building_number,
400                 building_name_key,sub_building_name_key,number_of_households,organisation_key,postcode_type,concatenation_indicator,
401                 delivery_point_suffix,small_user_organisation_indicator,po_box_number)
402         SELECT nullif(trim(substring(data,1,4)),''),
403                 nullif(trim(substring(data,5,3)),''),
404                 nullif(substring(data,8,8)::integer,0),
405                 nullif(substring(data,16,6)::integer,0),
406                 nullif(substring(data,22,8)::integer,0),
407                 nullif(substring(data,30,4)::integer,0),
408                 nullif(substring(data,34,8)::integer,0),
409                 nullif(substring(data,42,4)::integer,0),
410                 nullif(substring(data,46,4)::integer,0),
411                 nullif(substring(data,50,8)::integer,0),
412                 nullif(substring(data,58,8)::integer,0),
413                 nullif(substring(data,66,4)::integer,0),
414                 nullif(substring(data,70,8)::integer,0),
415                 nullif(trim(substring(data,78,1)),''),
416                 nullif(trim(substring(data,79,1)),''),
417                 nullif(trim(substring(data,80,2)),''),
418                 nullif(trim(substring(data,82,1)),''),
419                 nullif(trim(substring(data,83,6)),'')
420         FROM data_stage;
421         
422         GET DIAGNOSTICS v_processed = ROW_COUNT;
423         
424         IF (v_table_created) THEN
425                 -- Not all of the indexes below are required, however without these the update_pc_paf function will
426                 -- take a while to commit as the deferred foreign key constraints are checked
427                 CREATE INDEX pc_paf_mainfile_postcode ON public.pc_paf_mainfile USING btree (outward_code, inward_code);
428                 CREATE INDEX pc_paf_mainfile_building_name_key ON public.pc_paf_mainfile USING BTREE (building_name_key);
429                 CREATE INDEX pc_paf_mainfile_locality_key ON public.pc_paf_mainfile USING BTREE (locality_key);
430                 CREATE INDEX pc_paf_mainfile_organisation_key ON public.pc_paf_mainfile USING BTREE (organisation_key, postcode_type);
431                 CREATE INDEX pc_paf_mainfile_sub_building_name_key ON public.pc_paf_mainfile USING BTREE (sub_building_name_key);
432                 CREATE INDEX pc_paf_mainfile_thoroughfare_key ON public.pc_paf_mainfile USING BTREE (thoroughfare_key);
433                 
434                 ALTER TABLE public.pc_paf_mainfile 
435                 ADD PRIMARY KEY (paf_record_key),
436                 ADD CONSTRAINT pc_paf_mainfile_unique UNIQUE (address_key, organisation_key, postcode_type),
437                 ADD FOREIGN KEY (locality_key) REFERENCES public.pc_paf_localities(locality_key) DEFERRABLE INITIALLY IMMEDIATE,
438                 ADD FOREIGN KEY (thoroughfare_key) REFERENCES public.pc_paf_thoroughfares(thoroughfare_key) DEFERRABLE INITIALLY IMMEDIATE, 
439                 ADD FOREIGN KEY (thoroughfare_descriptor_key) REFERENCES public.pc_paf_thoroughfares(thoroughfare_key) DEFERRABLE INITIALLY IMMEDIATE,
440                 ADD FOREIGN KEY (building_name_key) REFERENCES public.pc_paf_building_names(building_name_key) DEFERRABLE INITIALLY IMMEDIATE,
441                 ADD FOREIGN KEY (sub_building_name_key) REFERENCES public.pc_paf_sub_building_names(sub_building_name_key) DEFERRABLE INITIALLY IMMEDIATE,
442                 ADD FOREIGN KEY (organisation_key, postcode_type) REFERENCES public.pc_paf_organisations(organisation_key, postcode_type) DEFERRABLE INITIALLY IMMEDIATE,
443                 ADD FOREIGN KEY (outward_code,inward_code) REFERENCES public.pc_paf_postzon_100m(outward_code,inward_code) DEFERRABLE INITIALLY IMMEDIATE;
444                 COMMENT ON COLUMN public.pc_paf_mainfile.organisation_key IS 'When postcode type is L then address_key relates to organisation_key - good work RM!';
445         END IF;
446         
447         RAISE NOTICE '%: Done importing mainfile (imported % records)', clock_timestamp(), v_processed;
448         
449         -- 9) Welsh Mainfile
450         TRUNCATE TABLE data_stage;
451         RAISE NOTICE '%: Begin staging welsh alternative mainfile', clock_timestamp();
452         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'wfmainfl.c06');
453         EXECUTE v_sql;
454         
455         DELETE FROM data_stage WHERE data like '%ADDRESS '  || in_edition ||  '%' OR data like v_main_footer;
456         
457         RAISE NOTICE '%: Done staging welsh alternative mainfile, importing', clock_timestamp();
458
459         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_mainfile_welsh') THEN
460            RAISE NOTICE 'Table "public"."pc_paf_mainfile_welsh" already exists';
461            v_table_created := false;
462         ELSE
463                 RAISE NOTICE 'Creating table "public"."pc_paf_mainfile_welsh"';
464                 CREATE TABLE public.pc_paf_mainfile_welsh (
465                         paf_record_key serial NOT NULL,
466                         outward_code varchar(4) NOT NULL,
467                         inward_code varchar(3) NOT NULL,
468                         address_key integer NOT NULL,
469                         locality_key integer,
470                         thoroughfare_key integer,
471                         thoroughfare_descriptor_key integer,
472                         dependent_thoroughfare_key integer,
473                         dependent_thoroughfare_descriptor_key integer,
474                         building_number integer,
475                         building_name_key integer,
476                         sub_building_name_key integer,
477                         number_of_households integer,
478                         organisation_key integer,
479                         postcode_type varchar(1),
480                         concatenation_indicator varchar(1),
481                         delivery_point_suffix varchar(2),
482                         small_user_organisation_indicator varchar(1),
483                         po_box_number varchar(6)
484                 );
485                 v_table_created := true;
486         END IF;
487         
488         TRUNCATE TABLE public.pc_paf_mainfile_welsh RESTART IDENTITY;
489                 INSERT INTO public.pc_paf_mainfile_welsh(outward_code,inward_code,address_key,locality_key,thoroughfare_key,
490                         thoroughfare_descriptor_key,dependent_thoroughfare_key,dependent_thoroughfare_descriptor_key,building_number,
491                         building_name_key,sub_building_name_key,number_of_households,organisation_key,postcode_type,concatenation_indicator,
492                         delivery_point_suffix,small_user_organisation_indicator,po_box_number)
493                 SELECT nullif(trim(substring(data,1,4)),''),
494                 nullif(trim(substring(data,5,3)),''),
495                 nullif(substring(data,8,8)::integer,0),
496                 nullif(substring(data,16,6)::integer,0),
497                 nullif(substring(data,22,8)::integer,0),
498                 nullif(substring(data,30,4)::integer,0),
499                 nullif(substring(data,34,8)::integer,0),
500                 nullif(substring(data,42,4)::integer,0),
501                 nullif(substring(data,46,4)::integer,0),
502                 nullif(substring(data,50,8)::integer,0),
503                 nullif(substring(data,58,8)::integer,0),
504                 nullif(substring(data,66,4)::integer,0),
505                 nullif(substring(data,70,8)::integer,0),
506                 nullif(trim(substring(data,78,1)),''),
507                 nullif(trim(substring(data,79,1)),''),
508                 nullif(trim(substring(data,80,2)),''),
509                 nullif(trim(substring(data,82,1)),''),
510                 nullif(trim(substring(data,83,6)),'')
511         FROM data_stage;
512         
513         GET DIAGNOSTICS v_processed = ROW_COUNT;
514         
515         IF (v_table_created) THEN
516                 -- As above, these are required for faster commit on update_pc_paf
517                 CREATE INDEX pc_paf_mainfile_welsh_postcode ON public.pc_paf_mainfile_welsh USING btree (outward_code, inward_code);
518                 CREATE INDEX pc_paf_mainfile_welsh_building_name_key ON public.pc_paf_mainfile_welsh USING BTREE (building_name_key);
519                 CREATE INDEX pc_paf_mainfile_welsh_locality_key ON public.pc_paf_mainfile_welsh USING BTREE (locality_key);
520                 CREATE INDEX pc_paf_mainfile_welsh_organisation_key ON public.pc_paf_mainfile_welsh USING BTREE (organisation_key, postcode_type);
521                 CREATE INDEX pc_paf_mainfile_welsh_sub_building_name_key ON public.pc_paf_mainfile_welsh USING BTREE (sub_building_name_key);
522                 CREATE INDEX pc_paf_mainfile_welsh_thoroughfare_key ON public.pc_paf_mainfile_welsh USING BTREE (thoroughfare_key);
523                 
524                 ALTER TABLE public.pc_paf_mainfile_welsh 
525                 ADD PRIMARY KEY (paf_record_key),
526                 ADD CONSTRAINT pc_paf_mainfile_welsh_unique UNIQUE (address_key, organisation_key, postcode_type),
527                 ADD FOREIGN KEY (locality_key) REFERENCES public.pc_paf_localities(locality_key) DEFERRABLE INITIALLY IMMEDIATE,
528                 ADD FOREIGN KEY (thoroughfare_key) REFERENCES public.pc_paf_thoroughfares(thoroughfare_key) DEFERRABLE INITIALLY IMMEDIATE, 
529                 ADD FOREIGN KEY (thoroughfare_descriptor_key) REFERENCES public.pc_paf_thoroughfares(thoroughfare_key) DEFERRABLE INITIALLY IMMEDIATE,
530                 ADD FOREIGN KEY (building_name_key) REFERENCES public.pc_paf_building_names(building_name_key) DEFERRABLE INITIALLY IMMEDIATE,
531                 ADD FOREIGN KEY (sub_building_name_key) REFERENCES public.pc_paf_sub_building_names(sub_building_name_key) DEFERRABLE INITIALLY IMMEDIATE,
532                 ADD FOREIGN KEY (organisation_key, postcode_type) REFERENCES public.pc_paf_organisations(organisation_key, postcode_type) DEFERRABLE INITIALLY IMMEDIATE,
533                 ADD FOREIGN KEY (outward_code,inward_code) REFERENCES public.pc_paf_postzon_100m(outward_code,inward_code) DEFERRABLE INITIALLY IMMEDIATE;
534                 COMMENT ON COLUMN public.pc_paf_mainfile_welsh.organisation_key IS 'When postcode type is L then address_key relates to organisation_key - good work RM!';
535         END IF;
536         
537         RAISE NOTICE '%: Done importing welsh alternative mainfile (imported % records)', clock_timestamp(), v_processed;       
538
539         -- 10) Alias file
540         TRUNCATE TABLE data_stage;
541         RAISE NOTICE '%: Begin staging alias file', clock_timestamp();
542         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_alias || 'aliasfle.c01');
543         EXECUTE v_sql;
544         
545         DELETE FROM data_stage WHERE data like '%ALIASFLE'  || in_edition ||  '%' OR data like v_std_footer;
546         
547         RAISE NOTICE '%: Done staging alias file, importing', clock_timestamp();        
548
549         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_counties') THEN
550                 RAISE NOTICE 'Table "public"."pc_paf_counties" already exists';
551                 v_table_created := false;
552         ELSE
553                 RAISE NOTICE 'Creating table "public"."pc_paf_counties"';
554                 CREATE TABLE public.pc_paf_counties (
555                         county_key integer NOT NULL,
556                         county_name varchar(30),
557                         county_type varchar(1)
558                 );
559                 v_table_created := true;
560         END IF;
561
562         TRUNCATE TABLE public.pc_paf_counties;
563         INSERT INTO public.pc_paf_counties
564         SELECT substring(data,2,4)::integer AS county_key,
565                 trim(substring(data,6,30)) AS county_name,
566                 trim(substring(data,36,1)) AS county_type
567         FROM data_stage
568         WHERE substring(data,1,1)::integer = 4;
569         
570         GET DIAGNOSTICS v_processed = ROW_COUNT;
571                 
572         IF (v_table_created) THEN
573                 ALTER TABLE public.pc_paf_counties ADD PRIMARY KEY (county_key),
574                 ADD CONSTRAINT pc_paf_counties_unique UNIQUE (county_name, county_type);
575                 COMMENT ON COLUMN public.pc_paf_counties.county_type IS 'T (Traditional County), P (Former Postal County) or A (Administrative County)';
576         END IF;
577         
578         RAISE NOTICE '%: Done importing counties (imported % records)', clock_timestamp(), v_processed; 
579         
580         IF EXISTS (SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' AND tablename = 'pc_paf_county_alias') THEN
581                 RAISE NOTICE 'Table "public"."pc_paf_county_alias" already exists';
582                 v_table_created := false;
583         ELSE
584                 RAISE NOTICE 'Creating table "public"."pc_paf_county_alias"';
585                 CREATE TABLE public.pc_paf_county_alias (
586                         county_alias_key serial NOT NULL,
587                         postcode varchar(7) NOT NULL,
588                         former_postal_county integer,
589                         traditional_county integer,
590                         administrative_county integer
591                 );
592                 v_table_created := true;
593         END IF; 
594         
595         TRUNCATE TABLE public.pc_paf_county_alias;
596         INSERT INTO public.pc_paf_county_alias (postcode, former_postal_county, traditional_county, administrative_county)
597         SELECT trim(substring(data,2,7)) AS postcode,
598                 nullif(substring(data,9,4)::integer,0) AS former_postal_county,
599                 nullif(substring(data,13,4)::integer,0) AS traditional_county,
600                 nullif(substring(data,17,4)::integer,0) AS administrative_county
601         FROM data_stage
602         WHERE substring(data,1,1)::integer = 5;
603         
604         GET DIAGNOSTICS v_processed = ROW_COUNT;
605
606         IF (v_table_created) THEN
607                 ALTER TABLE public.pc_paf_county_alias ADD PRIMARY KEY (county_alias_key),
608                 ADD CONSTRAINT pc_paf_county_alias_unique UNIQUE (postcode),
609                 ADD FOREIGN KEY (former_postal_county) REFERENCES public.pc_paf_counties(county_key) DEFERRABLE INITIALLY IMMEDIATE,
610                 ADD FOREIGN KEY (traditional_county) REFERENCES public.pc_paf_counties(county_key) DEFERRABLE INITIALLY IMMEDIATE,
611                 ADD FOREIGN KEY (administrative_county) REFERENCES public.pc_paf_counties(county_key) DEFERRABLE INITIALLY IMMEDIATE;
612         END IF;
613         
614         RAISE NOTICE '%: Done importing county_alias (imported % records)', clock_timestamp(), v_processed;     
615
616         RAISE NOTICE '%: Completed', clock_timestamp();
617         RETURN true;    
618 END;
619
620 $BODY$
621 LANGUAGE 'plpgsql' VOLATILE;
622
623 --
624
625 DROP FUNCTION IF EXISTS public.update_pc_paf(varchar, varchar);
626 CREATE OR REPLACE FUNCTION public.update_pc_paf(in_edition varchar, in_data varchar)
627 RETURNS boolean AS
628 $BODY$
629
630 DECLARE
631         v_data_root varchar;
632         v_data_update varchar;
633         v_data_postzon varchar;
634         v_data_alias varchar;
635         v_processed integer;
636         v_main_footer varchar;
637         v_std_footer varchar;
638         v_sql varchar;
639         v_table_created boolean;
640 BEGIN
641         v_data_root := in_data || '/' || in_edition || '_CHANGES/';
642         v_data_update := v_data_root || 'CONSOLIDATED CHANGES/';
643         v_data_postzon := v_data_root || 'POSTZON 100M/';
644         v_data_alias := v_data_root || 'ALIAS/';
645         
646         v_main_footer = '       %';
647         
648         RAISE NOTICE '%: Import starting for edition % with data root %', clock_timestamp(), in_edition, v_data_root;
649         CREATE TEMPORARY TABLE data_stage (data text) ON COMMIT DROP;
650
651         --changes1.c01 - Changes1 (Changes to satelite tables except Organisations) - A single changes file
652         RAISE NOTICE '%: Begin staging Changes1 file', clock_timestamp();
653
654         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_update || 'changes1.c01');
655         EXECUTE v_sql;  
656         DELETE FROM data_stage WHERE data like '%CHANGES1' || in_edition || '%' OR data like v_main_footer;
657         
658         RAISE NOTICE '%: Done staging Changes1 file',clock_timestamp();
659         
660         SET CONSTRAINTS ALL DEFERRED;
661                 
662         --Record Type 1  - Localities
663         RAISE NOTICE '%: Preparing to update localities', clock_timestamp();
664         
665         CREATE TEMPORARY TABLE tmp_localities ON COMMIT DROP AS
666         SELECT substring(data,2,8)::integer AS locality_key,
667                 to_timestamp(nullif(substring(data,10,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
668                 nullif(trim(substring(data,24,1)),'') AS amendment_type,
669                 nullif(trim(substring(data,70,30)),'') AS post_town,
670                 nullif(trim(substring(data,100,35)),'') AS dependent_locality,
671                 nullif(trim(substring(data,135,35)),'') AS double_dependent_locality
672         FROM data_stage 
673         WHERE substring(data,1,1)::integer = 1 
674         AND nullif(trim(substring(data,24,1)),'') IS NOT NULL
675         ORDER BY to_timestamp(nullif(substring(data,10,14),'00000000000000'), 'YYYYMMDDHH24MISS');
676         
677         GET DIAGNOSTICS v_processed = ROW_COUNT;
678         RAISE NOTICE '%: Prepared % records for update on localities',clock_timestamp(), v_processed;
679         
680         DELETE FROM public.pc_paf_localities l
681         USING tmp_localities lt WHERE lt.locality_key = l.locality_key
682         AND lt.amendment_type IN ('D', 'B')
683         AND l.* = (lt.locality_key, lt.post_town::character varying(30), lt.dependent_locality::character varying(30), lt.double_dependent_locality::character varying(30));
684         
685         GET DIAGNOSTICS v_processed = ROW_COUNT;
686         RAISE NOTICE '%: Removed % records from localities',clock_timestamp(), v_processed;
687         
688         INSERT INTO public.pc_paf_localities
689         SELECT lt.locality_key, lt.post_town, lt.dependent_locality, lt.double_dependent_locality
690         FROM tmp_localities lt
691         WHERE lt.amendment_type IN ('I', 'C');
692         
693         GET DIAGNOSTICS v_processed = ROW_COUNT;
694         RAISE NOTICE '%: Created % records in localities',clock_timestamp(), v_processed;
695         
696         --Record Type 2  - Thoroughfares
697         RAISE NOTICE '%: Preparing to update thoroughfares', clock_timestamp();
698         
699         CREATE TEMPORARY TABLE tmp_thoroughfares ON COMMIT DROP AS
700         SELECT substring(data,2,8)::integer AS thoroughfare_key,
701                         to_timestamp(nullif(substring(data,10,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
702                         nullif(trim(substring(data,24,1)),'') AS amendment_type,
703                         nullif(trim(substring(data,25,60)),'') AS thoroughfare_name
704         FROM data_stage 
705         WHERE substring(data,1,1)::integer = 2 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
706         
707         GET DIAGNOSTICS v_processed = ROW_COUNT;
708         RAISE NOTICE '%: Prepared % records for update on thoroughfares',clock_timestamp(), v_processed;
709         
710         DELETE FROM public.pc_paf_thoroughfares t
711         USING tmp_thoroughfares tt WHERE tt.thoroughfare_key = t.thoroughfare_key
712         AND tt.amendment_type IN ('D', 'B')
713         AND t.* = (tt.thoroughfare_key, t.thoroughfare_name::character varying(60));
714         
715         GET DIAGNOSTICS v_processed = ROW_COUNT;
716         RAISE NOTICE '%: Removed % records from thoroughfares',clock_timestamp(), v_processed;
717         
718         INSERT INTO public.pc_paf_thoroughfares
719         SELECT tt.thoroughfare_key, tt.thoroughfare_name
720         FROM tmp_thoroughfares tt
721         WHERE tt.amendment_type IN ('I', 'C');
722         
723         GET DIAGNOSTICS v_processed = ROW_COUNT;
724         RAISE NOTICE '%: Created % records in thoroughfares',clock_timestamp(), v_processed;
725         
726         --Record Type 3  - Thoroughfare Descriptors 
727         RAISE NOTICE '%: Preparing to update thoroughfare_descriptor', clock_timestamp();
728         
729         CREATE TEMPORARY TABLE tmp_thoroughfare_descriptor ON COMMIT DROP AS
730         SELECT substring(data,2,8)::integer AS thoroughfare_descriptor_key ,
731                 to_timestamp(nullif(substring(data,10,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
732                 nullif(trim(substring(data,24,1)),'') AS amendment_type,
733                 nullif(trim(substring(data,25,20)),'') AS thoroughfare_descriptor,
734                 nullif(trim(substring(data,45,6)),'') AS approved_abbreviation
735         FROM data_stage 
736         WHERE substring(data,1,1)::integer = 3 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
737         
738         GET DIAGNOSTICS v_processed = ROW_COUNT;
739         RAISE NOTICE '%: Prepared % records for update on thoroughfare_descriptor',clock_timestamp(), v_processed;
740         
741         DELETE FROM public.pc_paf_thoroughfare_descriptor td
742         USING tmp_thoroughfare_descriptor tdt WHERE tdt.thoroughfare_descriptor_key  = td.thoroughfare_descriptor_key 
743         AND tdt.amendment_type IN ('D', 'B')
744         AND td.* = (tdt.thoroughfare_descriptor_key , tdt.thoroughfare_descriptor::character varying(20), tdt.approved_abbreviation::character varying(6));
745         
746         GET DIAGNOSTICS v_processed = ROW_COUNT;
747         RAISE NOTICE '%: Removed % records from thoroughfare_descriptor',clock_timestamp(), v_processed;
748         
749         INSERT INTO public.pc_paf_thoroughfare_descriptor
750         SELECT tdt.thoroughfare_descriptor_key , tdt.thoroughfare_descriptor, tdt.approved_abbreviation
751         FROM tmp_thoroughfare_descriptor tdt
752         WHERE tdt.amendment_type IN ('I', 'C');
753         
754         GET DIAGNOSTICS v_processed = ROW_COUNT;
755         RAISE NOTICE '%: Created % records in thoroughfare_descriptor',clock_timestamp(), v_processed;
756         
757         --Record Type 4  - Building Names 
758         RAISE NOTICE '%: Preparing to update building_names', clock_timestamp();
759         
760         CREATE TEMPORARY TABLE tmp_building_names ON COMMIT DROP AS
761         SELECT substring(data,2,8)::integer AS building_name_key,
762                 to_timestamp(nullif(substring(data,10,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
763                 nullif(trim(substring(data,24,1)),'') AS amendment_type,
764                 nullif(trim(substring(data,25,50)),'') AS building_name
765         FROM data_stage 
766         WHERE substring(data,1,1)::integer = 4 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
767
768         GET DIAGNOSTICS v_processed = ROW_COUNT;
769         RAISE NOTICE '%: Prepared % records for update on building_names',clock_timestamp(), v_processed;
770         
771         DELETE FROM public.pc_paf_building_names bn
772         USING tmp_building_names tbn WHERE bn.building_name_key = tbn.building_name_key
773         AND tbn.amendment_type IN ('D', 'B')
774         AND bn.* = (tbn.building_name_key, tbn.building_name::character varying(50));
775         
776         GET DIAGNOSTICS v_processed = ROW_COUNT;
777         RAISE NOTICE '%: Removed % records from building_names',clock_timestamp(), v_processed;
778         
779         INSERT INTO public.pc_paf_building_names
780         SELECT tbn.building_name_key, tbn.building_name
781         FROM tmp_building_names tbn
782         WHERE tbn.amendment_type IN ('I', 'C');
783         
784         GET DIAGNOSTICS v_processed = ROW_COUNT;
785         RAISE NOTICE '%: Created % records in building_names',clock_timestamp(), v_processed;
786         
787         --Record Type 5  - Sub Building Names 
788         RAISE NOTICE '%: Preparing to update sub_building_names', clock_timestamp();
789         
790         CREATE TEMPORARY TABLE tmp_sub_building_names ON COMMIT DROP AS
791         SELECT substring(data,2,8)::integer AS sub_building_name_key,
792                 to_timestamp(nullif(substring(data,10,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
793                 nullif(trim(substring(data,24,1)),'') AS amendment_type,
794                 nullif(trim(substring(data,25,30)),'') AS sub_building_name
795         FROM data_stage 
796         WHERE substring(data,1,1)::integer = 5 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
797         
798         GET DIAGNOSTICS v_processed = ROW_COUNT;
799         RAISE NOTICE '%: Prepared % records for update on sub_building_names',clock_timestamp(), v_processed;
800         
801         DELETE FROM public.pc_paf_sub_building_names sbn
802         USING tmp_sub_building_names tsbn WHERE sbn.sub_building_name_key = tsbn.sub_building_name_key
803         AND tsbn.amendment_type IN ('D', 'B')
804         AND sbn.* = (tsbn.sub_building_name_key, tsbn.sub_building_name::character varying(50));
805         
806         GET DIAGNOSTICS v_processed = ROW_COUNT;
807         RAISE NOTICE '%: Removed % records from sub_building_names',clock_timestamp(), v_processed;
808         
809         INSERT INTO public.pc_paf_sub_building_names
810         SELECT tsbn.sub_building_name_key, tsbn.sub_building_name
811         FROM tmp_sub_building_names tsbn
812         WHERE tsbn.amendment_type IN ('I', 'C');
813         
814         GET DIAGNOSTICS v_processed = ROW_COUNT;
815         RAISE NOTICE '%: Created % records in sub_building_names',clock_timestamp(), v_processed;
816         
817         TRUNCATE TABLE data_stage;
818         
819         -- fpchgsng.c01 - Changes2 (Changes to Mainfile and Organisations) -- fpchgsng.c01 = Single changes file, fpchngs2.c01 = Timeline changes file
820         RAISE NOTICE '%: Begin staging Changes2 file', clock_timestamp();
821         
822         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_update || 'fpchgsng.c01');
823         EXECUTE v_sql;  
824         DELETE FROM data_stage WHERE data like '%CHANGES2' || in_edition || '%' OR data like v_main_footer;
825         
826         RAISE NOTICE '%: Done staging Changes2 file',clock_timestamp();
827         
828         -- Mainfile
829         RAISE NOTICE '%: Preparing to update mainfile', clock_timestamp();
830         
831         CREATE TEMPORARY TABLE tmp_mainfile ON COMMIT DROP AS
832         SELECT to_timestamp(nullif(substring(data,1,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
833                 nullif(trim(substring(data,15,4)),'') AS outward_code,
834                 nullif(trim(substring(data,19,3)),'') AS inward_code,
835                 nullif(trim(substring(data,22,2)),'') AS delivery_point_suffix,
836                 nullif(trim(substring(data,24,1)),'') AS postcode_type,
837                 nullif(substring(data,25,8)::integer,0) AS address_key,
838                 nullif(substring(data,33,8)::integer,0) AS organisation_key,
839                 nullif(trim(substring(data,41,1)),'') AS amendment_type,
840                 nullif(substring(data,42,1)::integer,0) AS record_type,
841                 nullif(substring(data,43,8)::integer,0) AS locality_key,
842                 nullif(substring(data,51,8)::integer,0) AS thoroughfare_key,
843                 nullif(substring(data,59,8)::integer,0) AS thoroughfare_descriptor_key,
844                 nullif(substring(data,67,8)::integer,0) AS dependent_thoroughfare_key,
845                 nullif(substring(data,75,8)::integer,0) AS dependent_thoroughfare_descriptor_key,
846                 nullif(substring(data,83,4)::integer,0) AS building_number,
847                 nullif(substring(data,87,8)::integer,0) AS building_name_key,
848                 nullif(substring(data,95,8)::integer,0) AS sub_building_name_key,
849                 nullif(substring(data,103,4)::integer,0) AS number_of_households,
850                 nullif(trim(substring(data,107,1)),'') AS concatenation_indicator,
851                 nullif(trim(substring(data,108,6)),'') AS po_box_number,
852                 nullif(trim(substring(data,114,1)),'') AS small_user_organisation_indicator,
853                 CASE nullif(substring(data,115,2)::integer,0) 
854                         WHEN 1 THEN 'New' 
855                         WHEN 2 THEN 'Correction'
856                         WHEN 3 THEN '' 
857                         WHEN 4 THEN 'Coding Revision'
858                         WHEN 5 THEN 'Organisation Change'
859                         WHEN 6 THEN 'Status Change'
860                         WHEN 7 THEN 'Large User Deleted'
861                         WHEN 8 THEN 'Building/Sub Building Change'
862                         WHEN 9 THEN 'Large User Change'
863                 END AS reason_for_amendment,            
864                 nullif(trim(substring(data,117,4)),'') AS new_outward_code,
865                 nullif(trim(substring(data,121,3)),'') AS new_inward_code
866         FROM data_stage 
867         WHERE nullif(substring(data,42,1)::integer,0) = 1;
868         
869         GET DIAGNOSTICS v_processed = ROW_COUNT;
870         RAISE NOTICE '%: Prepared % records for update on mainfile',clock_timestamp(), v_processed;
871         
872         DELETE FROM public.pc_paf_mainfile m
873         USING tmp_mainfile tm WHERE tm.address_key = m.address_key AND tm.organisation_key IS NOT DISTINCT FROM m.organisation_key AND tm.postcode_type = m.postcode_type
874         AND tm.amendment_type IN ('D', 'B');
875         
876         GET DIAGNOSTICS v_processed = ROW_COUNT;
877         RAISE NOTICE '%: Removed % records from mainfile',clock_timestamp(), v_processed;
878         
879         INSERT INTO public.pc_paf_mainfile (outward_code, inward_code, address_key, locality_key, thoroughfare_key, 
880                 thoroughfare_descriptor_key, dependent_thoroughfare_key, dependent_thoroughfare_descriptor_key, 
881                 building_number, building_name_key, sub_building_name_key, number_of_households, 
882                 organisation_key, postcode_type, concatenation_indicator, delivery_point_suffix, 
883                 small_user_organisation_indicator, po_box_number)
884         SELECT COALESCE(tm.new_outward_code, tm.outward_code), COALESCE(tm.new_inward_code, tm.inward_code), tm.address_key, tm.locality_key, tm.thoroughfare_key, 
885                 tm.thoroughfare_descriptor_key, tm.dependent_thoroughfare_key, tm.dependent_thoroughfare_descriptor_key, 
886                 tm.building_number, tm.building_name_key, tm.sub_building_name_key, tm.number_of_households, 
887                 tm.organisation_key, tm.postcode_type, tm.concatenation_indicator, tm.delivery_point_suffix, 
888                 tm.small_user_organisation_indicator, tm.po_box_number
889         FROM (
890                 SELECT cume_dist() OVER w, *
891                 FROM tmp_mainfile
892                 WINDOW w AS (PARTITION BY address_key, organisation_key, postcode_type ORDER BY timestamp)
893         ) tm WHERE tm.cume_dist = 1 AND tm.amendment_type IN ('I', 'C');
894         
895         
896         GET DIAGNOSTICS v_processed = ROW_COUNT;
897         RAISE NOTICE '%: Created % records in mainfile',clock_timestamp(), v_processed;
898         
899         -- Organisations
900         RAISE NOTICE '%: Preparing to update organisations', clock_timestamp();
901         
902         CREATE TEMPORARY TABLE tmp_organisations ON COMMIT DROP AS
903         SELECT to_timestamp(nullif(substring(data,1,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
904                         nullif(trim(substring(data,15,4)),'') AS outward_code,
905                         nullif(trim(substring(data,19,3)),'') AS inward_code,
906                         nullif(trim(substring(data,22,2)),'') AS delivery_point_suffix,
907                         nullif(trim(substring(data,24,1)),'') AS postcode_type,
908                         nullif(substring(data,25,8)::integer,0) AS address_key,
909                         nullif(substring(data,33,8)::integer,0) AS organisation_key,
910                         nullif(trim(substring(data,41,1)),'') AS amendment_type,
911                         nullif(substring(data,42,1)::integer,0) AS record_type,
912                         nullif(trim(substring(data,43,60)),'') AS organisation_name,
913                         nullif(trim(substring(data,103,60)),'') AS department_name,
914                         nullif(trim(substring(data,163,6)),'') AS po_box_number
915         FROM data_stage 
916         WHERE nullif(substring(data,42,1)::integer,0) IN (2,3); 
917
918         GET DIAGNOSTICS v_processed = ROW_COUNT;
919         RAISE NOTICE '%: Prepared % records for update on organisations',clock_timestamp(), v_processed;
920         
921         DELETE FROM public.pc_paf_organisations o
922         USING tmp_organisations tx WHERE COALESCE(tx.organisation_key, tx.address_key) = o.organisation_key AND tx.postcode_type = o.postcode_type
923         AND tx.amendment_type IN ('D', 'B');
924
925         GET DIAGNOSTICS v_processed = ROW_COUNT;
926         RAISE NOTICE '%: Removed % records from organisations',clock_timestamp(), v_processed;
927         
928         INSERT INTO public.pc_paf_organisations
929         SELECT COALESCE(tx.organisation_key, tx.address_key), tx.postcode_type, tx.organisation_name, tx.department_name
930         FROM (
931                 SELECT cume_dist() OVER w, *
932                 FROM tmp_organisations
933                 WINDOW w AS (PARTITION BY COALESCE(organisation_key, address_key), postcode_type ORDER BY timestamp)
934         ) tx WHERE tx.cume_dist = 1 AND tx.amendment_type IN ('I', 'C');
935         
936
937         GET DIAGNOSTICS v_processed = ROW_COUNT;
938         RAISE NOTICE '%: Created % records in organisations',clock_timestamp(), v_processed;
939         
940         -- wchanges.c01  -- Welsh changes (In same format as regular changes2 file (hence has timelined data, not single changes)
941         TRUNCATE TABLE data_stage;
942         
943         RAISE NOTICE '%: Begin staging WChanges file', clock_timestamp();
944         
945         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_update || 'wchanges.c01');
946         EXECUTE v_sql;  
947         DELETE FROM data_stage WHERE data like '%WCHANGES' || in_edition || '%' OR data like v_main_footer;
948         
949         RAISE NOTICE '%: Done staging WChanges file',clock_timestamp();
950         
951         RAISE NOTICE '%: Preparing to update welsh mainfile', clock_timestamp();
952
953         CREATE TEMPORARY TABLE tmp_mainfile_welsh ON COMMIT DROP AS
954         SELECT to_timestamp(nullif(substring(data,1,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
955                 nullif(trim(substring(data,15,4)),'') AS outward_code,
956                 nullif(trim(substring(data,19,3)),'') AS inward_code,
957                 nullif(trim(substring(data,22,2)),'') AS delivery_point_suffix,
958                 nullif(trim(substring(data,24,1)),'') AS postcode_type,
959                 nullif(substring(data,25,8)::integer,0) AS address_key,
960                 nullif(substring(data,33,8)::integer,0) AS organisation_key,
961                 nullif(trim(substring(data,41,1)),'') AS amendment_type,
962                 nullif(substring(data,42,1)::integer,0) AS record_type,
963                 nullif(substring(data,43,8)::integer,0) AS locality_key,
964                 nullif(substring(data,51,8)::integer,0) AS thoroughfare_key,
965                 nullif(substring(data,59,8)::integer,0) AS thoroughfare_descriptor_key,
966                 nullif(substring(data,67,8)::integer,0) AS dependent_thoroughfare_key,
967                 nullif(substring(data,75,8)::integer,0) AS dependent_thoroughfare_descriptor_key,
968                 nullif(substring(data,83,4)::integer,0) AS building_number,
969                 nullif(substring(data,87,8)::integer,0) AS building_name_key,
970                 nullif(substring(data,95,8)::integer,0) AS sub_building_name_key,
971                 nullif(substring(data,103,4)::integer,0) AS number_of_households,
972                 nullif(trim(substring(data,107,1)),'') AS concatenation_indicator,
973                 nullif(trim(substring(data,108,6)),'') AS po_box_number,
974                 nullif(trim(substring(data,114,1)),'') AS small_user_organisation_indicator,
975                 CASE nullif(substring(data,115,2)::integer,0) 
976                         WHEN 1 THEN 'New' 
977                         WHEN 2 THEN 'Correction'
978                         WHEN 3 THEN '' 
979                         WHEN 4 THEN 'Coding Revision'
980                         WHEN 5 THEN 'Organisation Change'
981                         WHEN 6 THEN 'Status Change'
982                         WHEN 7 THEN 'Large User Deleted'
983                         WHEN 8 THEN 'Building/Sub Building Change'
984                         WHEN 9 THEN 'Large User Change'
985                 END AS reason_for_amendment,            
986                 nullif(trim(substring(data,117,4)),'') AS new_outward_code,
987                 nullif(trim(substring(data,121,3)),'') AS new_inward_code
988         FROM data_stage 
989         WHERE nullif(substring(data,42,1)::integer,0) = 1;
990         
991         GET DIAGNOSTICS v_processed = ROW_COUNT;
992         RAISE NOTICE '%: Prepared % records for update on welsh mainfile',clock_timestamp(), v_processed;
993         
994         DELETE FROM public.pc_paf_mainfile_welsh mw
995         USING tmp_mainfile_welsh twm WHERE twm.address_key = mw.address_key AND twm.organisation_key IS NOT DISTINCT FROM mw.organisation_key AND twm.postcode_type = mw.postcode_type
996         AND twm.amendment_type IN ('D', 'B');
997
998         GET DIAGNOSTICS v_processed = ROW_COUNT;
999         RAISE NOTICE '%: Removed % records from welsh mainfile',clock_timestamp(), v_processed;
1000         
1001         INSERT INTO public.pc_paf_mainfile_welsh (outward_code, inward_code, address_key, locality_key, thoroughfare_key, 
1002                 thoroughfare_descriptor_key, dependent_thoroughfare_key, dependent_thoroughfare_descriptor_key, 
1003                 building_number, building_name_key, sub_building_name_key, number_of_households, 
1004                 organisation_key, postcode_type, concatenation_indicator, delivery_point_suffix, 
1005                 small_user_organisation_indicator, po_box_number)
1006         WITH twm AS (
1007                 SELECT cume_dist() OVER w, * 
1008                 FROM tmp_mainfile_welsh 
1009                 WINDOW w AS (PARTITION BY address_key, organisation_key, postcode_type ORDER BY timestamp)
1010         )
1011         SELECT COALESCE(twm.new_outward_code, twm.outward_code), COALESCE(twm.new_inward_code, twm.inward_code), twm.address_key, twm.locality_key, twm.thoroughfare_key, 
1012                 twm.thoroughfare_descriptor_key, twm.dependent_thoroughfare_key, twm.dependent_thoroughfare_descriptor_key, 
1013                 twm.building_number, twm.building_name_key, twm.sub_building_name_key, twm.number_of_households, 
1014                 twm.organisation_key, twm.postcode_type, twm.concatenation_indicator, twm.delivery_point_suffix, 
1015                 twm.small_user_organisation_indicator, twm.po_box_number
1016         FROM twm
1017         WHERE twm.cume_dist = 1 AND twm.amendment_type IN ('I', 'C');
1018         
1019         GET DIAGNOSTICS v_processed = ROW_COUNT;
1020         RAISE NOTICE '%: Created % records in welsh mainfile',clock_timestamp(), v_processed;
1021                 
1022         -- Organisations
1023         RAISE NOTICE '%: Preparing to update welsh organisations', clock_timestamp();
1024         
1025         CREATE TEMPORARY TABLE tmp_organisations_welsh ON COMMIT DROP AS
1026         SELECT to_timestamp(nullif(substring(data,1,14),'00000000000000'), 'YYYYMMDDHH24MISS') AS timestamp,
1027                         nullif(trim(substring(data,15,4)),'') AS outward_code,
1028                         nullif(trim(substring(data,19,3)),'') AS inward_code,
1029                         nullif(trim(substring(data,22,2)),'') AS delivery_point_suffix,
1030                         nullif(trim(substring(data,24,1)),'') AS postcode_type,
1031                         nullif(substring(data,25,8)::integer,0) AS address_key,
1032                         nullif(substring(data,33,8)::integer,0) AS organisation_key,
1033                         nullif(trim(substring(data,41,1)),'') AS amendment_type,
1034                         nullif(substring(data,42,1)::integer,0) AS record_type,
1035                         nullif(trim(substring(data,43,60)),'') AS organisation_name,
1036                         nullif(trim(substring(data,103,60)),'') AS department_name,
1037                         nullif(trim(substring(data,163,6)),'') AS po_box_number
1038         FROM data_stage 
1039         WHERE nullif(substring(data,42,1)::integer,0) IN (2,3); 
1040
1041         GET DIAGNOSTICS v_processed = ROW_COUNT;
1042         RAISE NOTICE '%: Prepared % records for update on welsh organisations',clock_timestamp(), v_processed;
1043         
1044         DELETE FROM public.pc_paf_organisations o
1045         USING tmp_organisations_welsh txw WHERE COALESCE(txw.organisation_key, txw.address_key) = o.organisation_key AND txw.postcode_type = o.postcode_type
1046         AND txw.amendment_type IN ('D', 'B');
1047
1048         GET DIAGNOSTICS v_processed = ROW_COUNT;
1049         RAISE NOTICE '%: Removed % records from welsh organisations',clock_timestamp(), v_processed;
1050         
1051         INSERT INTO public.pc_paf_organisations
1052         WITH txw AS (
1053                 SELECT cume_dist() OVER w, * 
1054                 FROM tmp_organisations_welsh            
1055                 WINDOW w AS (PARTITION BY COALESCE(organisation_key, address_key), postcode_type ORDER BY timestamp)
1056         )
1057         SELECT COALESCE(txw.organisation_key, txw.address_key), txw.postcode_type, txw.organisation_name, txw.department_name
1058         FROM  txw
1059         WHERE txw.cume_dist = 1 AND amendment_type IN ('I', 'C');
1060
1061         GET DIAGNOSTICS v_processed = ROW_COUNT;
1062         RAISE NOTICE '%: Created % records in welsh organisations',clock_timestamp(), v_processed;
1063         
1064         -- It would appear postzon is supplied as a full refresh rather than an update file
1065         TRUNCATE TABLE data_stage;      
1066         RAISE NOTICE '%: Begin staging postzon', clock_timestamp();
1067                 
1068         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_postzon || 'pzone100.c01');
1069         EXECUTE v_sql;                  
1070         DELETE FROM data_stage WHERE data like '%PZONE100' || in_edition || '%' OR data like v_main_footer; 
1071         
1072         RAISE NOTICE '%: Done staging postzon file, now staging with lat long data', clock_timestamp();
1073         
1074         CREATE TEMPORARY TABLE tmp_postzon_100m ON COMMIT DROP AS
1075         SELECT nullif(trim(substring(data,1,4)),'') AS outward_code,
1076                 nullif(trim(substring(data,5,3)),'') AS inward_code,
1077                 to_date(substring(data,8,6), 'YYYYMM') AS introduction_date,
1078                 (nullif(trim(substring(data,14,5)),'') || '0')::integer AS grid_reference_east,         
1079                 (CASE WHEN substring(nullif(trim(substring(data,19,5)),''),1,1) ~ '[\x4f-\x5a]' THEN '1' || translate(nullif(trim(substring(data,19,5)),''), 'POUTZY', '221100') ELSE nullif(trim(substring(data,19,5)),'') END || '0')::integer AS grid_reference_north,
1080                 nullif(trim(substring(data,24,9)),'') AS country_code,
1081                 nullif(trim(substring(data,33,9)),'') AS area_code_county,
1082                 nullif(trim(substring(data,42,9)),'') AS area_code_district,
1083                 nullif(trim(substring(data,51,9)),'') AS ward_code,
1084                 nullif(trim(substring(data,60,9)),'') AS nhs_region,
1085                 nullif(trim(substring(data,69,9)),'') AS nhs_code,
1086                 nullif(trim(substring(data,78,1)),'')::smallint AS user_type,
1087                 nullif(trim(substring(data,79,1)),'')::smallint AS grid_status,
1088                 ST_y(ST_transform(ST_GeomFromText('POINT('||(nullif(trim(substring(data,14,5)),'') || '0')||' '||(CASE WHEN substring(nullif(trim(substring(data,19,5)),''),1,1) ~ '[\x4f-\x5a]' THEN '1' || translate(nullif(trim(substring(data,19,5)),''), 'POUTZY', '221100') ELSE nullif(trim(substring(data,19,5)),'') END || '0')||')',CASE WHEN nullif(trim(substring(data,1,4)),'') LIKE 'BT%' THEN 29903 ELSE 27700 END),4326))::numeric(8,6) AS latitude,
1089                 ST_x(ST_transform(ST_GeomFromText('POINT('||(nullif(trim(substring(data,14,5)),'') || '0')||' '||(CASE WHEN substring(nullif(trim(substring(data,19,5)),''),1,1) ~ '[\x4f-\x5a]' THEN '1' || translate(nullif(trim(substring(data,19,5)),''), 'POUTZY', '221100') ELSE nullif(trim(substring(data,19,5)),'') END || '0')||')',CASE WHEN nullif(trim(substring(data,1,4)),'') LIKE 'BT%' THEN 29903 ELSE 27700 END),4326))::numeric(8,6) AS longitude
1090         FROM data_stage;
1091         
1092         RAISE NOTICE '%: Done staging postzon with lat long data, updating', clock_timestamp();
1093         
1094         INSERT INTO public.pc_paf_postzon_100m (outward_code, inward_code, introduction_date, grid_reference_east, grid_reference_north,
1095         country_code, area_code_county, area_code_district, ward_code, nhs_region, nhs_code, user_type, grid_status, latitude, longitude)
1096         SELECT * FROM tmp_postzon_100m tp
1097         WHERE NOT EXISTS (SELECT 1 FROM public.pc_paf_postzon_100m WHERE tp.outward_code = outward_code AND tp.inward_code = inward_code);
1098         
1099         GET DIAGNOSTICS v_processed = ROW_COUNT;
1100         RAISE NOTICE '%: Inserted % new records for update on postzon_100m',clock_timestamp(), v_processed;
1101         
1102         UPDATE public.pc_paf_postzon_100m pz
1103         SET (outward_code, inward_code, introduction_date, grid_reference_east, grid_reference_north,
1104         country_code, area_code_county, area_code_district, ward_code, nhs_region, nhs_code, user_type, grid_status, latitude, longitude)
1105         = (tp.outward_code, tp.inward_code, tp.introduction_date, tp.grid_reference_east, tp.grid_reference_north,
1106         tp.country_code, tp.area_code_county, tp.area_code_district, tp.ward_code, tp.nhs_region, tp.nhs_code, tp.user_type, tp.grid_status, tp.latitude, tp.longitude)
1107         FROM tmp_postzon_100m tp
1108         WHERE tp.outward_code = pz.outward_code AND tp.inward_code = pz.inward_code
1109         AND (pz.introduction_date, pz.grid_reference_east, pz.grid_reference_north,
1110         pz.country_code, pz.area_code_county, pz.area_code_district, pz.ward_code, pz.nhs_region, pz.nhs_code, pz.user_type, pz.grid_status, pz.latitude, pz.longitude)
1111         <> (tp.introduction_date, tp.grid_reference_east, tp.grid_reference_north,
1112         tp.country_code, tp.area_code_county, tp.area_code_district, tp.ward_code, tp.nhs_region, tp.nhs_code, tp.user_type, tp.grid_status, tp.latitude, tp.longitude);
1113
1114         GET DIAGNOSTICS v_processed = ROW_COUNT;
1115         RAISE NOTICE '%: Updated % records for update on postzon_100m',clock_timestamp(), v_processed;  
1116         
1117         DELETE FROM public.pc_paf_postzon_100m pz
1118         WHERE NOT EXISTS (SELECT 1 FROM tmp_postzon_100m WHERE pz.outward_code = outward_code AND pz.inward_code = inward_code)
1119         AND postzon_100m_key <> 0;
1120         
1121         GET DIAGNOSTICS v_processed = ROW_COUNT;
1122         RAISE NOTICE '%: Deleted % records for update on postzon_100m',clock_timestamp(), v_processed;  
1123                 
1124         -- alias is also supplied as a full refresh rather than an update file
1125         TRUNCATE TABLE data_stage;
1126         RAISE NOTICE '%: Begin staging alias file', clock_timestamp();
1127         v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_alias || 'aliasfle.c01');
1128         EXECUTE v_sql;  
1129         DELETE FROM data_stage WHERE data like '%ALIASFLE'  || in_edition ||  '%' OR data like v_std_footer;
1130         
1131         RAISE NOTICE '%: Done staging alias file', clock_timestamp();   
1132                 
1133         CREATE TEMPORARY TABLE tmp_counties ON COMMIT DROP AS
1134         SELECT substring(data,2,4)::integer AS county_key,
1135                 trim(substring(data,6,30)) AS county_name,
1136                 trim(substring(data,36,1)) AS county_type
1137         FROM data_stage
1138         WHERE substring(data,1,1)::integer = 4;
1139         
1140         RAISE NOTICE '%: Done staging counties , updating', clock_timestamp();
1141         
1142         INSERT INTO public.pc_paf_counties 
1143         SELECT * FROM tmp_counties tc
1144         WHERE NOT EXISTS (SELECT 1 FROM public.pc_paf_counties WHERE tc.county_key = county_key);
1145         
1146         GET DIAGNOSTICS v_processed = ROW_COUNT;
1147         RAISE NOTICE '%: Inserted % new records for update on counties',clock_timestamp(), v_processed;
1148         
1149         UPDATE public.pc_paf_counties pc
1150         SET (county_name, county_type) = (tc.county_name, tc.county_type)
1151         FROM tmp_counties tc
1152         WHERE tc.county_key = pc.county_key
1153         AND (pc.county_name, pc.county_type) <> (tc.county_name, tc.county_type);
1154         
1155         GET DIAGNOSTICS v_processed = ROW_COUNT;
1156         RAISE NOTICE '%: Updated % records for update on counties',clock_timestamp(), v_processed;
1157         
1158         DELETE FROM public.pc_paf_counties pc
1159         WHERE NOT EXISTS (SELECT 1 FROM tmp_counties WHERE pc.county_key = county_key)
1160         AND pc.county_key <> 0;
1161         
1162         GET DIAGNOSTICS v_processed = ROW_COUNT;
1163         RAISE NOTICE '%: Deleted % records for update on counties',clock_timestamp(), v_processed;
1164         
1165         CREATE TEMPORARY TABLE tmp_county_alias ON COMMIT DROP AS
1166         SELECT trim(substring(data,2,7)) AS postcode,
1167                 nullif(substring(data,9,4)::integer,0) AS former_postal_county,
1168                 nullif(substring(data,13,4)::integer,0) AS traditional_county,
1169                 nullif(substring(data,17,4)::integer,0) AS administrative_county
1170         FROM data_stage
1171         WHERE substring(data,1,1)::integer = 5;
1172         
1173         RAISE NOTICE '%: Done staging county_alias , updating', clock_timestamp();
1174         
1175         INSERT INTO public.pc_paf_county_alias (postcode, former_postal_county, traditional_county, administrative_county)
1176         SELECT * FROM tmp_county_alias tca
1177         WHERE NOT EXISTS (SELECT 1 FROM public.pc_paf_county_alias WHERE tca.postcode = postcode);
1178         
1179         GET DIAGNOSTICS v_processed = ROW_COUNT;
1180         RAISE NOTICE '%: Inserted % new records for update on county_alias',clock_timestamp(), v_processed;     
1181         
1182         UPDATE public.pc_paf_county_alias pca
1183         SET (former_postal_county, traditional_county, administrative_county) = 
1184         (tca.former_postal_county, tca.traditional_county, tca.administrative_county)
1185         FROM tmp_county_alias tca
1186         WHERE tca.postcode = pca.postcode
1187         AND (pca.former_postal_county, pca.traditional_county, pca.administrative_county) <> (tca.former_postal_county, tca.traditional_county, tca.administrative_county);
1188         
1189         GET DIAGNOSTICS v_processed = ROW_COUNT;
1190         RAISE NOTICE '%: Updated % records for update on county_alias',clock_timestamp(), v_processed;
1191         
1192         DELETE FROM public.pc_paf_county_alias pca
1193         WHERE NOT EXISTS (SELECT 1 FROM tmp_county_alias WHERE pca.postcode = postcode)
1194         AND pca.county_alias_key <> 0;
1195                 
1196         GET DIAGNOSTICS v_processed = ROW_COUNT;
1197         RAISE NOTICE '%: Deleted % records for update on county_alias',clock_timestamp(), v_processed;
1198         
1199         RAISE NOTICE '%: Completed', clock_timestamp();
1200         
1201         RETURN true;    
1202 END;
1203
1204 $BODY$
1205 LANGUAGE 'plpgsql' VOLATILE;
1206
1207 -- Standard master view for postcode tables
1208 CREATE OR REPLACE VIEW public.pc_paf_master_view AS
1209 SELECT m.building_number,
1210         b.building_name,
1211         sb.sub_building_name,
1212         COALESCE(o.organisation_name, lo.organisation_name) AS organisation_name,
1213         COALESCE(o.department_name, lo.department_name) AS department_name,
1214         m.po_box_number,
1215         t.thoroughfare_name,
1216         td.thoroughfare_descriptor,
1217         td.approved_abbreviation,
1218         l.post_town,
1219         l.dependent_locality,
1220         l.double_dependent_locality,
1221         apc.county_name,
1222         m.outward_code,
1223         m.inward_code,
1224         m.number_of_households,
1225         m.postcode_type,
1226         m.concatenation_indicator,
1227         m.delivery_point_suffix,
1228         m.small_user_organisation_indicator,
1229         EXISTS(SELECT 1 FROM public.pc_paf_mainfile_welsh mw WHERE mw.address_key = m.address_key AND mw.organisation_key IS NOT DISTINCT FROM m.organisation_key AND mw.postcode_type IS NOT DISTINCT FROM m.postcode_type) AS welsh_alternative_available,
1230         p.longitude,
1231         p.latitude
1232 FROM public.pc_paf_mainfile m
1233 LEFT OUTER JOIN public.pc_paf_localities l USING (locality_key)
1234 LEFT OUTER JOIN public.pc_paf_thoroughfares t USING (thoroughfare_key)
1235 LEFT OUTER JOIN public.pc_paf_thoroughfare_descriptor td USING (thoroughfare_descriptor_key)
1236 LEFT OUTER JOIN public.pc_paf_building_names b USING (building_name_key)
1237 LEFT OUTER JOIN public.pc_paf_sub_building_names sb USING (sub_building_name_key)
1238 LEFT OUTER JOIN public.pc_paf_organisations o USING (organisation_key, postcode_type)
1239 LEFT OUTER JOIN public.pc_paf_organisations lo ON m.address_key = lo.organisation_key AND m.postcode_type = lo.postcode_type AND m.postcode_type = 'L'
1240 LEFT OUTER JOIN public.pc_paf_postzon_100m p USING (outward_code, inward_code)
1241 LEFT OUTER JOIN public.pc_paf_county_alias a ON a.postcode = (rpad(m.outward_code,4) || m.inward_code)
1242 LEFT OUTER JOIN public.pc_paf_counties apc ON apc.county_key = a.former_postal_county;
1243
1244 -- Welsh master view for postcode tables
1245 CREATE OR REPLACE VIEW public.pc_paf_master_welsh_view AS
1246 SELECT mw.building_number,
1247         b.building_name,
1248         sb.sub_building_name,
1249         COALESCE(o.organisation_name, lo.organisation_name) AS organisation_name,
1250         COALESCE(o.department_name, lo.department_name) AS department_name,
1251         mw.po_box_number,
1252         t.thoroughfare_name,
1253         td.thoroughfare_descriptor,
1254         td.approved_abbreviation,
1255         l.post_town,
1256         l.dependent_locality,
1257         l.double_dependent_locality,
1258         apc.county_name,
1259         mw.outward_code,
1260         mw.inward_code,
1261         mw.number_of_households,
1262         mw.postcode_type,
1263         mw.concatenation_indicator,
1264         mw.delivery_point_suffix,
1265         mw.small_user_organisation_indicator,
1266         p.longitude,
1267         p.latitude
1268 FROM public.pc_paf_mainfile_welsh mw
1269 LEFT OUTER JOIN public.pc_paf_localities l USING (locality_key)
1270 LEFT OUTER JOIN public.pc_paf_thoroughfares t USING (thoroughfare_key)
1271 LEFT OUTER JOIN public.pc_paf_thoroughfare_descriptor td USING (thoroughfare_descriptor_key)
1272 LEFT OUTER JOIN public.pc_paf_building_names b USING (building_name_key)
1273 LEFT OUTER JOIN public.pc_paf_sub_building_names sb USING (sub_building_name_key)
1274 LEFT OUTER JOIN public.pc_paf_organisations o USING (organisation_key, postcode_type)
1275 LEFT OUTER JOIN public.pc_paf_organisations lo ON mw.address_key = lo.organisation_key AND mw.postcode_type = lo.postcode_type AND mw.postcode_type = 'L'
1276 LEFT OUTER JOIN public.pc_paf_postzon_100m p USING (outward_code, inward_code)
1277 LEFT OUTER JOIN public.pc_paf_county_alias a ON a.postcode = (rpad(mw.outward_code,4) || mw.inward_code)
1278 LEFT OUTER JOIN public.pc_paf_counties apc ON apc.county_key = a.former_postal_county;