git.8kb.co.uk Git - postgresql/geographic_data/blob - 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
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
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;