1 # SPDX-License-Identifier: GPL-3.0-or-later
3 # This file is part of Nominatim. (https://nominatim.org)
5 # Copyright (C) 2023 by the Nominatim developer community.
6 # For a full list of authors see the git log.
8 Conversion from token assignment to an abstract DB search.
10 from typing import Optional, List, Tuple, Iterator, Dict
13 from nominatim.api.types import SearchDetails, DataLayer
14 from nominatim.api.search.query import QueryStruct, Token, TokenType, TokenRange, BreakType
15 from nominatim.api.search.token_assignment import TokenAssignment
16 import nominatim.api.search.db_search_fields as dbf
17 import nominatim.api.search.db_searches as dbs
18 import nominatim.api.search.db_search_lookups as lookups
21 def wrap_near_search(categories: List[Tuple[str, str]],
22 search: dbs.AbstractSearch) -> dbs.NearSearch:
23 """ Create a new search that wraps the given search in a search
24 for near places of the given category.
26 return dbs.NearSearch(penalty=search.penalty,
27 categories=dbf.WeightedCategories(categories,
28 [0.0] * len(categories)),
32 def build_poi_search(category: List[Tuple[str, str]],
33 countries: Optional[List[str]]) -> dbs.PoiSearch:
34 """ Create a new search for places by the given category, possibly
35 constraint to the given countries.
38 ccs = dbf.WeightedStrings(countries, [0.0] * len(countries))
40 ccs = dbf.WeightedStrings([], [])
42 class _PoiData(dbf.SearchData):
44 qualifiers = dbf.WeightedCategories(category, [0.0] * len(category))
47 return dbs.PoiSearch(_PoiData())
51 """ Build the abstract search queries from token assignments.
54 def __init__(self, query: QueryStruct, details: SearchDetails) -> None:
56 self.details = details
60 def configured_for_country(self) -> bool:
61 """ Return true if the search details are configured to
62 allow countries in the result.
64 return self.details.min_rank <= 4 and self.details.max_rank >= 4 \
65 and self.details.layer_enabled(DataLayer.ADDRESS)
69 def configured_for_postcode(self) -> bool:
70 """ Return true if the search details are configured to
71 allow postcodes in the result.
73 return self.details.min_rank <= 5 and self.details.max_rank >= 11\
74 and self.details.layer_enabled(DataLayer.ADDRESS)
78 def configured_for_housenumbers(self) -> bool:
79 """ Return true if the search details are configured to
80 allow addresses in the result.
82 return self.details.max_rank >= 30 \
83 and self.details.layer_enabled(DataLayer.ADDRESS)
86 def build(self, assignment: TokenAssignment) -> Iterator[dbs.AbstractSearch]:
87 """ Yield all possible abstract searches for the given token assignment.
89 sdata = self.get_search_data(assignment)
93 near_items = self.get_near_items(assignment)
94 if near_items is not None and not near_items:
95 return # impossible compbination of near items and category parameter
97 if assignment.name is None:
98 if near_items and not sdata.postcodes:
99 sdata.qualifiers = near_items
101 builder = self.build_poi_search(sdata)
102 elif assignment.housenumber:
103 hnr_tokens = self.query.get_tokens(assignment.housenumber,
104 TokenType.HOUSENUMBER)
105 builder = self.build_housenumber_search(sdata, hnr_tokens, assignment.address)
107 builder = self.build_special_search(sdata, assignment.address,
110 builder = self.build_name_search(sdata, assignment.name, assignment.address,
114 penalty = min(near_items.penalties)
115 near_items.penalties = [p - penalty for p in near_items.penalties]
116 for search in builder:
117 search_penalty = search.penalty
119 yield dbs.NearSearch(penalty + assignment.penalty + search_penalty,
122 for search in builder:
123 search.penalty += assignment.penalty
127 def build_poi_search(self, sdata: dbf.SearchData) -> Iterator[dbs.AbstractSearch]:
128 """ Build abstract search query for a simple category search.
129 This kind of search requires an additional geographic constraint.
131 if not sdata.housenumbers \
132 and ((self.details.viewbox and self.details.bounded_viewbox) or self.details.near):
133 yield dbs.PoiSearch(sdata)
136 def build_special_search(self, sdata: dbf.SearchData,
137 address: List[TokenRange],
138 is_category: bool) -> Iterator[dbs.AbstractSearch]:
139 """ Build abstract search queries for searches that do not involve
143 # No special searches over qualifiers supported.
146 if sdata.countries and not address and not sdata.postcodes \
147 and self.configured_for_country:
148 yield dbs.CountrySearch(sdata)
150 if sdata.postcodes and (is_category or self.configured_for_postcode):
151 penalty = 0.0 if sdata.countries else 0.1
153 sdata.lookups = [dbf.FieldLookup('nameaddress_vector',
154 [t.token for r in address
155 for t in self.query.get_partials_list(r)],
158 yield dbs.PostcodeSearch(penalty, sdata)
161 def build_housenumber_search(self, sdata: dbf.SearchData, hnrs: List[Token],
162 address: List[TokenRange]) -> Iterator[dbs.AbstractSearch]:
163 """ Build a simple address search for special entries where the
164 housenumber is the main name token.
166 sdata.lookups = [dbf.FieldLookup('name_vector', [t.token for t in hnrs], lookups.LookupAny)]
167 expected_count = sum(t.count for t in hnrs)
169 partials = {t.token: t.count for trange in address
170 for t in self.query.get_partials_list(trange)}
172 if expected_count < 8000:
173 sdata.lookups.append(dbf.FieldLookup('nameaddress_vector',
174 list(partials), lookups.Restrict))
175 elif len(partials) != 1 or list(partials.values())[0] < 10000:
176 sdata.lookups.append(dbf.FieldLookup('nameaddress_vector',
177 list(partials), lookups.LookupAll))
179 addr_fulls = [t.token for t
180 in self.query.get_tokens(address[0], TokenType.WORD)]
181 if len(addr_fulls) > 5:
183 sdata.lookups.append(
184 dbf.FieldLookup('nameaddress_vector', addr_fulls, lookups.LookupAny))
186 sdata.housenumbers = dbf.WeightedStrings([], [])
187 yield dbs.PlaceSearch(0.05, sdata, expected_count)
190 def build_name_search(self, sdata: dbf.SearchData,
191 name: TokenRange, address: List[TokenRange],
192 is_category: bool) -> Iterator[dbs.AbstractSearch]:
193 """ Build abstract search queries for simple name or address searches.
195 if is_category or not sdata.housenumbers or self.configured_for_housenumbers:
196 ranking = self.get_name_ranking(name)
197 name_penalty = ranking.normalize_penalty()
199 sdata.rankings.append(ranking)
200 for penalty, count, lookup in self.yield_lookups(name, address):
201 sdata.lookups = lookup
202 yield dbs.PlaceSearch(penalty + name_penalty, sdata, count)
205 def yield_lookups(self, name: TokenRange, address: List[TokenRange])\
206 -> Iterator[Tuple[float, int, List[dbf.FieldLookup]]]:
207 """ Yield all variants how the given name and address should best
208 be searched for. This takes into account how frequent the terms
209 are and tries to find a lookup that optimizes index use.
211 penalty = 0.0 # extra penalty
212 name_partials = {t.token: t for t in self.query.get_partials_list(name)}
214 addr_partials = [t for r in address for t in self.query.get_partials_list(r)]
215 addr_tokens = list({t.token for t in addr_partials})
217 partials_indexed = all(t.is_indexed for t in name_partials.values()) \
218 and all(t.is_indexed for t in addr_partials)
219 exp_count = min(t.count for t in name_partials.values()) / (2**(len(name_partials) - 1))
221 if (len(name_partials) > 3 or exp_count < 8000) and partials_indexed:
222 yield penalty, exp_count, dbf.lookup_by_names(list(name_partials.keys()), addr_tokens)
225 # Partial term to frequent. Try looking up by rare full names first.
226 name_fulls = self.query.get_tokens(name, TokenType.WORD)
228 fulls_count = sum(t.count for t in name_fulls)
229 if len(name_partials) == 1:
230 penalty += min(1, max(0, (exp_count - 50 * fulls_count) / (1000 * fulls_count)))
231 # At this point drop unindexed partials from the address.
232 # This might yield wrong results, nothing we can do about that.
233 if not partials_indexed:
234 addr_tokens = [t.token for t in addr_partials if t.is_indexed]
235 penalty += 1.2 * sum(t.penalty for t in addr_partials if not t.is_indexed)
236 # Any of the full names applies with all of the partials from the address
237 yield penalty, fulls_count / (2**len(addr_tokens)),\
238 dbf.lookup_by_any_name([t.token for t in name_fulls],
240 fulls_count > 30000 / max(1, len(addr_tokens)))
242 # To catch remaining results, lookup by name and address
243 # We only do this if there is a reasonable number of results expected.
244 exp_count = exp_count / (2**len(addr_tokens)) if addr_tokens else exp_count
245 if exp_count < 10000 and all(t.is_indexed for t in name_partials.values()):
246 lookup = [dbf.FieldLookup('name_vector', list(name_partials.keys()), lookups.LookupAll)]
248 lookup.append(dbf.FieldLookup('nameaddress_vector', addr_tokens, lookups.LookupAll))
249 penalty += 0.35 * max(1 if name_fulls else 0.1,
250 5 - len(name_partials) - len(addr_tokens))
251 yield penalty, exp_count, lookup
254 def get_name_ranking(self, trange: TokenRange,
255 db_field: str = 'name_vector') -> dbf.FieldRanking:
256 """ Create a ranking expression for a name term in the given range.
258 name_fulls = self.query.get_tokens(trange, TokenType.WORD)
259 ranks = [dbf.RankedTokens(t.penalty, [t.token]) for t in name_fulls]
260 ranks.sort(key=lambda r: r.penalty)
261 # Fallback, sum of penalty for partials
262 name_partials = self.query.get_partials_list(trange)
263 default = sum(t.penalty for t in name_partials) + 0.2
264 return dbf.FieldRanking(db_field, default, ranks)
267 def get_addr_ranking(self, trange: TokenRange) -> dbf.FieldRanking:
268 """ Create a list of ranking expressions for an address term
269 for the given ranges.
271 todo: List[Tuple[int, int, dbf.RankedTokens]] = []
272 heapq.heappush(todo, (0, trange.start, dbf.RankedTokens(0.0, [])))
273 ranks: List[dbf.RankedTokens] = []
275 while todo: # pylint: disable=too-many-nested-blocks
276 neglen, pos, rank = heapq.heappop(todo)
277 for tlist in self.query.nodes[pos].starting:
278 if tlist.ttype in (TokenType.PARTIAL, TokenType.WORD):
279 if tlist.end < trange.end:
280 chgpenalty = PENALTY_WORDCHANGE[self.query.nodes[tlist.end].btype]
281 if tlist.ttype == TokenType.PARTIAL:
282 penalty = rank.penalty + chgpenalty \
283 + max(t.penalty for t in tlist.tokens)
284 heapq.heappush(todo, (neglen - 1, tlist.end,
285 dbf.RankedTokens(penalty, rank.tokens)))
287 for t in tlist.tokens:
288 heapq.heappush(todo, (neglen - 1, tlist.end,
289 rank.with_token(t, chgpenalty)))
290 elif tlist.end == trange.end:
291 if tlist.ttype == TokenType.PARTIAL:
292 ranks.append(dbf.RankedTokens(rank.penalty
293 + max(t.penalty for t in tlist.tokens),
296 ranks.extend(rank.with_token(t, 0.0) for t in tlist.tokens)
298 # Too many variants, bail out and only add
299 # Worst-case Fallback: sum of penalty of partials
300 name_partials = self.query.get_partials_list(trange)
301 default = sum(t.penalty for t in name_partials) + 0.2
302 ranks.append(dbf.RankedTokens(rank.penalty + default, []))
303 # Bail out of outer loop
307 ranks.sort(key=lambda r: len(r.tokens))
308 default = ranks[0].penalty + 0.3
310 ranks.sort(key=lambda r: r.penalty)
312 return dbf.FieldRanking('nameaddress_vector', default, ranks)
315 def get_search_data(self, assignment: TokenAssignment) -> Optional[dbf.SearchData]:
316 """ Collect the tokens for the non-name search fields in the
319 sdata = dbf.SearchData()
320 sdata.penalty = assignment.penalty
321 if assignment.country:
322 tokens = self.get_country_tokens(assignment.country)
325 sdata.set_strings('countries', tokens)
326 elif self.details.countries:
327 sdata.countries = dbf.WeightedStrings(self.details.countries,
328 [0.0] * len(self.details.countries))
329 if assignment.housenumber:
330 sdata.set_strings('housenumbers',
331 self.query.get_tokens(assignment.housenumber,
332 TokenType.HOUSENUMBER))
333 if assignment.postcode:
334 sdata.set_strings('postcodes',
335 self.query.get_tokens(assignment.postcode,
337 if assignment.qualifier:
338 tokens = self.get_qualifier_tokens(assignment.qualifier)
341 sdata.set_qualifiers(tokens)
342 elif self.details.categories:
343 sdata.qualifiers = dbf.WeightedCategories(self.details.categories,
344 [0.0] * len(self.details.categories))
346 if assignment.address:
347 if not assignment.name and assignment.housenumber:
348 # housenumber search: the first item needs to be handled like
349 # a name in ranking or penalties are not comparable with
351 sdata.set_ranking([self.get_name_ranking(assignment.address[0],
352 db_field='nameaddress_vector')]
353 + [self.get_addr_ranking(r) for r in assignment.address[1:]])
355 sdata.set_ranking([self.get_addr_ranking(r) for r in assignment.address])
362 def get_country_tokens(self, trange: TokenRange) -> List[Token]:
363 """ Return the list of country tokens for the given range,
364 optionally filtered by the country list from the details
367 tokens = self.query.get_tokens(trange, TokenType.COUNTRY)
368 if self.details.countries:
369 tokens = [t for t in tokens if t.lookup_word in self.details.countries]
374 def get_qualifier_tokens(self, trange: TokenRange) -> List[Token]:
375 """ Return the list of qualifier tokens for the given range,
376 optionally filtered by the qualifier list from the details
379 tokens = self.query.get_tokens(trange, TokenType.QUALIFIER)
380 if self.details.categories:
381 tokens = [t for t in tokens if t.get_category() in self.details.categories]
386 def get_near_items(self, assignment: TokenAssignment) -> Optional[dbf.WeightedCategories]:
387 """ Collect tokens for near items search or use the categories
388 requested per parameter.
389 Returns None if no category search is requested.
391 if assignment.near_item:
392 tokens: Dict[Tuple[str, str], float] = {}
393 for t in self.query.get_tokens(assignment.near_item, TokenType.NEAR_ITEM):
394 cat = t.get_category()
395 # The category of a near search will be that of near_item.
396 # Thus, if search is restricted to a category parameter,
397 # the two sets must intersect.
398 if (not self.details.categories or cat in self.details.categories)\
399 and t.penalty < tokens.get(cat, 1000.0):
400 tokens[cat] = t.penalty
401 return dbf.WeightedCategories(list(tokens.keys()), list(tokens.values()))
406 PENALTY_WORDCHANGE = {
407 BreakType.START: 0.0,
409 BreakType.PHRASE: 0.0,