]> git.openstreetmap.org Git - nominatim.git/blob - lib-sql/functions/placex_triggers.sql
introduce external processing in indexer
[nominatim.git] / lib-sql / functions / placex_triggers.sql
1 -- Trigger functions for the placex table.
2
3 -- Retrieve the data needed by the indexer for updating the place.
4 --
5 -- Return parameters:
6 --  name            list of names
7 --  address         list of address tags, either from the object or a surrounding
8 --                  building
9 --  country_feature If the place is a country feature, this contains the
10 --                  country code, otherwise it is null.
11 CREATE OR REPLACE FUNCTION placex_prepare_update(p placex,
12                                                  OUT name HSTORE,
13                                                  OUT address HSTORE,
14                                                  OUT country_feature VARCHAR)
15   AS $$
16 BEGIN
17   -- For POI nodes, check if the address should be derived from a surrounding
18   -- building.
19   IF p.rank_search < 30 OR p.osm_type != 'N' OR p.address is not null THEN
20     RAISE WARNING 'self address for % %', p.osm_type, p.osm_id;
21     address := p.address;
22   ELSE
23     -- The additional && condition works around the misguided query
24     -- planner of postgis 3.0.
25     SELECT placex.address || hstore('_inherited', '') INTO address
26       FROM placex
27      WHERE ST_Covers(geometry, p.centroid)
28            and geometry && p.centroid
29            and (placex.address ? 'housenumber' or placex.address ? 'street' or placex.address ? 'place')
30            and rank_search > 28 AND ST_GeometryType(geometry) in ('ST_Polygon','ST_MultiPolygon')
31      LIMIT 1;
32     RAISE WARNING 'other address for % %: % (%)', p.osm_type, p.osm_id, address, p.centroid;
33   END IF;
34
35   address := address - '_unlisted_place'::TEXT;
36   name := p.name;
37
38   country_feature := CASE WHEN p.admin_level = 2
39                                and p.class = 'boundary' and p.type = 'administrative'
40                                and p.osm_type = 'R'
41                           THEN p.country_code
42                           ELSE null
43                      END;
44 END;
45 $$
46 LANGUAGE plpgsql STABLE;
47
48
49 -- Find the parent road of a POI.
50 --
51 -- \returns Place ID of parent object or NULL if none
52 --
53 -- Copy data from linked items (POIs on ways, addr:street links, relations).
54 --
55 CREATE OR REPLACE FUNCTION find_parent_for_poi(poi_osm_type CHAR(1),
56                                                poi_osm_id BIGINT,
57                                                poi_partition SMALLINT,
58                                                bbox GEOMETRY,
59                                                addr_street TEXT,
60                                                addr_place TEXT,
61                                                fallback BOOL = true)
62   RETURNS BIGINT
63   AS $$
64 DECLARE
65   parent_place_id BIGINT DEFAULT NULL;
66   location RECORD;
67   parent RECORD;
68 BEGIN
69     {% if debug %}RAISE WARNING 'finding street for % %', poi_osm_type, poi_osm_id;{% endif %}
70
71     -- Is this object part of an associatedStreet relation?
72     FOR location IN
73       SELECT members FROM planet_osm_rels
74       WHERE parts @> ARRAY[poi_osm_id]
75         and members @> ARRAY[lower(poi_osm_type) || poi_osm_id]
76         and tags @> ARRAY['associatedStreet']
77     LOOP
78       FOR i IN 1..array_upper(location.members, 1) BY 2 LOOP
79         IF location.members[i+1] = 'street' THEN
80           FOR parent IN
81             SELECT place_id from placex
82              WHERE osm_type = 'W' and osm_id = substring(location.members[i],2)::bigint
83                and name is not null
84                and rank_search between 26 and 27
85           LOOP
86             RETURN parent.place_id;
87           END LOOP;
88         END IF;
89       END LOOP;
90     END LOOP;
91
92     parent_place_id := find_parent_for_address(addr_street, addr_place,
93                                                poi_partition, bbox);
94     IF parent_place_id is not null THEN
95       RETURN parent_place_id;
96     END IF;
97
98     IF poi_osm_type = 'N' THEN
99       -- Is this node part of an interpolation?
100       FOR parent IN
101         SELECT q.parent_place_id
102           FROM location_property_osmline q, planet_osm_ways x
103          WHERE q.linegeo && bbox and x.id = q.osm_id
104                and poi_osm_id = any(x.nodes)
105          LIMIT 1
106       LOOP
107         {% if debug %}RAISE WARNING 'Get parent from interpolation: %', parent.parent_place_id;{% endif %}
108         RETURN parent.parent_place_id;
109       END LOOP;
110
111       -- Is this node part of any other way?
112       FOR location IN
113         SELECT p.place_id, p.osm_id, p.rank_search, p.address,
114                coalesce(p.centroid, ST_Centroid(p.geometry)) as centroid
115           FROM placex p, planet_osm_ways w
116          WHERE p.osm_type = 'W' and p.rank_search >= 26
117                and p.geometry && bbox
118                and w.id = p.osm_id and poi_osm_id = any(w.nodes)
119       LOOP
120         {% if debug %}RAISE WARNING 'Node is part of way % ', location.osm_id;{% endif %}
121
122         -- Way IS a road then we are on it - that must be our road
123         IF location.rank_search < 28 THEN
124           {% if debug %}RAISE WARNING 'node in way that is a street %',location;{% endif %}
125           return location.place_id;
126         END IF;
127
128         SELECT find_parent_for_poi('W', location.osm_id, poi_partition,
129                                    location.centroid,
130                                    location.address->'street',
131                                    location.address->'place',
132                                    false)
133           INTO parent_place_id;
134         IF parent_place_id is not null THEN
135           RETURN parent_place_id;
136         END IF;
137       END LOOP;
138     END IF;
139
140     IF fallback THEN
141       IF addr_street is null and addr_place is not null THEN
142         -- The address is attached to a place we don't know.
143         -- Instead simply use the containing area with the largest rank.
144         FOR location IN
145           SELECT place_id FROM placex
146             WHERE bbox && geometry AND _ST_Covers(geometry, ST_Centroid(bbox))
147                   AND rank_address between 5 and 25
148             ORDER BY rank_address desc
149         LOOP
150             RETURN location.place_id;
151         END LOOP;
152       ELSEIF ST_Area(bbox) < 0.005 THEN
153         -- for smaller features get the nearest road
154         SELECT getNearestRoadPlaceId(poi_partition, bbox) INTO parent_place_id;
155         {% if debug %}RAISE WARNING 'Checked for nearest way (%)', parent_place_id;{% endif %}
156       ELSE
157         -- for larger features simply find the area with the largest rank that
158         -- contains the bbox, only use addressable features
159         FOR location IN
160           SELECT place_id FROM placex
161             WHERE bbox && geometry AND _ST_Covers(geometry, ST_Centroid(bbox))
162                   AND rank_address between 5 and 25
163             ORDER BY rank_address desc
164         LOOP
165             RETURN location.place_id;
166         END LOOP;
167       END IF;
168     END IF;
169
170     RETURN parent_place_id;
171 END;
172 $$
173 LANGUAGE plpgsql STABLE;
174
175 -- Try to find a linked place for the given object.
176 CREATE OR REPLACE FUNCTION find_linked_place(bnd placex)
177   RETURNS placex
178   AS $$
179 DECLARE
180   relation_members TEXT[];
181   rel_member RECORD;
182   linked_placex placex%ROWTYPE;
183   bnd_name TEXT;
184 BEGIN
185   IF bnd.rank_search >= 26 or bnd.rank_address = 0
186      or ST_GeometryType(bnd.geometry) NOT IN ('ST_Polygon','ST_MultiPolygon')
187      or bnd.type IN ('postcode', 'postal_code')
188   THEN
189     RETURN NULL;
190   END IF;
191
192   IF bnd.osm_type = 'R' THEN
193     -- see if we have any special relation members
194     SELECT members FROM planet_osm_rels WHERE id = bnd.osm_id INTO relation_members;
195     {% if debug %}RAISE WARNING 'Got relation members';{% endif %}
196
197     -- Search for relation members with role 'lable'.
198     IF relation_members IS NOT NULL THEN
199       FOR rel_member IN
200         SELECT get_rel_node_members(relation_members, ARRAY['label']) as member
201       LOOP
202         {% if debug %}RAISE WARNING 'Found label member %', rel_member.member;{% endif %}
203
204         FOR linked_placex IN
205           SELECT * from placex
206           WHERE osm_type = 'N' and osm_id = rel_member.member
207             and class = 'place'
208         LOOP
209           {% if debug %}RAISE WARNING 'Linked label member';{% endif %}
210           RETURN linked_placex;
211         END LOOP;
212
213       END LOOP;
214     END IF;
215   END IF;
216
217   IF bnd.name ? 'name' THEN
218     bnd_name := lower(bnd.name->'name');
219     IF bnd_name = '' THEN
220       bnd_name := NULL;
221     END IF;
222   END IF;
223
224   -- If extratags has a place tag, look for linked nodes by their place type.
225   -- Area and node still have to have the same name.
226   IF bnd.extratags ? 'place' and bnd_name is not null THEN
227     FOR linked_placex IN
228       SELECT * FROM placex
229       WHERE (position(lower(name->'name') in bnd_name) > 0
230              OR position(bnd_name in lower(name->'name')) > 0)
231         AND placex.class = 'place' AND placex.type = bnd.extratags->'place'
232         AND placex.osm_type = 'N'
233         AND placex.linked_place_id is null
234         AND placex.rank_search < 26 -- needed to select the right index
235         AND placex.type != 'postcode'
236         AND ST_Covers(bnd.geometry, placex.geometry)
237     LOOP
238       {% if debug %}RAISE WARNING 'Found type-matching place node %', linked_placex.osm_id;{% endif %}
239       RETURN linked_placex;
240     END LOOP;
241   END IF;
242
243   IF bnd.extratags ? 'wikidata' THEN
244     FOR linked_placex IN
245       SELECT * FROM placex
246       WHERE placex.class = 'place' AND placex.osm_type = 'N'
247         AND placex.extratags ? 'wikidata' -- needed to select right index
248         AND placex.extratags->'wikidata' = bnd.extratags->'wikidata'
249         AND placex.linked_place_id is null
250         AND placex.rank_search < 26
251         AND _st_covers(bnd.geometry, placex.geometry)
252       ORDER BY lower(name->'name') = bnd_name desc
253     LOOP
254       {% if debug %}RAISE WARNING 'Found wikidata-matching place node %', linked_placex.osm_id;{% endif %}
255       RETURN linked_placex;
256     END LOOP;
257   END IF;
258
259   -- Name searches can be done for ways as well as relations
260   IF bnd_name is not null THEN
261     {% if debug %}RAISE WARNING 'Looking for nodes with matching names';{% endif %}
262     FOR linked_placex IN
263       SELECT placex.* from placex
264       WHERE lower(name->'name') = bnd_name
265         AND ((bnd.rank_address > 0
266               and bnd.rank_address = (compute_place_rank(placex.country_code,
267                                                          'N', placex.class,
268                                                          placex.type, 15::SMALLINT,
269                                                          false, placex.postcode)).address_rank)
270              OR (bnd.rank_address = 0 and placex.rank_search = bnd.rank_search))
271         AND placex.osm_type = 'N'
272         AND placex.class = 'place'
273         AND placex.linked_place_id is null
274         AND placex.rank_search < 26 -- needed to select the right index
275         AND placex.type != 'postcode'
276         AND ST_Covers(bnd.geometry, placex.geometry)
277     LOOP
278       {% if debug %}RAISE WARNING 'Found matching place node %', linked_placex.osm_id;{% endif %}
279       RETURN linked_placex;
280     END LOOP;
281   END IF;
282
283   RETURN NULL;
284 END;
285 $$
286 LANGUAGE plpgsql STABLE;
287
288
289 -- Insert address of a place into the place_addressline table.
290 --
291 -- \param obj_place_id  Place_id of the place to compute the address for.
292 -- \param partition     Partition number where the place is in.
293 -- \param maxrank       Rank of the place. All address features must have
294 --                      a search rank lower than the given rank.
295 -- \param address       Address terms for the place.
296 -- \param geometry      Geometry to which the address objects should be close.
297 --
298 -- \retval parent_place_id  Place_id of the address object that is the direct
299 --                          ancestor.
300 -- \retval postcode         Postcode computed from the address. This is the
301 --                          addr:postcode of one of the address objects. If
302 --                          more than one of has a postcode, the highest ranking
303 --                          one is used. May be NULL.
304 -- \retval nameaddress_vector  Search terms for the address. This is the sum
305 --                             of name terms of all address objects.
306 CREATE OR REPLACE FUNCTION insert_addresslines(obj_place_id BIGINT,
307                                                partition SMALLINT,
308                                                maxrank SMALLINT,
309                                                address HSTORE,
310                                                geometry GEOMETRY,
311                                                country TEXT,
312                                                OUT parent_place_id BIGINT,
313                                                OUT postcode TEXT,
314                                                OUT nameaddress_vector INT[])
315   AS $$
316 DECLARE
317   address_havelevel BOOLEAN[];
318
319   location_isaddress BOOLEAN;
320   current_boundary GEOMETRY := NULL;
321   current_node_area GEOMETRY := NULL;
322
323   parent_place_rank INT := 0;
324   addr_place_ids BIGINT[];
325
326   location RECORD;
327 BEGIN
328   parent_place_id := 0;
329   nameaddress_vector := '{}'::int[];
330
331   address_havelevel := array_fill(false, ARRAY[maxrank]);
332
333   FOR location IN
334     SELECT * FROM get_places_for_addr_tags(partition, geometry,
335                                                    address, country)
336     ORDER BY rank_address, distance, isguess desc
337   LOOP
338     {% if not db.reverse_only %}
339       nameaddress_vector := array_merge(nameaddress_vector,
340                                         location.keywords::int[]);
341     {% endif %}
342
343     IF location.place_id is not null THEN
344       location_isaddress := not address_havelevel[location.rank_address];
345       IF not address_havelevel[location.rank_address] THEN
346         address_havelevel[location.rank_address] := true;
347         IF parent_place_rank < location.rank_address THEN
348           parent_place_id := location.place_id;
349           parent_place_rank := location.rank_address;
350         END IF;
351       END IF;
352
353       INSERT INTO place_addressline (place_id, address_place_id, fromarea,
354                                      isaddress, distance, cached_rank_address)
355         VALUES (obj_place_id, location.place_id, not location.isguess,
356                 true, location.distance, location.rank_address);
357
358       addr_place_ids := array_append(addr_place_ids, location.place_id);
359     END IF;
360   END LOOP;
361
362   FOR location IN
363     SELECT * FROM getNearFeatures(partition, geometry, maxrank)
364     WHERE addr_place_ids is null or not addr_place_ids @> ARRAY[place_id]
365     ORDER BY rank_address, isguess asc,
366              distance *
367                CASE WHEN rank_address = 16 AND rank_search = 15 THEN 0.2
368                     WHEN rank_address = 16 AND rank_search = 16 THEN 0.25
369                     WHEN rank_address = 16 AND rank_search = 18 THEN 0.5
370                     ELSE 1 END ASC
371   LOOP
372     -- Ignore all place nodes that do not fit in a lower level boundary.
373     CONTINUE WHEN location.isguess
374                   and current_boundary is not NULL
375                   and not ST_Contains(current_boundary, location.centroid);
376
377     -- If this is the first item in the rank, then assume it is the address.
378     location_isaddress := not address_havelevel[location.rank_address];
379
380     -- Further sanity checks to ensure that the address forms a sane hierarchy.
381     IF location_isaddress THEN
382       IF location.isguess and current_node_area is not NULL THEN
383         location_isaddress := ST_Contains(current_node_area, location.centroid);
384       END IF;
385       IF not location.isguess and current_boundary is not NULL
386          and location.rank_address != 11 AND location.rank_address != 5 THEN
387         location_isaddress := ST_Contains(current_boundary, location.centroid);
388       END IF;
389     END IF;
390
391     IF location_isaddress THEN
392       address_havelevel[location.rank_address] := true;
393       parent_place_id := location.place_id;
394
395       -- Set postcode if we have one.
396       -- (Returned will be the highest ranking one.)
397       IF location.postcode is not NULL THEN
398         postcode = location.postcode;
399       END IF;
400
401       -- Recompute the areas we need for hierarchy sanity checks.
402       IF location.rank_address != 11 AND location.rank_address != 5 THEN
403         IF location.isguess THEN
404           current_node_area := place_node_fuzzy_area(location.centroid,
405                                                      location.rank_search);
406         ELSE
407           current_node_area := NULL;
408           SELECT p.geometry FROM placex p
409               WHERE p.place_id = location.place_id INTO current_boundary;
410         END IF;
411       END IF;
412     END IF;
413
414     -- Add it to the list of search terms
415     {% if not db.reverse_only %}
416       nameaddress_vector := array_merge(nameaddress_vector,
417                                         location.keywords::integer[]);
418     {% endif %}
419
420     INSERT INTO place_addressline (place_id, address_place_id, fromarea,
421                                      isaddress, distance, cached_rank_address)
422         VALUES (obj_place_id, location.place_id, not location.isguess,
423                 location_isaddress, location.distance, location.rank_address);
424   END LOOP;
425 END;
426 $$
427 LANGUAGE plpgsql;
428
429
430 CREATE OR REPLACE FUNCTION placex_insert()
431   RETURNS TRIGGER
432   AS $$
433 DECLARE
434   postcode TEXT;
435   result BOOLEAN;
436   is_area BOOLEAN;
437   country_code VARCHAR(2);
438   diameter FLOAT;
439   classtable TEXT;
440 BEGIN
441   {% if debug %}RAISE WARNING '% % % %',NEW.osm_type,NEW.osm_id,NEW.class,NEW.type;{% endif %}
442
443   NEW.place_id := nextval('seq_place');
444   NEW.indexed_status := 1; --STATUS_NEW
445
446   NEW.centroid := ST_PointOnSurface(NEW.geometry);
447   NEW.country_code := lower(get_country_code(NEW.centroid));
448
449   NEW.partition := get_partition(NEW.country_code);
450   NEW.geometry_sector := geometry_sector(NEW.partition, NEW.centroid);
451
452   IF NEW.osm_type = 'X' THEN
453     -- E'X'ternal records should already be in the right format so do nothing
454   ELSE
455     is_area := ST_GeometryType(NEW.geometry) IN ('ST_Polygon','ST_MultiPolygon');
456
457     IF NEW.class in ('place','boundary')
458        AND NEW.type in ('postcode','postal_code')
459     THEN
460       IF NEW.address IS NULL OR NOT NEW.address ? 'postcode' THEN
461           -- most likely just a part of a multipolygon postcode boundary, throw it away
462           RETURN NULL;
463       END IF;
464
465       NEW.name := hstore('ref', NEW.address->'postcode');
466
467     ELSEIF NEW.class = 'highway' AND is_area AND NEW.name is null
468            AND NEW.extratags ? 'area' AND NEW.extratags->'area' = 'yes'
469     THEN
470         RETURN NULL;
471     ELSEIF NEW.class = 'boundary' AND NOT is_area
472     THEN
473         RETURN NULL;
474     ELSEIF NEW.class = 'boundary' AND NEW.type = 'administrative'
475            AND NEW.admin_level <= 4 AND NEW.osm_type = 'W'
476     THEN
477         RETURN NULL;
478     END IF;
479
480     SELECT * INTO NEW.rank_search, NEW.rank_address
481       FROM compute_place_rank(NEW.country_code,
482                               CASE WHEN is_area THEN 'A' ELSE NEW.osm_type END,
483                               NEW.class, NEW.type, NEW.admin_level,
484                               (NEW.extratags->'capital') = 'yes',
485                               NEW.address->'postcode');
486
487     -- a country code make no sense below rank 4 (country)
488     IF NEW.rank_search < 4 THEN
489       NEW.country_code := NULL;
490     END IF;
491
492   END IF;
493
494   {% if debug %}RAISE WARNING 'placex_insert:END: % % % %',NEW.osm_type,NEW.osm_id,NEW.class,NEW.type;{% endif %}
495
496 {% if not disable_diff_updates %}
497   -- The following is not needed until doing diff updates, and slows the main index process down
498
499   IF NEW.osm_type = 'N' and NEW.rank_search > 28 THEN
500       -- might be part of an interpolation
501       result := osmline_reinsert(NEW.osm_id, NEW.geometry);
502   ELSEIF NEW.rank_address > 0 THEN
503     IF (ST_GeometryType(NEW.geometry) in ('ST_Polygon','ST_MultiPolygon') AND ST_IsValid(NEW.geometry)) THEN
504       -- Performance: We just can't handle re-indexing for country level changes
505       IF st_area(NEW.geometry) < 1 THEN
506         -- mark items within the geometry for re-indexing
507   --    RAISE WARNING 'placex poly insert: % % % %',NEW.osm_type,NEW.osm_id,NEW.class,NEW.type;
508
509         UPDATE placex SET indexed_status = 2
510          WHERE ST_Intersects(NEW.geometry, placex.geometry)
511                and indexed_status = 0
512                and ((rank_address = 0 and rank_search > NEW.rank_address)
513                     or rank_address > NEW.rank_address
514                     or (class = 'place' and osm_type = 'N')
515                    )
516                and (rank_search < 28
517                     or name is not null
518                     or (NEW.rank_address >= 16 and address ? 'place'));
519       END IF;
520     ELSE
521       -- mark nearby items for re-indexing, where 'nearby' depends on the features rank_search and is a complete guess :(
522       diameter := update_place_diameter(NEW.rank_search);
523       IF diameter > 0 THEN
524   --      RAISE WARNING 'placex point insert: % % % % %',NEW.osm_type,NEW.osm_id,NEW.class,NEW.type,diameter;
525         IF NEW.rank_search >= 26 THEN
526           -- roads may cause reparenting for >27 rank places
527           update placex set indexed_status = 2 where indexed_status = 0 and rank_search > NEW.rank_search and ST_DWithin(placex.geometry, NEW.geometry, diameter);
528           -- reparenting also for OSM Interpolation Lines (and for Tiger?)
529           update location_property_osmline set indexed_status = 2 where indexed_status = 0 and ST_DWithin(location_property_osmline.linegeo, NEW.geometry, diameter);
530         ELSEIF NEW.rank_search >= 16 THEN
531           -- up to rank 16, street-less addresses may need reparenting
532           update placex set indexed_status = 2 where indexed_status = 0 and rank_search > NEW.rank_search and ST_DWithin(placex.geometry, NEW.geometry, diameter) and (rank_search < 28 or name is not null or address ? 'place');
533         ELSE
534           -- for all other places the search terms may change as well
535           update placex set indexed_status = 2 where indexed_status = 0 and rank_search > NEW.rank_search and ST_DWithin(placex.geometry, NEW.geometry, diameter) and (rank_search < 28 or name is not null);
536         END IF;
537       END IF;
538     END IF;
539   END IF;
540
541
542    -- add to tables for special search
543    -- Note: won't work on initial import because the classtype tables
544    -- do not yet exist. It won't hurt either.
545   classtable := 'place_classtype_' || NEW.class || '_' || NEW.type;
546   SELECT count(*)>0 FROM pg_tables WHERE tablename = classtable and schemaname = current_schema() INTO result;
547   IF result THEN
548     EXECUTE 'INSERT INTO ' || classtable::regclass || ' (place_id, centroid) VALUES ($1,$2)' 
549     USING NEW.place_id, ST_Centroid(NEW.geometry);
550   END IF;
551
552 {% endif %} -- not disable_diff_updates
553
554   RETURN NEW;
555
556 END;
557 $$
558 LANGUAGE plpgsql;
559
560 CREATE OR REPLACE FUNCTION placex_update()
561   RETURNS TRIGGER
562   AS $$
563 DECLARE
564   i INTEGER;
565   location RECORD;
566   relation_members TEXT[];
567
568   geom GEOMETRY;
569   parent_address_level SMALLINT;
570   place_address_level SMALLINT;
571
572   addr_street TEXT;
573   addr_place TEXT;
574
575   max_rank SMALLINT;
576
577   name_vector INTEGER[];
578   nameaddress_vector INTEGER[];
579   addr_nameaddress_vector INTEGER[];
580
581   linked_node_id BIGINT;
582   linked_importance FLOAT;
583   linked_wikipedia TEXT;
584
585   result BOOLEAN;
586 BEGIN
587   -- deferred delete
588   IF OLD.indexed_status = 100 THEN
589     {% if debug %}RAISE WARNING 'placex_update delete % %',NEW.osm_type,NEW.osm_id;{% endif %}
590     delete from placex where place_id = OLD.place_id;
591     RETURN NULL;
592   END IF;
593
594   IF NEW.indexed_status != 0 OR OLD.indexed_status = 0 THEN
595     RETURN NEW;
596   END IF;
597
598   {% if debug %}RAISE WARNING 'placex_update % % (%)',NEW.osm_type,NEW.osm_id,NEW.place_id;{% endif %}
599
600   NEW.indexed_date = now();
601
602   {% if 'search_name' in db.tables %}
603     DELETE from search_name WHERE place_id = NEW.place_id;
604   {% endif %}
605   result := deleteSearchName(NEW.partition, NEW.place_id);
606   DELETE FROM place_addressline WHERE place_id = NEW.place_id;
607   result := deleteRoad(NEW.partition, NEW.place_id);
608   result := deleteLocationArea(NEW.partition, NEW.place_id, NEW.rank_search);
609   UPDATE placex set linked_place_id = null, indexed_status = 2
610          where linked_place_id = NEW.place_id;
611   -- update not necessary for osmline, cause linked_place_id does not exist
612
613   NEW.extratags := NEW.extratags - 'linked_place'::TEXT;
614
615   IF NEW.linked_place_id is not null THEN
616     {% if debug %}RAISE WARNING 'place already linked to %', NEW.linked_place_id;{% endif %}
617     RETURN NEW;
618   END IF;
619
620   -- Postcodes are just here to compute the centroids. They are not searchable
621   -- unless they are a boundary=postal_code.
622   -- There was an error in the style so that boundary=postal_code used to be
623   -- imported as place=postcode. That's why relations are allowed to pass here.
624   -- This can go away in a couple of versions.
625   IF NEW.class = 'place'  and NEW.type = 'postcode' and NEW.osm_type != 'R' THEN
626     RETURN NEW;
627   END IF;
628
629   -- Speed up searches - just use the centroid of the feature
630   -- cheaper but less acurate
631   NEW.centroid := ST_PointOnSurface(NEW.geometry);
632   {% if debug %}RAISE WARNING 'Computing preliminary centroid at %',ST_AsText(NEW.centroid);{% endif %}
633
634   -- recompute the ranks, they might change when linking changes
635   SELECT * INTO NEW.rank_search, NEW.rank_address
636     FROM compute_place_rank(NEW.country_code,
637                             CASE WHEN ST_GeometryType(NEW.geometry)
638                                         IN ('ST_Polygon','ST_MultiPolygon')
639                             THEN 'A' ELSE NEW.osm_type END,
640                             NEW.class, NEW.type, NEW.admin_level,
641                             (NEW.extratags->'capital') = 'yes',
642                             NEW.address->'postcode');
643   -- We must always increase the address level relative to the admin boundary.
644   IF NEW.class = 'boundary' and NEW.type = 'administrative'
645      and NEW.osm_type = 'R' and NEW.rank_address > 0
646   THEN
647     -- First, check that admin boundaries do not overtake each other rank-wise.
648     parent_address_level := 3;
649     FOR location IN
650       SELECT rank_address,
651              (CASE WHEN extratags ? 'wikidata' and NEW.extratags ? 'wikidata'
652                         and extratags->'wikidata' = NEW.extratags->'wikidata'
653                    THEN ST_Equals(geometry, NEW.geometry)
654                    ELSE false END) as is_same
655       FROM placex
656       WHERE osm_type = 'R' and class = 'boundary' and type = 'administrative'
657             and admin_level < NEW.admin_level and admin_level > 3
658             and rank_address > 0
659             and geometry && NEW.centroid and _ST_Covers(geometry, NEW.centroid)
660       ORDER BY admin_level desc LIMIT 1
661     LOOP
662       IF location.is_same THEN
663         -- Looks like the same boundary is replicated on multiple admin_levels.
664         -- Usual tagging in Poland. Remove our boundary from addresses.
665         NEW.rank_address := 0;
666       ELSE
667         parent_address_level := location.rank_address;
668         IF location.rank_address >= NEW.rank_address THEN
669           IF location.rank_address >= 24 THEN
670             NEW.rank_address := 25;
671           ELSE
672             NEW.rank_address := location.rank_address + 2;
673           END IF;
674         END IF;
675       END IF;
676     END LOOP;
677
678     IF NEW.rank_address > 9 THEN
679         -- Second check that the boundary is not completely contained in a
680         -- place area with a higher address rank
681         FOR location IN
682           SELECT rank_address FROM placex
683           WHERE class = 'place' and rank_address < 24
684                 and rank_address > NEW.rank_address
685                 and geometry && NEW.geometry
686                 and geometry ~ NEW.geometry -- needed because ST_Relate does not do bbox cover test
687                 and ST_Relate(geometry, NEW.geometry, 'T*T***FF*') -- contains but not equal
688           ORDER BY rank_address desc LIMIT 1
689         LOOP
690           NEW.rank_address := location.rank_address + 2;
691         END LOOP;
692     END IF;
693   ELSEIF NEW.class = 'place' and NEW.osm_type = 'N'
694      and NEW.rank_address between 16 and 23
695   THEN
696     -- If a place node is contained in a admin boundary with the same address level
697     -- and has not been linked, then make the node a subpart by increasing the
698     -- address rank (city level and above).
699     FOR location IN
700         SELECT rank_address FROM placex
701         WHERE osm_type = 'R' and class = 'boundary' and type = 'administrative'
702               and rank_address = NEW.rank_address
703               and geometry && NEW.centroid and _ST_Covers(geometry, NEW.centroid)
704         LIMIT 1
705     LOOP
706       NEW.rank_address = NEW.rank_address + 2;
707     END LOOP;
708   ELSE
709     parent_address_level := 3;
710   END IF;
711
712   {% if debug %}RAISE WARNING 'Copy over address tags';{% endif %}
713   -- housenumber is a computed field, so start with an empty value
714   NEW.housenumber := NULL;
715   IF NEW.address is not NULL THEN
716       IF NEW.address ? 'conscriptionnumber' THEN
717         IF NEW.address ? 'streetnumber' THEN
718             NEW.housenumber := (NEW.address->'conscriptionnumber') || '/' || (NEW.address->'streetnumber');
719         ELSE
720             NEW.housenumber := NEW.address->'conscriptionnumber';
721         END IF;
722       ELSEIF NEW.address ? 'streetnumber' THEN
723         NEW.housenumber := NEW.address->'streetnumber';
724       ELSEIF NEW.address ? 'housenumber' THEN
725         NEW.housenumber := NEW.address->'housenumber';
726       END IF;
727       NEW.housenumber := create_housenumber_id(NEW.housenumber);
728
729       addr_street := NEW.address->'street';
730       addr_place := NEW.address->'place';
731
732       IF NEW.address ? 'postcode' and NEW.address->'postcode' not similar to '%(:|,|;)%' THEN
733         i := getorcreate_postcode_id(NEW.address->'postcode');
734       END IF;
735   END IF;
736
737   NEW.postcode := null;
738
739   -- recalculate country and partition
740   IF NEW.rank_search = 4 AND NEW.address is not NULL AND NEW.address ? 'country' THEN
741     -- for countries, believe the mapped country code,
742     -- so that we remain in the right partition if the boundaries
743     -- suddenly expand.
744     NEW.country_code := lower(NEW.address->'country');
745     NEW.partition := get_partition(lower(NEW.country_code));
746     IF NEW.partition = 0 THEN
747       NEW.country_code := lower(get_country_code(NEW.centroid));
748       NEW.partition := get_partition(NEW.country_code);
749     END IF;
750   ELSE
751     IF NEW.rank_search >= 4 THEN
752       NEW.country_code := lower(get_country_code(NEW.centroid));
753     ELSE
754       NEW.country_code := NULL;
755     END IF;
756     NEW.partition := get_partition(NEW.country_code);
757   END IF;
758   {% if debug %}RAISE WARNING 'Country updated: "%"', NEW.country_code;{% endif %}
759
760   -- waterway ways are linked when they are part of a relation and have the same class/type
761   IF NEW.osm_type = 'R' and NEW.class = 'waterway' THEN
762       FOR relation_members IN select members from planet_osm_rels r where r.id = NEW.osm_id and r.parts != array[]::bigint[]
763       LOOP
764           FOR i IN 1..array_upper(relation_members, 1) BY 2 LOOP
765               IF relation_members[i+1] in ('', 'main_stream', 'side_stream') AND substring(relation_members[i],1,1) = 'w' THEN
766                 {% if debug %}RAISE WARNING 'waterway parent %, child %/%', NEW.osm_id, i, relation_members[i];{% endif %}
767                 FOR linked_node_id IN SELECT place_id FROM placex
768                   WHERE osm_type = 'W' and osm_id = substring(relation_members[i],2,200)::bigint
769                   and class = NEW.class and type in ('river', 'stream', 'canal', 'drain', 'ditch')
770                   and ( relation_members[i+1] != 'side_stream' or NEW.name->'name' = name->'name')
771                 LOOP
772                   UPDATE placex SET linked_place_id = NEW.place_id WHERE place_id = linked_node_id;
773                   {% if 'search_name' in db.tables %}
774                     DELETE FROM search_name WHERE place_id = linked_node_id;
775                   {% endif %}
776                 END LOOP;
777               END IF;
778           END LOOP;
779       END LOOP;
780       {% if debug %}RAISE WARNING 'Waterway processed';{% endif %}
781   END IF;
782
783   NEW.importance := null;
784   SELECT wikipedia, importance
785     FROM compute_importance(NEW.extratags, NEW.country_code, NEW.osm_type, NEW.osm_id)
786     INTO NEW.wikipedia,NEW.importance;
787
788 {% if debug %}RAISE WARNING 'Importance computed from wikipedia: %', NEW.importance;{% endif %}
789
790   -- ---------------------------------------------------------------------------
791   -- For low level elements we inherit from our parent road
792   IF NEW.rank_search > 27 THEN
793
794     {% if debug %}RAISE WARNING 'finding street for % %', NEW.osm_type, NEW.osm_id;{% endif %}
795     NEW.parent_place_id := null;
796
797     -- We have to find our parent road.
798     NEW.parent_place_id := find_parent_for_poi(NEW.osm_type, NEW.osm_id,
799                                                NEW.partition,
800                                                ST_Envelope(NEW.geometry),
801                                                addr_street, addr_place);
802
803     -- If we found the road take a shortcut here.
804     -- Otherwise fall back to the full address getting method below.
805     IF NEW.parent_place_id is not null THEN
806
807       -- Get the details of the parent road
808       SELECT p.country_code, p.postcode, p.name FROM placex p
809        WHERE p.place_id = NEW.parent_place_id INTO location;
810
811       IF addr_street is null and addr_place is not null THEN
812         -- Check if the addr:place tag is part of the parent name
813         SELECT count(*) INTO i
814           FROM svals(location.name) AS pname WHERE pname = addr_place;
815         IF i = 0 THEN
816           NEW.address = NEW.address || hstore('_unlisted_place', addr_place);
817         END IF;
818       END IF;
819
820       NEW.country_code := location.country_code;
821       {% if debug %}RAISE WARNING 'Got parent details from search name';{% endif %}
822
823       -- determine postcode
824       IF NEW.address is not null AND NEW.address ? 'postcode' THEN
825           NEW.postcode = upper(trim(NEW.address->'postcode'));
826       ELSE
827          NEW.postcode := location.postcode;
828       END IF;
829       IF NEW.postcode is null THEN
830         NEW.postcode := get_nearest_postcode(NEW.country_code, NEW.geometry);
831       END IF;
832
833       IF NEW.name is not NULL THEN
834           NEW.name := add_default_place_name(NEW.country_code, NEW.name);
835           name_vector := make_keywords(NEW.name);
836
837           IF NEW.rank_search <= 25 and NEW.rank_address > 0 THEN
838             result := add_location(NEW.place_id, NEW.country_code, NEW.partition,
839                                    name_vector, NEW.rank_search, NEW.rank_address,
840                                    upper(trim(NEW.address->'postcode')), NEW.geometry,
841                                    NEW.centroid);
842             {% if debug %}RAISE WARNING 'Place added to location table';{% endif %}
843           END IF;
844
845       END IF;
846
847       {% if not db.reverse_only %}
848       IF array_length(name_vector, 1) is not NULL
849          OR NEW.address is not NULL
850       THEN
851         SELECT * INTO name_vector, nameaddress_vector
852           FROM create_poi_search_terms(NEW.place_id,
853                                        NEW.partition, NEW.parent_place_id,
854                                        NEW.address,
855                                        NEW.country_code, NEW.housenumber,
856                                        name_vector, NEW.centroid);
857
858         IF array_length(name_vector, 1) is not NULL THEN
859           INSERT INTO search_name (place_id, search_rank, address_rank,
860                                    importance, country_code, name_vector,
861                                    nameaddress_vector, centroid)
862                  VALUES (NEW.place_id, NEW.rank_search, NEW.rank_address,
863                          NEW.importance, NEW.country_code, name_vector,
864                          nameaddress_vector, NEW.centroid);
865           {% if debug %}RAISE WARNING 'Place added to search table';{% endif %}
866         END IF;
867       END IF;
868       {% endif %}
869
870       -- If the address was inherited from a surrounding building,
871       -- do not add it permanently to the table.
872       IF NEW.address ? '_inherited' THEN
873         IF NEW.address ? '_unlisted_place' THEN
874           NEW.address := hstore('_unlisted_place', NEW.address->'_unlisted_place');
875         ELSE
876           NEW.address := null;
877         END IF;
878       END IF;
879
880       RETURN NEW;
881     END IF;
882
883   END IF;
884
885   -- ---------------------------------------------------------------------------
886   -- Full indexing
887   {% if debug %}RAISE WARNING 'Using full index mode for % %', NEW.osm_type, NEW.osm_id;{% endif %}
888   SELECT * INTO location FROM find_linked_place(NEW);
889   IF location.place_id is not null THEN
890     {% if debug %}RAISE WARNING 'Linked %', location;{% endif %}
891
892     -- Use the linked point as the centre point of the geometry,
893     -- but only if it is within the area of the boundary.
894     geom := coalesce(location.centroid, ST_Centroid(location.geometry));
895     IF geom is not NULL AND ST_Within(geom, NEW.geometry) THEN
896         NEW.centroid := geom;
897     END IF;
898
899     {% if debug %}RAISE WARNING 'parent address: % rank address: %', parent_address_level, location.rank_address;{% endif %}
900     IF location.rank_address > parent_address_level
901        and location.rank_address < 26
902     THEN
903       NEW.rank_address := location.rank_address;
904     END IF;
905
906     -- merge in the label name
907     IF NOT location.name IS NULL THEN
908       NEW.name := location.name || NEW.name;
909     END IF;
910
911     -- merge in extra tags
912     NEW.extratags := hstore('linked_' || location.class, location.type)
913                      || coalesce(location.extratags, ''::hstore)
914                      || coalesce(NEW.extratags, ''::hstore);
915
916     -- mark the linked place (excludes from search results)
917     UPDATE placex set linked_place_id = NEW.place_id
918       WHERE place_id = location.place_id;
919     -- ensure that those places are not found anymore
920     {% if 'search_name' in db.tables %}
921       DELETE FROM search_name WHERE place_id = location.place_id;
922     {% endif %}
923     PERFORM deleteLocationArea(NEW.partition, location.place_id, NEW.rank_search);
924
925     SELECT wikipedia, importance
926       FROM compute_importance(location.extratags, NEW.country_code,
927                               'N', location.osm_id)
928       INTO linked_wikipedia,linked_importance;
929
930     -- Use the maximum importance if one could be computed from the linked object.
931     IF linked_importance is not null AND
932        (NEW.importance is null or NEW.importance < linked_importance)
933     THEN
934       NEW.importance = linked_importance;
935     END IF;
936   ELSE
937     -- No linked place? As a last resort check if the boundary is tagged with
938     -- a place type and adapt the rank address.
939     IF NEW.rank_address > 0 and NEW.extratags ? 'place' THEN
940       SELECT address_rank INTO place_address_level
941         FROM compute_place_rank(NEW.country_code, 'A', 'place',
942                                 NEW.extratags->'place', 0::SMALLINT, False, null);
943       IF place_address_level > parent_address_level and
944          place_address_level < 26 THEN
945         NEW.rank_address := place_address_level;
946       END IF;
947     END IF;
948   END IF;
949
950   -- Initialise the name vector using our name
951   NEW.name := add_default_place_name(NEW.country_code, NEW.name);
952   name_vector := make_keywords(NEW.name);
953
954   -- make sure all names are in the word table
955   IF NEW.admin_level = 2
956      AND NEW.class = 'boundary' AND NEW.type = 'administrative'
957      AND NEW.country_code IS NOT NULL AND NEW.osm_type = 'R'
958   THEN
959     PERFORM create_country(NEW.name, lower(NEW.country_code));
960     {% if debug %}RAISE WARNING 'Country names updated';{% endif %}
961
962     -- Also update the list of country names. Adding an additional sanity
963     -- check here: make sure the country does overlap with the area where
964     -- we expect it to be as per static country grid.
965     FOR location IN
966       SELECT country_code FROM country_osm_grid
967        WHERE ST_Covers(geometry, NEW.centroid) and country_code = NEW.country_code
968        LIMIT 1
969     LOOP
970       {% if debug %}RAISE WARNING 'Updating names for country '%' with: %', NEW.country_code, NEW.name;{% endif %}
971       UPDATE country_name SET name = name || NEW.name WHERE country_code = NEW.country_code;
972     END LOOP;
973   END IF;
974
975   -- For linear features we need the full geometry for determining the address
976   -- because they may go through several administrative entities. Otherwise use
977   -- the centroid for performance reasons.
978   IF ST_GeometryType(NEW.geometry) in ('ST_LineString', 'ST_MultiLineString') THEN
979     geom := NEW.geometry;
980   ELSE
981     geom := NEW.centroid;
982   END IF;
983
984   IF NEW.rank_address = 0 THEN
985     max_rank := geometry_to_rank(NEW.rank_search, NEW.geometry, NEW.country_code);
986     -- Rank 0 features may also span multiple administrative areas (e.g. lakes)
987     -- so use the geometry here too. Just make sure the areas don't become too
988     -- large.
989     IF NEW.class = 'natural' or max_rank > 10 THEN
990       geom := NEW.geometry;
991     END IF;
992   ELSEIF NEW.rank_address > 25 THEN
993     max_rank := 25;
994   ELSE
995     max_rank = NEW.rank_address;
996   END IF;
997
998   SELECT * FROM insert_addresslines(NEW.place_id, NEW.partition, max_rank,
999                                     NEW.address, geom, NEW.country_code)
1000     INTO NEW.parent_place_id, NEW.postcode, nameaddress_vector;
1001
1002   {% if debug %}RAISE WARNING 'RETURN insert_addresslines: %, %, %', NEW.parent_place_id, NEW.postcode, nameaddress_vector;{% endif %}
1003
1004   IF NEW.address is not null AND NEW.address ? 'postcode' 
1005      AND NEW.address->'postcode' not similar to '%(,|;)%' THEN
1006     NEW.postcode := upper(trim(NEW.address->'postcode'));
1007   END IF;
1008
1009   IF NEW.postcode is null AND NEW.rank_search > 8 THEN
1010     NEW.postcode := get_nearest_postcode(NEW.country_code, NEW.geometry);
1011   END IF;
1012
1013   -- if we have a name add this to the name search table
1014   IF NEW.name IS NOT NULL THEN
1015
1016     IF NEW.rank_search <= 25 and NEW.rank_address > 0 THEN
1017       result := add_location(NEW.place_id, NEW.country_code, NEW.partition, name_vector, NEW.rank_search, NEW.rank_address, upper(trim(NEW.address->'postcode')), NEW.geometry, NEW.centroid);
1018       {% if debug %}RAISE WARNING 'added to location (full)';{% endif %}
1019     END IF;
1020
1021     IF NEW.rank_search between 26 and 27 and NEW.class = 'highway' THEN
1022       result := insertLocationRoad(NEW.partition, NEW.place_id, NEW.country_code, NEW.geometry);
1023       {% if debug %}RAISE WARNING 'insert into road location table (full)';{% endif %}
1024     END IF;
1025
1026     result := insertSearchName(NEW.partition, NEW.place_id, name_vector,
1027                                NEW.rank_search, NEW.rank_address, NEW.geometry);
1028     {% if debug %}RAISE WARNING 'added to search name (full)';{% endif %}
1029
1030     {% if not db.reverse_only %}
1031         INSERT INTO search_name (place_id, search_rank, address_rank,
1032                                  importance, country_code, name_vector,
1033                                  nameaddress_vector, centroid)
1034                VALUES (NEW.place_id, NEW.rank_search, NEW.rank_address,
1035                        NEW.importance, NEW.country_code, name_vector,
1036                        nameaddress_vector, NEW.centroid);
1037     {% endif %}
1038
1039   END IF;
1040
1041   {% if debug %}RAISE WARNING 'place update % % finsihed.', NEW.osm_type, NEW.osm_id;{% endif %}
1042
1043   RETURN NEW;
1044 END;
1045 $$
1046 LANGUAGE plpgsql;
1047
1048
1049 CREATE OR REPLACE FUNCTION placex_delete()
1050   RETURNS TRIGGER
1051   AS $$
1052 DECLARE
1053   b BOOLEAN;
1054   classtable TEXT;
1055 BEGIN
1056   -- RAISE WARNING 'placex_delete % %',OLD.osm_type,OLD.osm_id;
1057
1058   IF OLD.linked_place_id is null THEN
1059     update placex set linked_place_id = null, indexed_status = 2 where linked_place_id = OLD.place_id and indexed_status = 0;
1060     {% if debug %}RAISE WARNING 'placex_delete:01 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1061     update placex set linked_place_id = null where linked_place_id = OLD.place_id;
1062     {% if debug %}RAISE WARNING 'placex_delete:02 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1063   ELSE
1064     update placex set indexed_status = 2 where place_id = OLD.linked_place_id and indexed_status = 0;
1065   END IF;
1066
1067   IF OLD.rank_address < 30 THEN
1068
1069     -- mark everything linked to this place for re-indexing
1070     {% if debug %}RAISE WARNING 'placex_delete:03 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1071     UPDATE placex set indexed_status = 2 from place_addressline where address_place_id = OLD.place_id 
1072       and placex.place_id = place_addressline.place_id and indexed_status = 0 and place_addressline.isaddress;
1073
1074     {% if debug %}RAISE WARNING 'placex_delete:04 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1075     DELETE FROM place_addressline where address_place_id = OLD.place_id;
1076
1077     {% if debug %}RAISE WARNING 'placex_delete:05 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1078     b := deleteRoad(OLD.partition, OLD.place_id);
1079
1080     {% if debug %}RAISE WARNING 'placex_delete:06 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1081     update placex set indexed_status = 2 where parent_place_id = OLD.place_id and indexed_status = 0;
1082     {% if debug %}RAISE WARNING 'placex_delete:07 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1083     -- reparenting also for OSM Interpolation Lines (and for Tiger?)
1084     update location_property_osmline set indexed_status = 2 where indexed_status = 0 and parent_place_id = OLD.place_id;
1085
1086   END IF;
1087
1088   {% if debug %}RAISE WARNING 'placex_delete:08 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1089
1090   IF OLD.rank_address < 26 THEN
1091     b := deleteLocationArea(OLD.partition, OLD.place_id, OLD.rank_search);
1092   END IF;
1093
1094   {% if debug %}RAISE WARNING 'placex_delete:09 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1095
1096   IF OLD.name is not null THEN
1097     {% if 'search_name' in db.tables %}
1098       DELETE from search_name WHERE place_id = OLD.place_id;
1099     {% endif %}
1100     b := deleteSearchName(OLD.partition, OLD.place_id);
1101   END IF;
1102
1103   {% if debug %}RAISE WARNING 'placex_delete:10 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1104
1105   DELETE FROM place_addressline where place_id = OLD.place_id;
1106
1107   {% if debug %}RAISE WARNING 'placex_delete:11 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1108
1109   -- remove from tables for special search
1110   classtable := 'place_classtype_' || OLD.class || '_' || OLD.type;
1111   SELECT count(*)>0 FROM pg_tables WHERE tablename = classtable and schemaname = current_schema() INTO b;
1112   IF b THEN
1113     EXECUTE 'DELETE FROM ' || classtable::regclass || ' WHERE place_id = $1' USING OLD.place_id;
1114   END IF;
1115
1116   {% if debug %}RAISE WARNING 'placex_delete:12 % %',OLD.osm_type,OLD.osm_id;{% endif %}
1117
1118   RETURN OLD;
1119
1120 END;
1121 $$
1122 LANGUAGE plpgsql;