Support dependent layer code via "requires" param (#205)

This will resolve openmaptiles/openmaptiles#796

The code will produce parallel or linear SQL file(s)
taking into account the `requires` layer parameter.

If layer A is required by B and C, it will produce one joined ABC file,
while allowing other layers to execute in parallel.

* A few comments were added to the output.
* Another minor change -- layer object stores its index (position) within the tileset

Author: nyurik

bin/debug-mvt

@@ -116,8 +116,8 @@ async def main(args):
         )
         layer_sql = mvt.generate_layer(layer_def)
         if verbose:
-            print(f"\n======= Querying layer {layer_id} =======\n{query.strip()}\n"
-                  f"== MVT SQL\n{layer_sql}")
+            print(f"\n======= Querying layer {layer_id} (#{layer_def.index}) =======\n"
+                  f"{query.strip()}\n== MVT SQL\n{layer_sql}")
 
         # Re-order columns - move osm_id and geometry fields to the right of the table
         names = {

bin/generate-sql

@@ -46,6 +46,6 @@ if __name__ == '__main__':
 
         (path / 'run_first.sql').write_text(run_first)
         (path / 'run_last.sql').write_text(run_last)
-        for idx, sql in enumerate(parallel_sql):
-            (parallel_dir / f'{idx:03}.sql').write_text(sql)
+        for file, sql in parallel_sql.items():
+            (parallel_dir / f'{file}.sql').write_text(sql)
         print(f'Created {len(parallel_sql)} sql files for parallel execution at {path}')

openmaptiles/sql.py

@@ -1,37 +1,87 @@
 import re
-from typing import Union
+from typing import Union, Dict, Tuple
 
 from sys import stderr
 
 from openmaptiles.tileset import Tileset, Layer
 
 
-def collect_sql(tileset_filename, parallel=False, nodata=False):
-    """If parallel is True, returns a sql value that must be executed first,
-        and a lists of sql values that can be ran in parallel.
+def collect_sql(tileset_filename, parallel=False, nodata=False
+                ) -> Union[str, Tuple[str, Dict[str, str], str]]:
+    """If parallel is True, returns a sql value that must be executed first, last,
+        and a dict of names -> sql code that can be ran in parallel.
         If parallel is False, returns a single sql string.
         nodata=True replaces all "/* DELAY_MATERIALIZED_VIEW_CREATION */"
         with the "WITH NO DATA" SQL."""
-    tileset = Tileset.parse(tileset_filename)
+    tileset = Tileset(tileset_filename)
+
+    run_first = "-- This SQL code should be executed first\n\n" + \
+                get_slice_language_tags(tileset.languages)
+    # at this point we don't have any SQL to run at the end
+    run_last = "-- This SQL code should be executed last\n"
+
+    # resolved is a map of layer ID to some ID in results.
+    # the ID in results could be the same as layer ID, or it could be a tuple of IDs
+    resolved = {}
+    # results is an ID -> SQL content map
+    results = {}
+    unresolved = tileset.layers_by_id.copy()
+    last_count = -1
+    # safety to prevent infinite loop, even though it is also checked in tileset
+    while len(resolved) > last_count:
+        last_count = len(resolved)
+        for lid, layer in list(unresolved.items()):
+            if all((v in resolved for v in layer.requires)):
+                # All requirements have been resolved.
+                resolved[lid] = lid
+                results[lid] = layer_to_sql(layer, nodata)
+                del unresolved[lid]
+
+                if layer.requires:
+                    # If there are more than one requirement, merge them first,
+                    # e.g. if there are layers A, B, and C; and C requires A & B,
+                    # first concatenate A and B, and then append C to them.
+                    # Make sure the same code is not merged multiple times
+                    mix = list(layer.requires) + [lid]
+                    lid1 = mix[0]
+                    for idx in range(1, len(mix)):
+                        lid2 = mix[idx]
+                        res_id1 = resolved[lid1]
+                        res_id2 = resolved[lid2]
+                        if res_id1 == res_id2:
+                            continue
+                        merged_id = res_id1 + "__" + res_id2
+                        if merged_id in results:
+                            raise ValueError(f"Naming collision - {merged_id} exists")
+                        # NOTE: merging will move entity to the end of the list
+                        results[merged_id] = results[res_id1] + "\n" + results[res_id2]
+                        del results[res_id1]
+                        del results[res_id2]
+                        # Update resolved IDs to point to the merged result
+                        for k, v in resolved.items():
+                            if v == res_id1 or v == res_id2:
+                                resolved[k] = merged_id
+    if unresolved:
+        raise ValueError(f"Circular dependency found in layer requirements: " +
+                         ', '.join(unresolved.keys()))
+
+    if not parallel:
+        sql = '\n'.join(results.values())
+        return f"{run_first}\n{sql}\n{run_last}"
+    else:
+        return run_first, results, run_last
 
-    run_first = get_slice_language_tags(tileset.languages)
-    run_last = ''  # at this point we don't have any SQL to run at the end
 
-    parallel_sql = []
-    for layer in tileset.layers:
-        schemas = '\n\n'.join((to_sql(v, layer, nodata) for v in layer.schemas))
-        parallel_sql.append(f"""\
+def layer_to_sql(layer: Layer, nodata: bool):
+    schemas = '\n\n'.join((to_sql(v, layer, nodata) for v in layer.schemas))
+    sql = f"""\
 DO $$ BEGIN RAISE NOTICE 'Processing layer {layer.id}'; END$$;
 
 {schemas}
 
 DO $$ BEGIN RAISE NOTICE 'Finished layer {layer.id}'; END$$;
-""")
-
-    if parallel:
-        return run_first, parallel_sql, run_last
-    else:
-        return run_first + '\n'.join(parallel_sql) + run_last
+"""
+    return sql.strip() + "\n"
 
 
 def get_slice_language_tags(languages):
@@ -143,7 +193,7 @@ def sql_value(value):
         return "E'" + value.replace('\\', '\\\\').replace("'", "\\'") + "'"
 
 
-def to_sql(sql, layer, nodata):
+def to_sql(sql: str, layer: Layer, nodata: bool):
     """Clean up SQL, and perform any needed code injections"""
     sql = sql.strip()
 

openmaptiles/tileset.py

@@ -17,7 +17,7 @@ def tag_fields_to_sql(fields):
 
 @dataclass
 class ParsedData:
-    data: dict
+    data: Union[dict, str]
     path: Path
 
 
@@ -67,8 +67,15 @@ def parse(layer_source: Union[str, Path, ParsedData]) -> 'Layer':
 
     def __init__(self,
                  layer_source: Union[str, Path, ParsedData],
-                 tileset: 'Tileset' = None):
+                 tileset: 'Tileset' = None,
+                 index: int = None):
+        """
+        :param layer_source: load layer from this source, e.g. a file
+        :param tileset: parent tileset object (optional)
+        :param index: layer's position index within the tileset
+        """
         self.tileset = tileset
+        self.index = index
 
         if isinstance(layer_source, ParsedData):
             self.filename = layer_source.path
@@ -87,12 +94,29 @@ def __init__(self,
 
         self.imposm_mappings = [parse_file(f) for f in self.imposm_mapping_files]
 
-        self.schemas = [Path(layer_dir, f).read_text('utf-8')
-                        for f in self.definition.get('schema', [])]
+        schemas = [
+            (f.path, f.data) if isinstance(f, ParsedData)
+            else (f, Path(layer_dir, f).read_text('utf-8'))
+            for f in self.definition.get('schema', [])
+        ]
+        self.schemas = [f"-- Layer {self.id} - {p}\n\n{d}" for p, d in schemas]
 
         self.fields = [Field(k, v) for k, v in
                        self.definition['layer']['fields'].items()]
 
+        if 'requires' in self.definition['layer']:
+            self.requires = self.definition['layer']['requires']
+            if isinstance(self.requires, str):
+                self.requires = [self.requires]
+            if (
+                not isinstance(self.requires, list) or
+                any(not isinstance(v, str) or v == "" for v in self.requires)
+            ):
+                raise ValueError("If defined, 'requires' parameter must be the ID "
+                                 "of another layer, or a list of layer IDs")
+        else:
+            self.requires = []
+
         validate_properties(self, f"Layer {self.filename}")
 
         if any(v.name == self.geometry_field for v in self.fields):
@@ -219,8 +243,35 @@ def __init__(self, tileset_source: Union[str, Path, ParsedData]):
 
         self.definition = self.definition['tileset']
         self.layers = []
-        for layer_filename in self.definition['layers']:
-            self.layers.append(Layer(layer_filename, self))
+        self.layers_by_id = {}
+        for index, layer_filename in enumerate(self.definition['layers']):
+            layer = Layer(layer_filename, self, index)
+            if layer.id in self.layers_by_id:
+                raise ValueError(f"Layer '{layer.id}' is defined more than once")
+            self.layers.append(layer)
+            self.layers_by_id[layer.id] = layer
+
+        # Detect circular dependencies and missing layer IDs for the 'requires' field
+        resolved = set()
+        unresolved = self.layers_by_id.copy()
+        last_count = -1
+        while len(resolved) > last_count:
+            last_count = len(resolved)
+            for lid, layer in list(unresolved.items()):
+                for req in layer.requires:
+                    if req not in self.layers_by_id:
+                        raise ValueError(f"Unknown layer '{req}' required for "
+                                         f"layer {layer.id}")
+                    if req not in resolved:
+                        break
+                else:
+                    # all requirements are already in resolved (or no reqs)
+                    resolved.add(lid)
+                    del unresolved[lid]
+        if unresolved:
+            raise ValueError(f"Circular dependency found in layer requirements: " +
+                             ', '.join(unresolved.keys()))
+
         validate_properties(self, f"Tileset {self.filename}")
 
     @deprecated(version='3.2.0', reason='use named properties instead')

tests/expected/parallel_sql/parallel/001.sql renamed to tests/expected/parallel_sql/parallel/enumfield.sql

@@ -1,5 +1,7 @@
 DO $$ BEGIN RAISE NOTICE 'Processing layer enumfield'; END$$;
 
+-- Layer enumfield - ./enumfield.sql
+
 CREATE OR REPLACE FUNCTION map_landuse_class("natural" VARCHAR, landuse VARCHAR) RETURNS TEXT AS $$
     SELECT CASE
         WHEN "natural" = 'bare_rock' THEN 'rock'

tests/expected/parallel_sql/parallel/000.sql renamed to tests/expected/parallel_sql/parallel/housenumber.sql

@@ -1,9 +1,15 @@
 DO $$ BEGIN RAISE NOTICE 'Processing layer housenumber'; END$$;
 
+-- Layer housenumber - ./housenumber_centroid.sql
+
+
 -- etldoc: osm_housenumber_point -> osm_housenumber_point
 UPDATE osm_housenumber_point SET geometry=topoint(geometry)
 WHERE ST_GeometryType(geometry) <> 'ST_Point';
 
+-- Layer housenumber - ./layer.sql
+
+
 -- etldoc: layer_housenumber[shape=record fillcolor=lightpink, style="rounded,filled",
 -- etldoc:     label="layer_housenumber | <z14_> z14_" ] ;
 

tests/expected/parallel_sql/run_first.sql

@@ -1,3 +1,5 @@
+-- This SQL code should be executed first
+
 CREATE OR REPLACE FUNCTION slice_language_tags(tags hstore)
 RETURNS hstore AS $$
     SELECT delete_empty_keys(slice(tags, ARRAY['name:en', 'name:de', 'name:cs', 'int_name', 'loc_name', 'name', 'wikidata', 'wikipedia']))

tests/expected/parallel_sql/run_last.sql

@@ -0,0 +1 @@
+-- This SQL code should be executed last

tests/expected/parallel_sql2/parallel/001.sql renamed to tests/expected/parallel_sql2/parallel/enumfield.sql

@@ -1,5 +1,7 @@
 DO $$ BEGIN RAISE NOTICE 'Processing layer enumfield'; END$$;
 
+-- Layer enumfield - ./enumfield.sql
+
 CREATE OR REPLACE FUNCTION map_landuse_class("natural" VARCHAR, landuse VARCHAR) RETURNS TEXT AS $$
     SELECT CASE
         WHEN "natural" = 'bare_rock' THEN 'rock'

tests/expected/parallel_sql2/parallel/000.sql renamed to tests/expected/parallel_sql2/parallel/housenumber.sql

@@ -1,9 +1,15 @@
 DO $$ BEGIN RAISE NOTICE 'Processing layer housenumber'; END$$;
 
+-- Layer housenumber - ./housenumber_centroid.sql
+
+
 -- etldoc: osm_housenumber_point -> osm_housenumber_point
 UPDATE osm_housenumber_point SET geometry=topoint(geometry)
 WHERE ST_GeometryType(geometry) <> 'ST_Point';
 
+-- Layer housenumber - ./layer.sql
+
+
 -- etldoc: layer_housenumber[shape=record fillcolor=lightpink, style="rounded,filled",
 -- etldoc:     label="layer_housenumber | <z14_> z14_" ] ;