fix: Corrected pmtiles code so shape lines are sorted. (#1394)

Author: jcpitre

functions-python/pmtiles_builder/function_config.json

@@ -3,7 +3,7 @@
   "description": "The PMTiles Builder function creates PMTiles from dataset files",
   "entry_point": "build_pmtiles_handler",
   "timeout": 1680,
-  "memory": "8Gi",
+  "memory": "16Gi",
   "trigger_http": true,
   "include_folders": ["helpers"],
   "include_api_folders": ["database_gen", "database", "common"],
@@ -22,7 +22,7 @@
   ],
   "ingress_settings": "ALL",
   "max_instance_request_concurrency": 1,
-  "max_instance_count": 5,
+  "max_instance_count": 50,
   "min_instance_count": 0,
-  "available_cpu": 2
+  "available_cpu": 4
 }

functions-python/pmtiles_builder/src/main.py

@@ -87,8 +87,8 @@ def __init__(
         self.name: str
 
         os.makedirs(self._root, exist_ok=True)
-
-        self._ttl_seconds = int(os.getenv("WORKDIR_MAX_AGE_SECONDS", "3600"))
+        # 0 means just delete everything without looking at the file dates.
+        self._ttl_seconds = int(os.getenv("WORKDIR_MAX_AGE_SECONDS", "0"))
 
         if self._debug_dir:
             os.makedirs(self._debug_dir, exist_ok=True)
@@ -135,7 +135,7 @@ def _cleanup_old(self):
                 if age > self._ttl_seconds:
                     shutil.rmtree(entry.path, ignore_errors=True)
                     deleted_count += 1
-                    self._logger.debug(
+                    self._logger.warning(
                         "Removed expired workdir: %s age=%.0fs", entry.path, age
                     )
             except OSError as e:

functions-python/pmtiles_builder/src/shapes_index.py

@@ -28,6 +28,7 @@
 
 from shared.helpers.transform import get_safe_value, get_safe_float, get_safe_int
 from shared.helpers.utils import detect_encoding
+from shared.helpers.runtime_metrics import track_metrics
 
 
 class ShapesIndex:
@@ -119,6 +120,7 @@ def build_index(self) -> None:
                 f.seek(0)
                 line_count = 0
                 f.readline()  # Skip header
+                needs_sorting = False
                 for line in f:
                     try:
                         if not line.strip():
@@ -130,7 +132,17 @@ def build_index(self) -> None:
                         lon_arr, lat_arr, seq_arr = self.coordinates_arrays[shape_id]
                         lon_arr[position] = get_safe_float(row[lon_idx])
                         lat_arr[position] = get_safe_float(row[lat_idx])
-                        seq_arr[position] = get_safe_int(row[seq_idx])
+                        seq_value = get_safe_int(row[seq_idx])
+                        seq_arr[position] = seq_value
+
+                        # If any sequence number goes down, set needs_sorting True.
+                        # Just a little low hanging optimization so we don't sort for nothing.
+                        if (
+                            not needs_sorting
+                            and position > 0
+                            and seq_value < seq_arr[position - 1]
+                        ):
+                            needs_sorting = True
 
                         positions_in_coordinates_arrays[shape_id] = position + 1
 
@@ -148,13 +160,27 @@ def build_index(self) -> None:
                         self.logger.warning(
                             f"Skipping line {line_count} of shapes.txt because of error: {e}"
                         )
+                if needs_sorting:
+                    self.sort_coordinate_arrays()
+
             if self.lines_with_quotes > 0:
                 self.logger.debug(
                     f"Found {self.lines_with_quotes} lines with quotes while creating shapes index"
                 )
         except Exception as e:
             self.logger.warning("Cannot read shapes file: %s", e)
 
+    @track_metrics(metrics=("time", "memory", "cpu"))
+    def sort_coordinate_arrays(self):
+        """
+        Sorts the coordinate arrays in-place for each shape_id according to the sequence number.
+        """
+        for key, (lon_arr, lat_arr, seq_arr) in self.coordinates_arrays.items():
+            sort_idx = np.argsort(seq_arr)
+            lon_arr[:] = lon_arr[sort_idx]
+            lat_arr[:] = lat_arr[sort_idx]
+            seq_arr[:] = seq_arr[sort_idx]
+
     def get_objects(self, key: str):
         """Return a list of objects for rows matching key, using row_fn or raw row dicts."""
         return self.get_shape_points(key)

functions-python/pmtiles_builder/tests/test_build_pmtiles.py

@@ -212,8 +212,11 @@ def test_build_pmtiles_creates_correct_shapes_index(self):
             shapes_path = os.path.join(temp_dir, "shapes.txt")
             with open(shapes_path, "w", encoding="utf-8") as f:
                 f.write("shape_id,shape_pt_lat,shape_pt_lon,shape_pt_sequence\n")
-                f.write("s1,45.0,-73.0,1\n")
-                f.write("s1,45.1,-73.1,2\n")
+                # Also test the reordering of shapes lines. In the test file we put sequence number 10 before 8
+                # It also tests the reordering works fine if there are gaps in the sequence numbers
+                f.write("s1,45.1,-73.1,10\n")
+                f.write("s1,45.0,-73.0,8\n")
+                # f.write("s1,45.1,-73.1,2\n")
                 f.write("s2,46.0,-74.0,1\n")
 
             index = self.builder._create_shapes_index()
@@ -225,7 +228,12 @@ def test_build_pmtiles_creates_correct_shapes_index(self):
         self.assertIn("s2", index.coordinates_arrays)
         self.assertEqual(len(index.coordinates_arrays["s1"]), 3)
         self.assertEqual(len(index.coordinates_arrays["s1"][0]), 2)
-
+        self.assertEqual(len(index.coordinates_arrays["s2"][0]), 1)
+        # Assert that sequence numbers for s1 are sorted
+        self.assertListEqual(list(index.coordinates_arrays["s1"][2]), [8, 10])
+        # Assert that lon and lat for s1 are reordered to match sequence
+        self.assertListEqual(list(index.coordinates_arrays["s1"][0]), [-73.0, -73.1])
+        self.assertListEqual(list(index.coordinates_arrays["s1"][1]), [45.0, 45.1])
         self.assertEqual(len(index.coordinates_arrays["s2"][0]), 1)
 
     def test_create_routes_geojson(self):

infra/batch/main.tf

@@ -257,16 +257,16 @@ resource "google_cloud_tasks_queue" "pmtiles_builder_task_queue" {
   name     = "pmtiles-builder-queue-${var.environment}-${local.deployment_timestamp}"
 
   rate_limits {
-    max_concurrent_dispatches = 1
+    max_concurrent_dispatches = 20
     max_dispatches_per_second = 1
   }
 
   retry_config {
-    # This will make the cloud task retry for ~1 hour
-    max_attempts  = 31
+    # Retries span ~4 minutes: initial try + 2 retries, each 120s apart; total wait ≈ 240s plus processing time
+    max_attempts  = 3
     min_backoff   = "120s"
     max_backoff   = "120s"
-    max_doublings = 2
+    max_doublings = 2  # Moot here since min_backoff == max_backoff
   }
 }
 

infra/functions-python/main.tf

@@ -1153,16 +1153,16 @@ resource "google_cloud_tasks_queue" "pmtiles_builder_task_queue" {
   name     = "pmtiles-builder-queue-${var.environment}-${local.deployment_timestamp}"
 
   rate_limits {
-    max_concurrent_dispatches = 1
+    max_concurrent_dispatches = 20
     max_dispatches_per_second = 1
   }
 
   retry_config {
-    # This will make the cloud task retry for ~1 hour
-    max_attempts  = 31
+    # Retries span ~4 minutes: initial try + 2 retries, each 120s apart; total wait ≈ 240s plus processing time
+    max_attempts  = 3
     min_backoff   = "120s"
     max_backoff   = "120s"
-    max_doublings = 2
+    max_doublings = 2  # Moot here since min_backoff == max_backoff
   }
 }