]> git.openstreetmap.org Git - nominatim.git/blob - nominatim/api/search/db_searches.py
more unit tests for search
[nominatim.git] / nominatim / api / search / db_searches.py
1 # SPDX-License-Identifier: GPL-3.0-or-later
2 #
3 # This file is part of Nominatim. (https://nominatim.org)
4 #
5 # Copyright (C) 2023 by the Nominatim developer community.
6 # For a full list of authors see the git log.
7 """
8 Implementation of the acutal database accesses for forward search.
9 """
10 from typing import List, Tuple, AsyncIterator, Dict, Any, Callable
11 import abc
12
13 import sqlalchemy as sa
14 from sqlalchemy.dialects.postgresql import ARRAY, array_agg
15
16 from nominatim.typing import SaFromClause, SaScalarSelect, SaColumn, \
17                              SaExpression, SaSelect, SaLambdaSelect, SaRow, SaBind
18 from nominatim.api.connection import SearchConnection
19 from nominatim.api.types import SearchDetails, DataLayer, GeometryFormat, Bbox
20 import nominatim.api.results as nres
21 from nominatim.api.search.db_search_fields import SearchData, WeightedCategories
22 from nominatim.db.sqlalchemy_types import Geometry
23
24 #pylint: disable=singleton-comparison,not-callable
25 #pylint: disable=too-many-branches,too-many-arguments,too-many-locals,too-many-statements
26
27 def _details_to_bind_params(details: SearchDetails) -> Dict[str, Any]:
28     """ Create a dictionary from search parameters that can be used
29         as bind parameter for SQL execute.
30     """
31     return {'limit': details.max_results,
32             'min_rank': details.min_rank,
33             'max_rank': details.max_rank,
34             'viewbox': details.viewbox,
35             'viewbox2': details.viewbox_x2,
36             'near': details.near,
37             'near_radius': details.near_radius,
38             'excluded': details.excluded,
39             'countries': details.countries}
40
41
42 LIMIT_PARAM: SaBind = sa.bindparam('limit')
43 MIN_RANK_PARAM: SaBind = sa.bindparam('min_rank')
44 MAX_RANK_PARAM: SaBind = sa.bindparam('max_rank')
45 VIEWBOX_PARAM: SaBind = sa.bindparam('viewbox', type_=Geometry)
46 VIEWBOX2_PARAM: SaBind = sa.bindparam('viewbox2', type_=Geometry)
47 NEAR_PARAM: SaBind = sa.bindparam('near', type_=Geometry)
48 NEAR_RADIUS_PARAM: SaBind = sa.bindparam('near_radius')
49 COUNTRIES_PARAM: SaBind = sa.bindparam('countries')
50
51 def _within_near(t: SaFromClause) -> Callable[[], SaExpression]:
52     return lambda: t.c.geometry.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM)
53
54 def _exclude_places(t: SaFromClause) -> Callable[[], SaExpression]:
55     return lambda: t.c.place_id.not_in(sa.bindparam('excluded'))
56
57 def _select_placex(t: SaFromClause) -> SaSelect:
58     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
59                      t.c.class_, t.c.type,
60                      t.c.address, t.c.extratags,
61                      t.c.housenumber, t.c.postcode, t.c.country_code,
62                      t.c.importance, t.c.wikipedia,
63                      t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
64                      t.c.linked_place_id, t.c.admin_level,
65                      t.c.centroid,
66                      t.c.geometry.ST_Expand(0).label('bbox'))
67
68
69 def _add_geometry_columns(sql: SaLambdaSelect, col: SaColumn, details: SearchDetails) -> SaSelect:
70     out = []
71
72     if details.geometry_simplification > 0.0:
73         col = sa.func.ST_SimplifyPreserveTopology(col, details.geometry_simplification)
74
75     if details.geometry_output & GeometryFormat.GEOJSON:
76         out.append(sa.func.ST_AsGeoJSON(col).label('geometry_geojson'))
77     if details.geometry_output & GeometryFormat.TEXT:
78         out.append(sa.func.ST_AsText(col).label('geometry_text'))
79     if details.geometry_output & GeometryFormat.KML:
80         out.append(sa.func.ST_AsKML(col).label('geometry_kml'))
81     if details.geometry_output & GeometryFormat.SVG:
82         out.append(sa.func.ST_AsSVG(col).label('geometry_svg'))
83
84     return sql.add_columns(*out)
85
86
87 def _make_interpolation_subquery(table: SaFromClause, inner: SaFromClause,
88                                  numerals: List[int], details: SearchDetails) -> SaScalarSelect:
89     all_ids = array_agg(table.c.place_id) # type: ignore[no-untyped-call]
90     sql = sa.select(all_ids).where(table.c.parent_place_id == inner.c.place_id)
91
92     if len(numerals) == 1:
93         sql = sql.where(sa.between(numerals[0], table.c.startnumber, table.c.endnumber))\
94                  .where((numerals[0] - table.c.startnumber) % table.c.step == 0)
95     else:
96         sql = sql.where(sa.or_(
97                 *(sa.and_(sa.between(n, table.c.startnumber, table.c.endnumber),
98                           (n - table.c.startnumber) % table.c.step == 0)
99                   for n in numerals)))
100
101     if details.excluded:
102         sql = sql.where(_exclude_places(table))
103
104     return sql.scalar_subquery()
105
106
107 def _filter_by_layer(table: SaFromClause, layers: DataLayer) -> SaColumn:
108     orexpr: List[SaExpression] = []
109     if layers & DataLayer.ADDRESS and layers & DataLayer.POI:
110         orexpr.append(table.c.rank_address.between(1, 30))
111     elif layers & DataLayer.ADDRESS:
112         orexpr.append(table.c.rank_address.between(1, 29))
113         orexpr.append(sa.and_(table.c.rank_address == 30,
114                               sa.or_(table.c.housenumber != None,
115                                      table.c.address.has_key('addr:housename'))))
116     elif layers & DataLayer.POI:
117         orexpr.append(sa.and_(table.c.rank_address == 30,
118                               table.c.class_.not_in(('place', 'building'))))
119
120     if layers & DataLayer.MANMADE:
121         exclude = []
122         if not layers & DataLayer.RAILWAY:
123             exclude.append('railway')
124         if not layers & DataLayer.NATURAL:
125             exclude.extend(('natural', 'water', 'waterway'))
126         orexpr.append(sa.and_(table.c.class_.not_in(tuple(exclude)),
127                               table.c.rank_address == 0))
128     else:
129         include = []
130         if layers & DataLayer.RAILWAY:
131             include.append('railway')
132         if layers & DataLayer.NATURAL:
133             include.extend(('natural', 'water', 'waterway'))
134         orexpr.append(sa.and_(table.c.class_.in_(tuple(include)),
135                               table.c.rank_address == 0))
136
137     if len(orexpr) == 1:
138         return orexpr[0]
139
140     return sa.or_(*orexpr)
141
142
143 def _interpolated_position(table: SaFromClause, nr: SaColumn) -> SaColumn:
144     pos = sa.cast(nr - table.c.startnumber, sa.Float) / (table.c.endnumber - table.c.startnumber)
145     return sa.case(
146             (table.c.endnumber == table.c.startnumber, table.c.linegeo.ST_Centroid()),
147             else_=table.c.linegeo.ST_LineInterpolatePoint(pos)).label('centroid')
148
149
150 async def _get_placex_housenumbers(conn: SearchConnection,
151                                    place_ids: List[int],
152                                    details: SearchDetails) -> AsyncIterator[nres.SearchResult]:
153     t = conn.t.placex
154     sql = _select_placex(t).where(t.c.place_id.in_(place_ids))
155
156     if details.geometry_output:
157         sql = _add_geometry_columns(sql, t.c.geometry, details)
158
159     for row in await conn.execute(sql):
160         result = nres.create_from_placex_row(row, nres.SearchResult)
161         assert result
162         result.bbox = Bbox.from_wkb(row.bbox)
163         yield result
164
165
166 async def _get_osmline(conn: SearchConnection, place_ids: List[int],
167                        numerals: List[int],
168                        details: SearchDetails) -> AsyncIterator[nres.SearchResult]:
169     t = conn.t.osmline
170     values = sa.values(sa.Column('nr', sa.Integer()), name='housenumber')\
171                .data([(n,) for n in numerals])
172     sql = sa.select(t.c.place_id, t.c.osm_id,
173                     t.c.parent_place_id, t.c.address,
174                     values.c.nr.label('housenumber'),
175                     _interpolated_position(t, values.c.nr),
176                     t.c.postcode, t.c.country_code)\
177             .where(t.c.place_id.in_(place_ids))\
178             .join(values, values.c.nr.between(t.c.startnumber, t.c.endnumber))
179
180     if details.geometry_output:
181         sub = sql.subquery()
182         sql = _add_geometry_columns(sa.select(sub), sub.c.centroid, details)
183
184     for row in await conn.execute(sql):
185         result = nres.create_from_osmline_row(row, nres.SearchResult)
186         assert result
187         yield result
188
189
190 async def _get_tiger(conn: SearchConnection, place_ids: List[int],
191                      numerals: List[int], osm_id: int,
192                      details: SearchDetails) -> AsyncIterator[nres.SearchResult]:
193     t = conn.t.tiger
194     values = sa.values(sa.Column('nr', sa.Integer()), name='housenumber')\
195                .data([(n,) for n in numerals])
196     sql = sa.select(t.c.place_id, t.c.parent_place_id,
197                     sa.literal('W').label('osm_type'),
198                     sa.literal(osm_id).label('osm_id'),
199                     values.c.nr.label('housenumber'),
200                     _interpolated_position(t, values.c.nr),
201                     t.c.postcode)\
202             .where(t.c.place_id.in_(place_ids))\
203             .join(values, values.c.nr.between(t.c.startnumber, t.c.endnumber))
204
205     if details.geometry_output:
206         sub = sql.subquery()
207         sql = _add_geometry_columns(sa.select(sub), sub.c.centroid, details)
208
209     for row in await conn.execute(sql):
210         result = nres.create_from_tiger_row(row, nres.SearchResult)
211         assert result
212         yield result
213
214
215 class AbstractSearch(abc.ABC):
216     """ Encapuslation of a single lookup in the database.
217     """
218
219     def __init__(self, penalty: float) -> None:
220         self.penalty = penalty
221
222     @abc.abstractmethod
223     async def lookup(self, conn: SearchConnection,
224                      details: SearchDetails) -> nres.SearchResults:
225         """ Find results for the search in the database.
226         """
227
228
229 class NearSearch(AbstractSearch):
230     """ Category search of a place type near the result of another search.
231     """
232     def __init__(self, penalty: float, categories: WeightedCategories,
233                  search: AbstractSearch) -> None:
234         super().__init__(penalty)
235         self.search = search
236         self.categories = categories
237
238
239     async def lookup(self, conn: SearchConnection,
240                      details: SearchDetails) -> nres.SearchResults:
241         """ Find results for the search in the database.
242         """
243         results = nres.SearchResults()
244         base = await self.search.lookup(conn, details)
245
246         if not base:
247             return results
248
249         base.sort(key=lambda r: (r.accuracy, r.rank_search))
250         max_accuracy = base[0].accuracy + 0.5
251         base = nres.SearchResults(r for r in base if r.source_table == nres.SourceTable.PLACEX
252                                                      and r.accuracy <= max_accuracy
253                                                      and r.bbox and r.bbox.area < 20)
254
255         if base:
256             baseids = [b.place_id for b in base[:5] if b.place_id]
257
258             for category, penalty in self.categories:
259                 await self.lookup_category(results, conn, baseids, category, penalty, details)
260                 if len(results) >= details.max_results:
261                     break
262
263         return results
264
265
266     async def lookup_category(self, results: nres.SearchResults,
267                               conn: SearchConnection, ids: List[int],
268                               category: Tuple[str, str], penalty: float,
269                               details: SearchDetails) -> None:
270         """ Find places of the given category near the list of
271             place ids and add the results to 'results'.
272         """
273         table = await conn.get_class_table(*category)
274
275         t = conn.t.placex
276         tgeom = conn.t.placex.alias('pgeom')
277
278         sql = _select_placex(t).where(tgeom.c.place_id.in_(ids))\
279                                .where(t.c.class_ == category[0])\
280                                .where(t.c.type == category[1])
281
282         if table is None:
283             # No classtype table available, do a simplified lookup in placex.
284             sql = sql.join(tgeom, t.c.geometry.ST_DWithin(tgeom.c.centroid, 0.01))\
285                      .order_by(tgeom.c.centroid.ST_Distance(t.c.centroid))
286         else:
287             # Use classtype table. We can afford to use a larger
288             # radius for the lookup.
289             sql = sql.join(table, t.c.place_id == table.c.place_id)\
290                      .join(tgeom,
291                            table.c.centroid.ST_CoveredBy(
292                                sa.case((sa.and_(tgeom.c.rank_address < 9,
293                                                 tgeom.c.geometry.is_area()),
294                                         tgeom.c.geometry),
295                                        else_ = tgeom.c.centroid.ST_Expand(0.05))))\
296                      .order_by(tgeom.c.centroid.ST_Distance(table.c.centroid))
297
298         sql = sql.where(t.c.rank_address.between(MIN_RANK_PARAM, MAX_RANK_PARAM))
299         if details.countries:
300             sql = sql.where(t.c.country_code.in_(COUNTRIES_PARAM))
301         if details.excluded:
302             sql = sql.where(_exclude_places(t))
303         if details.layers is not None:
304             sql = sql.where(_filter_by_layer(t, details.layers))
305
306         sql = sql.limit(LIMIT_PARAM)
307         for row in await conn.execute(sql, _details_to_bind_params(details)):
308             result = nres.create_from_placex_row(row, nres.SearchResult)
309             assert result
310             result.accuracy = self.penalty + penalty
311             result.bbox = Bbox.from_wkb(row.bbox)
312             results.append(result)
313
314
315
316 class PoiSearch(AbstractSearch):
317     """ Category search in a geographic area.
318     """
319     def __init__(self, sdata: SearchData) -> None:
320         super().__init__(sdata.penalty)
321         self.qualifiers = sdata.qualifiers
322         self.countries = sdata.countries
323
324
325     async def lookup(self, conn: SearchConnection,
326                      details: SearchDetails) -> nres.SearchResults:
327         """ Find results for the search in the database.
328         """
329         bind_params = _details_to_bind_params(details)
330         t = conn.t.placex
331
332         rows: List[SaRow] = []
333
334         if details.near and details.near_radius is not None and details.near_radius < 0.2:
335             # simply search in placex table
336             def _base_query() -> SaSelect:
337                 return _select_placex(t) \
338                            .where(t.c.linked_place_id == None) \
339                            .where(t.c.geometry.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM)) \
340                            .order_by(t.c.centroid.ST_Distance(NEAR_PARAM)) \
341                            .limit(LIMIT_PARAM)
342
343             classtype = self.qualifiers.values
344             if len(classtype) == 1:
345                 cclass, ctype = classtype[0]
346                 sql: SaLambdaSelect = sa.lambda_stmt(lambda: _base_query()
347                                                  .where(t.c.class_ == cclass)
348                                                  .where(t.c.type == ctype))
349             else:
350                 sql = _base_query().where(sa.or_(*(sa.and_(t.c.class_ == cls, t.c.type == typ)
351                                                    for cls, typ in classtype)))
352
353             if self.countries:
354                 sql = sql.where(t.c.country_code.in_(self.countries.values))
355
356             if details.viewbox is not None and details.bounded_viewbox:
357                 sql = sql.where(t.c.geometry.intersects(VIEWBOX_PARAM))
358
359             rows.extend(await conn.execute(sql, bind_params))
360         else:
361             # use the class type tables
362             for category in self.qualifiers.values:
363                 table = await conn.get_class_table(*category)
364                 if table is not None:
365                     sql = _select_placex(t)\
366                                .join(table, t.c.place_id == table.c.place_id)\
367                                .where(t.c.class_ == category[0])\
368                                .where(t.c.type == category[1])
369
370                     if details.viewbox is not None and details.bounded_viewbox:
371                         sql = sql.where(table.c.centroid.intersects(VIEWBOX_PARAM))
372
373                     if details.near and details.near_radius is not None:
374                         sql = sql.order_by(table.c.centroid.ST_Distance(NEAR_PARAM))\
375                                  .where(table.c.centroid.ST_DWithin(NEAR_PARAM,
376                                                                     NEAR_RADIUS_PARAM))
377
378                     if self.countries:
379                         sql = sql.where(t.c.country_code.in_(self.countries.values))
380
381                     sql = sql.limit(LIMIT_PARAM)
382                     rows.extend(await conn.execute(sql, bind_params))
383
384         results = nres.SearchResults()
385         for row in rows:
386             result = nres.create_from_placex_row(row, nres.SearchResult)
387             assert result
388             result.accuracy = self.penalty + self.qualifiers.get_penalty((row.class_, row.type))
389             result.bbox = Bbox.from_wkb(row.bbox)
390             results.append(result)
391
392         return results
393
394
395 class CountrySearch(AbstractSearch):
396     """ Search for a country name or country code.
397     """
398     def __init__(self, sdata: SearchData) -> None:
399         super().__init__(sdata.penalty)
400         self.countries = sdata.countries
401
402
403     async def lookup(self, conn: SearchConnection,
404                      details: SearchDetails) -> nres.SearchResults:
405         """ Find results for the search in the database.
406         """
407         t = conn.t.placex
408
409         ccodes = self.countries.values
410         sql = _select_placex(t)\
411                 .where(t.c.country_code.in_(ccodes))\
412                 .where(t.c.rank_address == 4)
413
414         if details.geometry_output:
415             sql = _add_geometry_columns(sql, t.c.geometry, details)
416
417         if details.excluded:
418             sql = sql.where(_exclude_places(t))
419
420         if details.viewbox is not None and details.bounded_viewbox:
421             sql = sql.where(lambda: t.c.geometry.intersects(VIEWBOX_PARAM))
422
423         if details.near is not None and details.near_radius is not None:
424             sql = sql.where(_within_near(t))
425
426         results = nres.SearchResults()
427         for row in await conn.execute(sql, _details_to_bind_params(details)):
428             result = nres.create_from_placex_row(row, nres.SearchResult)
429             assert result
430             result.accuracy = self.penalty + self.countries.get_penalty(row.country_code, 5.0)
431             result.bbox = Bbox.from_wkb(row.bbox)
432             results.append(result)
433
434         return results or await self.lookup_in_country_table(conn, details)
435
436
437     async def lookup_in_country_table(self, conn: SearchConnection,
438                                       details: SearchDetails) -> nres.SearchResults:
439         """ Look up the country in the fallback country tables.
440         """
441         # Avoid the fallback search when this is a more search. Country results
442         # usually are in the first batch of results and it is not possible
443         # to exclude these fallbacks.
444         if details.excluded:
445             return nres.SearchResults()
446
447         t = conn.t.country_name
448         tgrid = conn.t.country_grid
449
450         sql = sa.select(tgrid.c.country_code,
451                         tgrid.c.geometry.ST_Centroid().ST_Collect().ST_Centroid()
452                               .label('centroid'),
453                         tgrid.c.geometry.ST_Collect().ST_Expand(0).label('bbox'))\
454                 .where(tgrid.c.country_code.in_(self.countries.values))\
455                 .group_by(tgrid.c.country_code)
456
457         if details.viewbox is not None and details.bounded_viewbox:
458             sql = sql.where(tgrid.c.geometry.intersects(VIEWBOX_PARAM))
459         if details.near is not None and details.near_radius is not None:
460             sql = sql.where(_within_near(tgrid))
461
462         sub = sql.subquery('grid')
463
464         sql = sa.select(t.c.country_code,
465                         (t.c.name
466                          + sa.func.coalesce(t.c.derived_name,
467                                             sa.cast('', type_=conn.t.types.Composite))
468                         ).label('name'),
469                         sub.c.centroid, sub.c.bbox)\
470                 .join(sub, t.c.country_code == sub.c.country_code)
471
472         if details.geometry_output:
473             sql = _add_geometry_columns(sql, sub.c.centroid, details)
474
475         results = nres.SearchResults()
476         for row in await conn.execute(sql, _details_to_bind_params(details)):
477             result = nres.create_from_country_row(row, nres.SearchResult)
478             assert result
479             result.bbox = Bbox.from_wkb(row.bbox)
480             result.accuracy = self.penalty + self.countries.get_penalty(row.country_code, 5.0)
481             results.append(result)
482
483         return results
484
485
486
487 class PostcodeSearch(AbstractSearch):
488     """ Search for a postcode.
489     """
490     def __init__(self, extra_penalty: float, sdata: SearchData) -> None:
491         super().__init__(sdata.penalty + extra_penalty)
492         self.countries = sdata.countries
493         self.postcodes = sdata.postcodes
494         self.lookups = sdata.lookups
495         self.rankings = sdata.rankings
496
497
498     async def lookup(self, conn: SearchConnection,
499                      details: SearchDetails) -> nres.SearchResults:
500         """ Find results for the search in the database.
501         """
502         t = conn.t.postcode
503         pcs = self.postcodes.values
504
505         sql = sa.select(t.c.place_id, t.c.parent_place_id,
506                         t.c.rank_search, t.c.rank_address,
507                         t.c.postcode, t.c.country_code,
508                         t.c.geometry.label('centroid'))\
509                 .where(t.c.postcode.in_(pcs))
510
511         if details.geometry_output:
512             sql = _add_geometry_columns(sql, t.c.geometry, details)
513
514         penalty: SaExpression = sa.literal(self.penalty)
515
516         if details.viewbox is not None:
517             if details.bounded_viewbox:
518                 sql = sql.where(t.c.geometry.intersects(VIEWBOX_PARAM))
519             else:
520                 penalty += sa.case((t.c.geometry.intersects(VIEWBOX_PARAM), 0.0),
521                                    (t.c.geometry.intersects(VIEWBOX2_PARAM), 1.0),
522                                    else_=2.0)
523
524         if details.near is not None:
525             if details.near_radius is not None:
526                 sql = sql.where(_within_near(t))
527             sql = sql.order_by(t.c.geometry.ST_Distance(NEAR_PARAM))
528
529         if self.countries:
530             sql = sql.where(t.c.country_code.in_(self.countries.values))
531
532         if details.excluded:
533             sql = sql.where(_exclude_places(t))
534
535         if self.lookups:
536             assert len(self.lookups) == 1
537             assert self.lookups[0].lookup_type == 'restrict'
538             tsearch = conn.t.search_name
539             sql = sql.where(tsearch.c.place_id == t.c.parent_place_id)\
540                      .where(sa.func.array_cat(tsearch.c.name_vector,
541                                               tsearch.c.nameaddress_vector,
542                                               type_=ARRAY(sa.Integer))
543                                     .contains(self.lookups[0].tokens))
544
545         for ranking in self.rankings:
546             penalty += ranking.sql_penalty(conn.t.search_name)
547         penalty += sa.case(*((t.c.postcode == v, p) for v, p in self.postcodes),
548                        else_=1.0)
549
550
551         sql = sql.add_columns(penalty.label('accuracy'))
552         sql = sql.order_by('accuracy').limit(LIMIT_PARAM)
553
554         results = nres.SearchResults()
555         for row in await conn.execute(sql, _details_to_bind_params(details)):
556             result = nres.create_from_postcode_row(row, nres.SearchResult)
557             assert result
558             result.accuracy = row.accuracy
559             results.append(result)
560
561         return results
562
563
564
565 class PlaceSearch(AbstractSearch):
566     """ Generic search for an address or named place.
567     """
568     def __init__(self, extra_penalty: float, sdata: SearchData, expected_count: int) -> None:
569         super().__init__(sdata.penalty + extra_penalty)
570         self.countries = sdata.countries
571         self.postcodes = sdata.postcodes
572         self.housenumbers = sdata.housenumbers
573         self.qualifiers = sdata.qualifiers
574         self.lookups = sdata.lookups
575         self.rankings = sdata.rankings
576         self.expected_count = expected_count
577
578
579     async def lookup(self, conn: SearchConnection,
580                      details: SearchDetails) -> nres.SearchResults:
581         """ Find results for the search in the database.
582         """
583         t = conn.t.placex
584         tsearch = conn.t.search_name
585
586         sql: SaLambdaSelect = sa.lambda_stmt(lambda:
587                   sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
588                             t.c.class_, t.c.type,
589                             t.c.address, t.c.extratags, t.c.admin_level,
590                             t.c.housenumber, t.c.postcode, t.c.country_code,
591                             t.c.wikipedia,
592                             t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
593                             t.c.centroid,
594                             t.c.geometry.ST_Expand(0).label('bbox'))
595                    .where(t.c.place_id == tsearch.c.place_id))
596
597
598         if details.geometry_output:
599             sql = _add_geometry_columns(sql, t.c.geometry, details)
600
601         penalty: SaExpression = sa.literal(self.penalty)
602         for ranking in self.rankings:
603             penalty += ranking.sql_penalty(tsearch)
604
605         for lookup in self.lookups:
606             sql = sql.where(lookup.sql_condition(tsearch))
607
608         if self.countries:
609             sql = sql.where(tsearch.c.country_code.in_(self.countries.values))
610
611         if self.postcodes:
612             # if a postcode is given, don't search for state or country level objects
613             sql = sql.where(tsearch.c.address_rank > 9)
614             tpc = conn.t.postcode
615             pcs = self.postcodes.values
616             if self.expected_count > 1000:
617                 # Many results expected. Restrict by postcode.
618                 sql = sql.where(sa.select(tpc.c.postcode)
619                                   .where(tpc.c.postcode.in_(pcs))
620                                   .where(tsearch.c.centroid.ST_DWithin(tpc.c.geometry, 0.12))
621                                   .exists())
622
623             # Less results, only have a preference for close postcodes
624             pc_near = sa.select(sa.func.min(tpc.c.geometry.ST_Distance(tsearch.c.centroid)))\
625                       .where(tpc.c.postcode.in_(pcs))\
626                       .scalar_subquery()
627             penalty += sa.case((t.c.postcode.in_(pcs), 0.0),
628                                else_=sa.func.coalesce(pc_near, 2.0))
629
630         if details.viewbox is not None:
631             if details.bounded_viewbox:
632                 if details.viewbox.area < 0.2:
633                     sql = sql.where(tsearch.c.centroid.intersects(VIEWBOX_PARAM))
634                 else:
635                     sql = sql.where(tsearch.c.centroid.ST_Intersects_no_index(VIEWBOX_PARAM))
636             elif self.expected_count >= 10000:
637                 if details.viewbox.area < 0.5:
638                     sql = sql.where(tsearch.c.centroid.intersects(VIEWBOX2_PARAM))
639                 else:
640                     sql = sql.where(tsearch.c.centroid.ST_Intersects_no_index(VIEWBOX2_PARAM))
641             else:
642                 penalty += sa.case((t.c.geometry.intersects(VIEWBOX_PARAM), 0.0),
643                                    (t.c.geometry.intersects(VIEWBOX2_PARAM), 1.0),
644                                    else_=2.0)
645
646         if details.near is not None:
647             if details.near_radius is not None:
648                 if details.near_radius < 0.1:
649                     sql = sql.where(tsearch.c.centroid.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM))
650                 else:
651                     sql = sql.where(tsearch.c.centroid.ST_DWithin_no_index(NEAR_PARAM,
652                                                                            NEAR_RADIUS_PARAM))
653             sql = sql.add_columns((-tsearch.c.centroid.ST_Distance(NEAR_PARAM))
654                                       .label('importance'))
655             sql = sql.order_by(sa.desc(sa.text('importance')))
656         else:
657             if self.expected_count < 10000\
658                or (details.viewbox is not None and details.viewbox.area < 0.5):
659                 sql = sql.order_by(
660                         penalty - sa.case((tsearch.c.importance > 0, tsearch.c.importance),
661                                     else_=0.75001-(sa.cast(tsearch.c.search_rank, sa.Float())/40)))
662             sql = sql.add_columns(t.c.importance)
663
664
665         sql = sql.add_columns(penalty.label('accuracy'))
666
667         if self.expected_count < 10000:
668             sql = sql.order_by(sa.text('accuracy'))
669
670         if self.housenumbers:
671             hnr_regexp = f"\\m({'|'.join(self.housenumbers.values)})\\M"
672             sql = sql.where(tsearch.c.address_rank.between(16, 30))\
673                      .where(sa.or_(tsearch.c.address_rank < 30,
674                                    t.c.housenumber.op('~*')(hnr_regexp)))
675
676             # Cross check for housenumbers, need to do that on a rather large
677             # set. Worst case there are 40.000 main streets in OSM.
678             inner = sql.limit(10000).subquery()
679
680             # Housenumbers from placex
681             thnr = conn.t.placex.alias('hnr')
682             pid_list = array_agg(thnr.c.place_id) # type: ignore[no-untyped-call]
683             place_sql = sa.select(pid_list)\
684                           .where(thnr.c.parent_place_id == inner.c.place_id)\
685                           .where(thnr.c.housenumber.op('~*')(hnr_regexp))\
686                           .where(thnr.c.linked_place_id == None)\
687                           .where(thnr.c.indexed_status == 0)
688
689             if details.excluded:
690                 place_sql = place_sql.where(thnr.c.place_id.not_in(sa.bindparam('excluded')))
691             if self.qualifiers:
692                 place_sql = place_sql.where(self.qualifiers.sql_restrict(thnr))
693
694             numerals = [int(n) for n in self.housenumbers.values
695                         if n.isdigit() and len(n) < 8]
696             interpol_sql: SaColumn
697             tiger_sql: SaColumn
698             if numerals and \
699                (not self.qualifiers or ('place', 'house') in self.qualifiers.values):
700                 # Housenumbers from interpolations
701                 interpol_sql = _make_interpolation_subquery(conn.t.osmline, inner,
702                                                             numerals, details)
703                 # Housenumbers from Tiger
704                 tiger_sql = sa.case((inner.c.country_code == 'us',
705                                      _make_interpolation_subquery(conn.t.tiger, inner,
706                                                                   numerals, details)
707                                     ), else_=None)
708             else:
709                 interpol_sql = sa.null()
710                 tiger_sql = sa.null()
711
712             unsort = sa.select(inner, place_sql.scalar_subquery().label('placex_hnr'),
713                                interpol_sql.label('interpol_hnr'),
714                                tiger_sql.label('tiger_hnr')).subquery('unsort')
715             sql = sa.select(unsort)\
716                     .order_by(sa.case((unsort.c.placex_hnr != None, 1),
717                                       (unsort.c.interpol_hnr != None, 2),
718                                       (unsort.c.tiger_hnr != None, 3),
719                                       else_=4),
720                               unsort.c.accuracy)
721         else:
722             sql = sql.where(t.c.linked_place_id == None)\
723                      .where(t.c.indexed_status == 0)
724             if self.qualifiers:
725                 sql = sql.where(self.qualifiers.sql_restrict(t))
726             if details.excluded:
727                 sql = sql.where(_exclude_places(tsearch))
728             if details.min_rank > 0:
729                 sql = sql.where(sa.or_(tsearch.c.address_rank >= MIN_RANK_PARAM,
730                                        tsearch.c.search_rank >= MIN_RANK_PARAM))
731             if details.max_rank < 30:
732                 sql = sql.where(sa.or_(tsearch.c.address_rank <= MAX_RANK_PARAM,
733                                        tsearch.c.search_rank <= MAX_RANK_PARAM))
734             if details.layers is not None:
735                 sql = sql.where(_filter_by_layer(t, details.layers))
736
737         sql = sql.limit(LIMIT_PARAM)
738
739         results = nres.SearchResults()
740         for row in await conn.execute(sql, _details_to_bind_params(details)):
741             result = nres.create_from_placex_row(row, nres.SearchResult)
742             assert result
743             result.bbox = Bbox.from_wkb(row.bbox)
744             result.accuracy = row.accuracy
745             if not details.excluded or not result.place_id in details.excluded:
746                 results.append(result)
747
748             if self.housenumbers and row.rank_address < 30:
749                 if row.placex_hnr:
750                     subs = _get_placex_housenumbers(conn, row.placex_hnr, details)
751                 elif row.interpol_hnr:
752                     subs = _get_osmline(conn, row.interpol_hnr, numerals, details)
753                 elif row.tiger_hnr:
754                     subs = _get_tiger(conn, row.tiger_hnr, numerals, row.osm_id, details)
755                 else:
756                     subs = None
757
758                 if subs is not None:
759                     async for sub in subs:
760                         assert sub.housenumber
761                         sub.accuracy = result.accuracy
762                         if not any(nr in self.housenumbers.values
763                                    for nr in sub.housenumber.split(';')):
764                             sub.accuracy += 0.6
765                         results.append(sub)
766
767                 result.accuracy += 1.0 # penalty for missing housenumber
768
769         return results