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)
9 v_data_postzon varchar;
12 v_main_footer varchar;
15 v_table_created boolean;
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/';
23 v_std_footer = '99999999%';
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;
29 TRUNCATE TABLE data_stage;
31 RAISE NOTICE '%: Begin staging localaties', clock_timestamp();
33 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'local.c01');
36 DELETE FROM data_stage WHERE data like '%LOCALITY' || in_edition || '%' OR data like v_std_footer;
38 RAISE NOTICE '%: Done staging localaties, importing', clock_timestamp();
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;
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)
51 v_table_created := true;
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)),'')
62 GET DIAGNOSTICS v_processed = ROW_COUNT;
64 IF (v_table_created) THEN
65 ALTER TABLE public.pc_paf_localities ADD PRIMARY KEY (locality_key);
68 RAISE NOTICE '%: Done importing localities (imported % records)', clock_timestamp(), v_processed;
71 TRUNCATE TABLE data_stage;
73 RAISE NOTICE '%: Begin staging thoroughfares', clock_timestamp();
75 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'thfare.c01');
78 DELETE FROM data_stage WHERE data like '%THOROUGH' || in_edition || '%' OR data like v_std_footer;
80 RAISE NOTICE '%: Done staging thoroughfares, importing', clock_timestamp();
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;
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)
91 v_table_created := true;
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)),'')
100 GET DIAGNOSTICS v_processed = ROW_COUNT;
102 IF (v_table_created) THEN
103 ALTER TABLE public.pc_paf_thoroughfares ADD PRIMARY KEY (thoroughfare_key);
106 RAISE NOTICE '%: Done importing thoroughfares (imported % records)', clock_timestamp(), v_processed;
108 -- 3) Thoroughfares Descriptor
109 TRUNCATE TABLE data_stage;
111 RAISE NOTICE '%: Begin staging thoroughfares descriptor', clock_timestamp();
113 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'thdesc.c01');
116 DELETE FROM data_stage WHERE data like '%THDESCRI' || in_edition || '%' OR data like v_std_footer;
118 RAISE NOTICE '%: Done staging thoroughfares descriptor, importing', clock_timestamp();
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;
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)
130 v_table_created := true;
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)),'')
140 GET DIAGNOSTICS v_processed = ROW_COUNT;
142 IF (v_table_created) THEN
143 ALTER TABLE public.pc_paf_thoroughfare_descriptor ADD PRIMARY KEY (thoroughfare_descriptor_key);
146 RAISE NOTICE '%: Done importing thoroughfares descriptor (imported % records)', clock_timestamp(), v_processed;
149 TRUNCATE TABLE data_stage;
151 RAISE NOTICE '%: Begin staging building names', clock_timestamp();
153 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'bname.c01');
156 DELETE FROM data_stage WHERE data like '%BUILDING' || in_edition || '%' OR data like v_std_footer;
158 RAISE NOTICE '%: Done staging building names, importing', clock_timestamp();
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;
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)
169 v_table_created := true;
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)),'')
178 GET DIAGNOSTICS v_processed = ROW_COUNT;
180 IF (v_table_created) THEN
181 ALTER TABLE public.pc_paf_building_names ADD PRIMARY KEY (building_name_key);
184 RAISE NOTICE '%: Done importing building names (imported % records)', clock_timestamp(), v_processed;
186 -- 5) Sub Building Names file (subbname.c01)
187 TRUNCATE TABLE data_stage;
189 RAISE NOTICE '%: Begin staging sub building names', clock_timestamp();
191 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'subbname.c01');
194 DELETE FROM data_stage WHERE data like '%SUBBUILD' || in_edition || '%' OR data like v_std_footer;
196 RAISE NOTICE '%: Done staging sub building names, importing', clock_timestamp();
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;
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)
207 v_table_created := true;
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)),'')
216 GET DIAGNOSTICS v_processed = ROW_COUNT;
218 IF (v_table_created) THEN
219 ALTER TABLE public.pc_paf_sub_building_names ADD PRIMARY KEY (sub_building_name_key);
222 RAISE NOTICE '%: Done importing sub building names (imported % records)', clock_timestamp(), v_processed;
225 TRUNCATE TABLE data_stage;
227 RAISE NOTICE '%: Begin staging organisations', clock_timestamp();
229 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'org.c01');
232 DELETE FROM data_stage WHERE data like '%ORGANISA' || in_edition || '%' OR data like v_std_footer;
234 RAISE NOTICE '%: Done staging organisations, importing', clock_timestamp();
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;
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)
247 v_table_created := true;
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)),'')
258 GET DIAGNOSTICS v_processed = ROW_COUNT;
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';
265 RAISE NOTICE '%: Done importing organisations (imported % records)', clock_timestamp(), v_processed;
267 -- 7) Postzon with latlon generated using postgis
268 TRUNCATE TABLE data_stage;
270 RAISE NOTICE '%: Begin staging postzon', clock_timestamp();
272 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_postzon || 'pzone100.c01');
275 DELETE FROM data_stage WHERE data like '%PZONE100' || in_edition || '%' OR data like v_main_footer;
277 RAISE NOTICE '%: Done staging postzon, importing', clock_timestamp();
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;
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),
298 grid_status smallint,
299 latitude double precision,
300 longitude double precision
302 v_table_created := true;
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) ~ E'[\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) ~ E'[\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) ~ E'[\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
325 GET DIAGNOSTICS v_processed = ROW_COUNT;
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);
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 E'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';
347 RAISE NOTICE '%: Done importing postzon (imported % records)', clock_timestamp(), v_processed;
350 TRUNCATE TABLE data_stage;
351 RAISE NOTICE '%: Begin staging mainfile', clock_timestamp();
353 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c02');
355 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c03');
357 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c04');
359 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c05');
361 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_main || 'fpmainfl.c06');
364 DELETE FROM data_stage WHERE data like '%ADDRESS ' || in_edition || '%' OR data like v_main_footer;
366 RAISE NOTICE '%: Done staging mainfile, importing', clock_timestamp();
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;
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)
394 v_table_created := true;
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)),'')
422 GET DIAGNOSTICS v_processed = ROW_COUNT;
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);
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!';
447 RAISE NOTICE '%: Done importing mainfile (imported % records)', clock_timestamp(), v_processed;
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');
455 DELETE FROM data_stage WHERE data like '%ADDRESS ' || in_edition || '%' OR data like v_main_footer;
457 RAISE NOTICE '%: Done staging welsh alternative mainfile, importing', clock_timestamp();
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;
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)
485 v_table_created := true;
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)),'')
513 GET DIAGNOSTICS v_processed = ROW_COUNT;
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);
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!';
537 RAISE NOTICE '%: Done importing welsh alternative mainfile (imported % records)', clock_timestamp(), v_processed;
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');
545 DELETE FROM data_stage WHERE data like '%ALIASFLE' || in_edition || '%' OR data like v_std_footer;
547 RAISE NOTICE '%: Done staging alias file, importing', clock_timestamp();
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;
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)
559 v_table_created := true;
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
568 WHERE substring(data,1,1)::integer = 4;
570 GET DIAGNOSTICS v_processed = ROW_COUNT;
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)';
578 RAISE NOTICE '%: Done importing counties (imported % records)', clock_timestamp(), v_processed;
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;
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
592 v_table_created := true;
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
602 WHERE substring(data,1,1)::integer = 5;
604 GET DIAGNOSTICS v_processed = ROW_COUNT;
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;
614 RAISE NOTICE '%: Done importing county_alias (imported % records)', clock_timestamp(), v_processed;
616 RAISE NOTICE '%: Completed', clock_timestamp();
621 LANGUAGE 'plpgsql' VOLATILE;
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)
632 v_data_update varchar;
633 v_data_postzon varchar;
634 v_data_alias varchar;
636 v_main_footer varchar;
637 v_std_footer varchar;
639 v_table_created boolean;
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/';
646 v_main_footer = ' %';
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;
651 --changes1.c01 - Changes1 (Changes to satelite tables except Organisations) - A single changes file
652 RAISE NOTICE '%: Begin staging Changes1 file', clock_timestamp();
654 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_update || 'changes1.c01');
656 DELETE FROM data_stage WHERE data like '%CHANGES1' || in_edition || '%' OR data like v_main_footer;
658 RAISE NOTICE '%: Done staging Changes1 file',clock_timestamp();
660 SET CONSTRAINTS ALL DEFERRED;
662 --Record Type 1 - Localities
663 RAISE NOTICE '%: Preparing to update localities', clock_timestamp();
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
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');
677 GET DIAGNOSTICS v_processed = ROW_COUNT;
678 RAISE NOTICE '%: Prepared % records for update on localities',clock_timestamp(), v_processed;
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));
685 GET DIAGNOSTICS v_processed = ROW_COUNT;
686 RAISE NOTICE '%: Removed % records from localities',clock_timestamp(), v_processed;
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');
693 GET DIAGNOSTICS v_processed = ROW_COUNT;
694 RAISE NOTICE '%: Created % records in localities',clock_timestamp(), v_processed;
696 --Record Type 2 - Thoroughfares
697 RAISE NOTICE '%: Preparing to update thoroughfares', clock_timestamp();
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
705 WHERE substring(data,1,1)::integer = 2 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
707 GET DIAGNOSTICS v_processed = ROW_COUNT;
708 RAISE NOTICE '%: Prepared % records for update on thoroughfares',clock_timestamp(), v_processed;
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));
715 GET DIAGNOSTICS v_processed = ROW_COUNT;
716 RAISE NOTICE '%: Removed % records from thoroughfares',clock_timestamp(), v_processed;
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');
723 GET DIAGNOSTICS v_processed = ROW_COUNT;
724 RAISE NOTICE '%: Created % records in thoroughfares',clock_timestamp(), v_processed;
726 --Record Type 3 - Thoroughfare Descriptors
727 RAISE NOTICE '%: Preparing to update thoroughfare_descriptor', clock_timestamp();
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
736 WHERE substring(data,1,1)::integer = 3 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
738 GET DIAGNOSTICS v_processed = ROW_COUNT;
739 RAISE NOTICE '%: Prepared % records for update on thoroughfare_descriptor',clock_timestamp(), v_processed;
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));
746 GET DIAGNOSTICS v_processed = ROW_COUNT;
747 RAISE NOTICE '%: Removed % records from thoroughfare_descriptor',clock_timestamp(), v_processed;
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');
754 GET DIAGNOSTICS v_processed = ROW_COUNT;
755 RAISE NOTICE '%: Created % records in thoroughfare_descriptor',clock_timestamp(), v_processed;
757 --Record Type 4 - Building Names
758 RAISE NOTICE '%: Preparing to update building_names', clock_timestamp();
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
766 WHERE substring(data,1,1)::integer = 4 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
768 GET DIAGNOSTICS v_processed = ROW_COUNT;
769 RAISE NOTICE '%: Prepared % records for update on building_names',clock_timestamp(), v_processed;
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));
776 GET DIAGNOSTICS v_processed = ROW_COUNT;
777 RAISE NOTICE '%: Removed % records from building_names',clock_timestamp(), v_processed;
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');
784 GET DIAGNOSTICS v_processed = ROW_COUNT;
785 RAISE NOTICE '%: Created % records in building_names',clock_timestamp(), v_processed;
787 --Record Type 5 - Sub Building Names
788 RAISE NOTICE '%: Preparing to update sub_building_names', clock_timestamp();
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
796 WHERE substring(data,1,1)::integer = 5 AND nullif(trim(substring(data,24,1)),'') IS NOT NULL;
798 GET DIAGNOSTICS v_processed = ROW_COUNT;
799 RAISE NOTICE '%: Prepared % records for update on sub_building_names',clock_timestamp(), v_processed;
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));
806 GET DIAGNOSTICS v_processed = ROW_COUNT;
807 RAISE NOTICE '%: Removed % records from sub_building_names',clock_timestamp(), v_processed;
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');
814 GET DIAGNOSTICS v_processed = ROW_COUNT;
815 RAISE NOTICE '%: Created % records in sub_building_names',clock_timestamp(), v_processed;
817 TRUNCATE TABLE data_stage;
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();
822 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_update || 'fpchgsng.c01');
824 DELETE FROM data_stage WHERE data like '%CHANGES2' || in_edition || '%' OR data like v_main_footer;
826 RAISE NOTICE '%: Done staging Changes2 file',clock_timestamp();
829 RAISE NOTICE '%: Preparing to update mainfile', clock_timestamp();
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)
855 WHEN 2 THEN 'Correction'
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
867 WHERE nullif(substring(data,42,1)::integer,0) = 1;
869 GET DIAGNOSTICS v_processed = ROW_COUNT;
870 RAISE NOTICE '%: Prepared % records for update on mainfile',clock_timestamp(), v_processed;
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');
876 GET DIAGNOSTICS v_processed = ROW_COUNT;
877 RAISE NOTICE '%: Removed % records from mainfile',clock_timestamp(), v_processed;
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
890 SELECT cume_dist() OVER w, *
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');
896 GET DIAGNOSTICS v_processed = ROW_COUNT;
897 RAISE NOTICE '%: Created % records in mainfile',clock_timestamp(), v_processed;
900 RAISE NOTICE '%: Preparing to update organisations', clock_timestamp();
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
916 WHERE nullif(substring(data,42,1)::integer,0) IN (2,3);
918 GET DIAGNOSTICS v_processed = ROW_COUNT;
919 RAISE NOTICE '%: Prepared % records for update on organisations',clock_timestamp(), v_processed;
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');
925 GET DIAGNOSTICS v_processed = ROW_COUNT;
926 RAISE NOTICE '%: Removed % records from organisations',clock_timestamp(), v_processed;
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
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');
937 GET DIAGNOSTICS v_processed = ROW_COUNT;
938 RAISE NOTICE '%: Created % records in organisations',clock_timestamp(), v_processed;
940 -- wchanges.c01 -- Welsh changes (In same format as regular changes2 file (hence has timelined data, not single changes)
941 TRUNCATE TABLE data_stage;
943 RAISE NOTICE '%: Begin staging WChanges file', clock_timestamp();
945 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_update || 'wchanges.c01');
947 DELETE FROM data_stage WHERE data like '%WCHANGES' || in_edition || '%' OR data like v_main_footer;
949 RAISE NOTICE '%: Done staging WChanges file',clock_timestamp();
951 RAISE NOTICE '%: Preparing to update welsh mainfile', clock_timestamp();
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)
977 WHEN 2 THEN 'Correction'
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
989 WHERE nullif(substring(data,42,1)::integer,0) = 1;
991 GET DIAGNOSTICS v_processed = ROW_COUNT;
992 RAISE NOTICE '%: Prepared % records for update on welsh mainfile',clock_timestamp(), v_processed;
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');
998 GET DIAGNOSTICS v_processed = ROW_COUNT;
999 RAISE NOTICE '%: Removed % records from welsh mainfile',clock_timestamp(), v_processed;
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)
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)
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
1017 WHERE twm.cume_dist = 1 AND twm.amendment_type IN ('I', 'C');
1019 GET DIAGNOSTICS v_processed = ROW_COUNT;
1020 RAISE NOTICE '%: Created % records in welsh mainfile',clock_timestamp(), v_processed;
1023 RAISE NOTICE '%: Preparing to update welsh organisations', clock_timestamp();
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
1039 WHERE nullif(substring(data,42,1)::integer,0) IN (2,3);
1041 GET DIAGNOSTICS v_processed = ROW_COUNT;
1042 RAISE NOTICE '%: Prepared % records for update on welsh organisations',clock_timestamp(), v_processed;
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');
1048 GET DIAGNOSTICS v_processed = ROW_COUNT;
1049 RAISE NOTICE '%: Removed % records from welsh organisations',clock_timestamp(), v_processed;
1051 INSERT INTO public.pc_paf_organisations
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)
1057 SELECT COALESCE(txw.organisation_key, txw.address_key), txw.postcode_type, txw.organisation_name, txw.department_name
1059 WHERE txw.cume_dist = 1 AND amendment_type IN ('I', 'C');
1061 GET DIAGNOSTICS v_processed = ROW_COUNT;
1062 RAISE NOTICE '%: Created % records in welsh organisations',clock_timestamp(), v_processed;
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();
1068 v_sql := 'COPY data_stage FROM ' || quote_literal(v_data_postzon || 'pzone100.c01');
1070 DELETE FROM data_stage WHERE data like '%PZONE100' || in_edition || '%' OR data like v_main_footer;
1072 RAISE NOTICE '%: Done staging postzon file, now staging with lat long data', clock_timestamp();
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) ~ E'[\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) ~ E'[\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) ~ E'[\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
1092 RAISE NOTICE '%: Done staging postzon with lat long data, updating', clock_timestamp();
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);
1099 GET DIAGNOSTICS v_processed = ROW_COUNT;
1100 RAISE NOTICE '%: Inserted % new records for update on postzon_100m',clock_timestamp(), v_processed;
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);
1114 GET DIAGNOSTICS v_processed = ROW_COUNT;
1115 RAISE NOTICE '%: Updated % records for update on postzon_100m',clock_timestamp(), v_processed;
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;
1121 GET DIAGNOSTICS v_processed = ROW_COUNT;
1122 RAISE NOTICE '%: Deleted % records for update on postzon_100m',clock_timestamp(), v_processed;
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');
1129 DELETE FROM data_stage WHERE data like '%ALIASFLE' || in_edition || '%' OR data like v_std_footer;
1131 RAISE NOTICE '%: Done staging alias file', clock_timestamp();
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
1138 WHERE substring(data,1,1)::integer = 4;
1140 RAISE NOTICE '%: Done staging counties , updating', clock_timestamp();
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);
1146 GET DIAGNOSTICS v_processed = ROW_COUNT;
1147 RAISE NOTICE '%: Inserted % new records for update on counties',clock_timestamp(), v_processed;
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);
1155 GET DIAGNOSTICS v_processed = ROW_COUNT;
1156 RAISE NOTICE '%: Updated % records for update on counties',clock_timestamp(), v_processed;
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;
1162 GET DIAGNOSTICS v_processed = ROW_COUNT;
1163 RAISE NOTICE '%: Deleted % records for update on counties',clock_timestamp(), v_processed;
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
1171 WHERE substring(data,1,1)::integer = 5;
1173 RAISE NOTICE '%: Done staging county_alias , updating', clock_timestamp();
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);
1179 GET DIAGNOSTICS v_processed = ROW_COUNT;
1180 RAISE NOTICE '%: Inserted % new records for update on county_alias',clock_timestamp(), v_processed;
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);
1189 GET DIAGNOSTICS v_processed = ROW_COUNT;
1190 RAISE NOTICE '%: Updated % records for update on county_alias',clock_timestamp(), v_processed;
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;
1196 GET DIAGNOSTICS v_processed = ROW_COUNT;
1197 RAISE NOTICE '%: Deleted % records for update on county_alias',clock_timestamp(), v_processed;
1199 RAISE NOTICE '%: Completed', clock_timestamp();
1205 LANGUAGE 'plpgsql' VOLATILE;
1207 -- Standard master view for postcode tables
1208 CREATE OR REPLACE VIEW public.pc_paf_master_view AS
1209 SELECT m.building_number,
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,
1215 t.thoroughfare_name,
1216 td.thoroughfare_descriptor,
1217 td.approved_abbreviation,
1219 l.dependent_locality,
1220 l.double_dependent_locality,
1224 m.number_of_households,
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,
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;
1244 -- Welsh master view for postcode tables
1245 CREATE OR REPLACE VIEW public.pc_paf_master_welsh_view AS
1246 SELECT mw.building_number,
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,
1252 t.thoroughfare_name,
1253 td.thoroughfare_descriptor,
1254 td.approved_abbreviation,
1256 l.dependent_locality,
1257 l.double_dependent_locality,
1261 mw.number_of_households,
1263 mw.concatenation_indicator,
1264 mw.delivery_point_suffix,
1265 mw.small_user_organisation_indicator,
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;