fix(fix): convert parquet history length km

Author: kshitijrajsharma

api/duck.py

@@ -31,6 +31,7 @@
 _pool: queue.Queue | None = None
 _pool_lock = threading.Lock()
 
+_QUERY_CACHE_DIR = os.getenv("OSMSG_QUERY_CACHE_DIR")  # memoize all-time mega aggregates here; unset -> off
 _HISTORY_URL = os.getenv("OSMSG_HISTORY_URL", "hf://datasets/kshitijrajsharma/osmsg-history")
 _ROLLUP = os.getenv("OSMSG_ROLLUP_BASE", f"{_HISTORY_URL}/rollup")
 _HASHTAG_CHANGESET = os.getenv("OSMSG_HASHTAG_CHANGESET", f"{_ROLLUP}/hashtag_changeset/data.parquet")
@@ -83,6 +84,7 @@ def _sources() -> Sources:
         frontier=_frontier(),
         users_rel=f"read_parquet('{_USERS}')",
         pg_attach=_PG_ATTACH,
+        cache_dir=_QUERY_CACHE_DIR,
     )
 
 

api/pg_schema.py

@@ -2,7 +2,7 @@
 # depends on. Created before the schema, only when absent (PG_TAG_TYPE_EXISTS_SQL): a plain CREATE TYPE
 # is a recognized write both under asyncpg and DuckDB's postgres_execute, unlike a DO/plpgsql block.
 PG_TAG_TYPE_EXISTS_SQL = "SELECT 1 FROM pg_type WHERE typname = 'osmsg_tag'"
-PG_TAG_TYPE_SQL = "CREATE TYPE osmsg_tag AS (k text, v text, c bigint, m bigint, len_m double precision)"
+PG_TAG_TYPE_SQL = "CREATE TYPE osmsg_tag AS (k text, v text, c bigint, m bigint, l double precision)"
 
 PG_SCHEMA = """
 CREATE TABLE IF NOT EXISTS users (

docs/length_backfill.md

@@ -0,0 +1,64 @@
+# Way length (`l`) and the history backfill
+
+osmsg records the length of open ways per tag. Each tag in the `changeset_stats.tags` struct carries
+`l`, the way length in metres (`STRUCT(k, v, c, m, l)`). The API `/tags` endpoint returns `length_m`
+per tag and the leaderboard UI shows it as a kilometre badge.
+
+## What gets a length
+
+A way gets a length when it is **open** (its first and last node references differ) and on **create**.
+The length is the haversine distance along the way's nodes, attached to every tag of that way. Closed
+ways (buildings, areas, roundabouts) get no length. A computed length above `MAX_WAY_LENGTH_M`
+(2,000 km) is dropped as a geometry error (for example an early node first uploaded at (0,0) and later
+corrected).
+
+Geometry source: osmium's node index keeps each node's first-seen coordinates, so a way is measured
+with near-creation-time geometry. On a sample of real ways this sits within about 2% of the exact
+creation-time length in aggregate.
+
+## Live path
+
+The worker computes length on every tick: it applies changefiles with node locations on, and measures
+open-way creates whose nodes are present in the same diff. A way whose nodes come from an earlier diff
+has no location in the current diff and is skipped; the history backfill measures those.
+
+## History backfill
+
+The published history has no length until the datasets are regenerated, because the length needs node
+coordinates that the earlier conversion did not read.
+
+`osmsg maintain convert` (and `convert()` in `osmsg/maintain/convert.py`) now streams the history in a
+**single pass** with a file-backed node index, so open-way lengths can be measured. The parallel
+blob-split path is not used: a way's nodes can live in another split part, and osmium
+`add-locations-to-ways` rejects history files, so a global node index in one pass is required.
+
+Run on a large machine:
+
+```
+osmsg maintain convert --osh history-latest.osh.pbf --changesets changesets-latest.osm.bz2 \
+  --start <min> --end <max> --work-dir /data/convert
+```
+
+Then republish the two datasets and the rollup to HuggingFace, and bump the manifest.
+
+### Cost
+
+- Disk: the node index is roughly 100-150 GB for the full planet (one entry per node id). Keep it on
+  the work volume, not the OS disk.
+- Time: the single streaming pass is single-threaded. The earlier parallel (24-part) streaming ran in
+  about 8.4 h; without that parallelism the streaming is substantially longer (plan for the order of a
+  day), plus aggregation and export. Size the run accordingly.
+- Optimisation (not built): pre-build the node index once, then let parallel part-readers use it
+  read-only. This restores parallelism and is the way to cut the single-pass time.
+
+## Deployment order (load-bearing)
+
+The field rename `len_m` -> `l` changes the schema, so the renamed code and the republished data must
+land together. Deploying the renamed reader against data that still has `len_m` fails.
+
+1. Backfill and republish the rollup, changefiles, and changesets with `l`.
+2. Migrate the production Postgres: run `docs/migrate_pg_len_to_l.sql` once (renames the `osmsg_tag`
+   composite attribute). A fresh store self-creates the type with `l`.
+3. Deploy the renamed worker and API images.
+
+Until all three are done, the current `len_m` data and the old images stay in place.

docs/migrate_pg_len_to_l.sql

@@ -0,0 +1,4 @@
+-- Rename the tag composite field len_m -> l on an existing Postgres store.
+-- Run ONCE before deploying the renamed image (a fresh deploy self-creates the type with `l`).
+-- The osmsg_tag composite backs changeset_stats.tags; renaming the attribute keeps existing rows.
+ALTER TYPE osmsg_tag RENAME ATTRIBUTE len_m TO l;

docs/rollups.md

@@ -21,7 +21,6 @@ Under `hf://datasets/kshitijrajsharma/osmsg-history`:
 | `changefiles` | per-changeset counts + native `tags` + `created_at` + centroid | the source of truth for counts |
 | `changesets` | per-changeset uid/username/editor/hashtags + centroid | the source of truth for metadata |
 | `rollup/hashtag_changeset` | one row per (hashtag, changeset), full breakdown + native `tags` | any hashtag query, exact |
-| `rollup/alltime_user` | one row per uid | the all-time leaderboard with no hashtag |
 | `rollup/users` | uid to username | display names, joined at read time |
 
 `manifest.json` records `min_month` and `max_month`. The frontier is the first instant after
@@ -31,7 +30,7 @@ Under `hf://datasets/kshitijrajsharma/osmsg-history`:
 
 One row per `(hashtag, changeset_id)`, sorted by lowercased hashtag so a prefix range prunes row
 groups. Columns: `hashtag`, `changeset_id`, `uid`, `editor`, `created_at`, the eleven count columns
-(`nodes_created` … `poi_modified`), and `tags`. `tags` is a native `LIST<STRUCT(k, v, c, m, len_m)>`
+(`nodes_created` … `poi_modified`), and `tags`. `tags` is a native `LIST<STRUCT(k, v, c, m, l)>` (`l` is way length in metres)
 (pre-exploded tag stats), so the tag breakdown reads a column instead of parsing JSON per query.
 
 Keeping `changeset_id` makes any prefix exact: a query dedups by `changeset_id`, so a changeset that
@@ -99,8 +98,8 @@ and refines from day to hour to minute as it catches up.
 ## Building and publishing
 
 `osmsg maintain month <YYYY-MM>` builds a finished month: it runs osmsg over the month from day diffs,
-exports the `changefiles` and `changesets` partitions, refreshes `hashtag_changeset`, `alltime_user`,
-and `users`, and uploads them. It refuses a month that stopped short of its boundary, so published
+exports the `changefiles` and `changesets` partitions, refreshes `hashtag_changeset` and `users`,
+and uploads them. It refuses a month that stopped short of its boundary, so published
 months are complete. Re-running a month rebuilds and overwrites it.
 
 ## Consistency

osmsg/catalog.py

@@ -93,7 +93,7 @@ def recent_tag_agg(attach, *, prefixes, frontier, start=None, end=None) -> str:
     inner = (
         f"WITH m AS ({m}) "
         f"SELECT (t).k AS k, (t).v AS v, sum((t).c) AS creates, sum((t).m) AS modifies, "
-        f"sum((t).len_m) AS length_m FROM m JOIN changeset_stats s USING (changeset_id), unnest(s.tags) AS t "
+        f"sum((t).l) AS length_m FROM m JOIN changeset_stats s USING (changeset_id), unnest(s.tags) AS t "
         f"GROUP BY (t).k, (t).v"
     )
     return _pg_query(attach, inner)
@@ -170,7 +170,7 @@ def recent_user_tags(attach, uids: list[int], *, prefixes, frontier, start=None,
     uid_list = ", ".join(str(int(u)) for u in uids)
     ranges = " OR ".join(f"(lower(h) >= {_pg_str(lo)} AND lower(h) < {_pg_str(hi)})" for lo, hi in prefixes)
     inner = (
-        f"SELECT s.uid AS uid, (t).k AS k, (t).v AS v, sum((t).c) AS c, sum((t).m) AS m, sum((t).len_m) AS len_m "
+        f"SELECT s.uid AS uid, (t).k AS k, (t).v AS v, sum((t).c) AS c, sum((t).m) AS m, sum((t).l) AS l "
         f"FROM changeset_stats s JOIN changesets c USING (changeset_id), unnest(s.tags) AS t "
         f"WHERE s.uid IN ({uid_list}) AND {_pg_user_window(frontier, start, end)} "
         f"AND EXISTS (SELECT 1 FROM unnest(c.hashtags) AS h WHERE {ranges}) GROUP BY s.uid, (t).k, (t).v"
@@ -223,7 +223,7 @@ def _recent_from_base(stats_rel: str, changesets_rel: str, window_sql: str, pref
         ),
         tag_rows AS (
             SELECT changeset_id, t.k AS k, t.v AS v,
-                   SUM(t.c) AS c, SUM(t.m) AS m, SUM(t.len_m) AS len_m
+                   SUM(t.c) AS c, SUM(t.m) AS m, SUM(t.l) AS l
             FROM (
                 SELECT s.changeset_id, UNNEST(s.tags) AS t
                 FROM {stats_rel} s JOIN matched USING (changeset_id)
@@ -232,7 +232,7 @@ def _recent_from_base(stats_rel: str, changesets_rel: str, window_sql: str, pref
             GROUP BY changeset_id, t.k, t.v
         ),
         tags AS (
-            SELECT changeset_id, list(struct_pack(k := k, v := v, c := c, m := m, len_m := len_m)) AS tags
+            SELECT changeset_id, list(struct_pack(k := k, v := v, c := c, m := m, l := l)) AS tags
             FROM tag_rows GROUP BY changeset_id
         )
         SELECT m.changeset_id, m.uid, m.editor, m.created_at, {payload}, COALESCE(t.tags, []) AS tags

osmsg/db/duckdb_schema.py

@@ -29,7 +29,7 @@
     rels_deleted   INTEGER DEFAULT 0,
     poi_created    INTEGER DEFAULT 0,
     poi_modified   INTEGER DEFAULT 0,
-    tags           STRUCT(k VARCHAR, v VARCHAR, c BIGINT, m BIGINT, len_m DOUBLE)[],
+    tags           STRUCT(k VARCHAR, v VARCHAR, c BIGINT, m BIGINT, l DOUBLE)[],
     PRIMARY KEY (seq_id, changeset_id)
 );
 CREATE INDEX IF NOT EXISTS idx_changeset_stats_uid ON changeset_stats(uid);

osmsg/db/ingest.py

@@ -11,15 +11,15 @@
 import pyarrow.parquet as pq
 
 # Native shard column for the per-changeset tag breakdown, matching the store's
-# STRUCT(k VARCHAR, v VARCHAR, c BIGINT, m BIGINT, len_m DOUBLE)[] so ingest is a direct copy.
+# STRUCT(k VARCHAR, v VARCHAR, c BIGINT, m BIGINT, l DOUBLE)[] so ingest is a direct copy.
 _TAG_PA_TYPE = pa.list_(
     pa.struct(
         [
             pa.field("k", pa.string()),
             pa.field("v", pa.string()),
             pa.field("c", pa.int64()),
             pa.field("m", pa.int64()),
-            pa.field("len_m", pa.float64()),
+            pa.field("l", pa.float64()),
         ]
     )
 )

osmsg/db/queries.py

@@ -15,13 +15,13 @@ def _rows(result) -> list[dict[str, Any]]:
 
 
 def _tags_to_nested(tags: list[dict[str, Any]] | None) -> dict[str, dict[str, dict[str, Any]]]:
-    """The native `tags` list (list of {k, v, c, m, len_m}) as the nested {key: {value: {c, m, len}}}
+    """The native `tags` list (list of {k, v, c, m, l}) as the nested {key: {value: {c, m, len}}}
     shape `_accumulate_tags` sums over. len is omitted when absent."""
     out: dict[str, dict[str, dict[str, Any]]] = {}
     for t in tags or []:
         entry: dict[str, Any] = {"c": t["c"], "m": t["m"]}
-        if t["len_m"] is not None:
-            entry["len"] = t["len_m"]
+        if t["l"] is not None:
+            entry["len"] = t["l"]
         out.setdefault(t["k"], {})[t["v"]] = entry
     return out
 

osmsg/export/psql.py

@@ -122,16 +122,20 @@ def _push_changeset_hashtags(conn: duckdb.DuckDBPyConnection, where: str = "") -
     )
 
 
-def _push_chunked(conn: duckdb.DuckDBPyConnection, source: str, push) -> None:
-    """Call push() once per changeset_id range so each range commits on its own."""
-    bounds = conn.execute(f"SELECT min(changeset_id), max(changeset_id) FROM {source}").fetchone()
+def _push_chunked(conn: duckdb.DuckDBPyConnection, source: str, push, extra: str = "") -> None:
+    """Call push() once per changeset_id range so each range commits on its own, keeping peak memory to
+    one chunk. `extra` is an extra predicate (no WHERE) ANDed into both the bounds probe and every chunk,
+    so an incremental push chunks only the rows it will actually push."""
+    where_extra = f" WHERE {extra}" if extra else ""
+    bounds = conn.execute(f"SELECT min(changeset_id), max(changeset_id) FROM {source}{where_extra}").fetchone()
     if not bounds or bounds[0] is None:
         return
     lo, hi = bounds
     step = (hi - lo) // _BULK_COMMIT_CHUNKS + 1
     cursor = lo
     while cursor <= hi:
-        push(conn, f"WHERE changeset_id >= {cursor} AND changeset_id < {cursor + step}")
+        rng = f"changeset_id >= {cursor} AND changeset_id < {cursor + step}"
+        push(conn, f"WHERE {rng} AND {extra}" if extra else f"WHERE {rng}")
         cursor += step
 
 
@@ -215,14 +219,16 @@ def to_psql(conn: duckdb.DuckDBPyConnection, dsn: str, *, bulk_load: bool = Fals
                 "INSERT INTO pg_target.users SELECT * FROM users "
                 "WHERE uid IN (SELECT uid FROM changeset_stats WHERE seq_id <> 0) ON CONFLICT DO NOTHING"
             )
-            _push_changesets(conn, f"WHERE {live_ids}")
-            _push_changeset_stats(conn, "WHERE seq_id <> 0")
-            _push_changeset_hashtags(conn, f"WHERE {live_ids}")
+            # Chunked like the bulk path: one INSERT of the whole live tail exceeds the worker's memory
+            # once the tail grows to a month; per-range commits keep peak memory to a single chunk.
+            _push_chunked(conn, "changesets", _push_changesets, live_ids)
+            _push_chunked(conn, "changeset_stats", _push_changeset_stats, "seq_id <> 0")
+            _push_chunked(conn, "changesets", _push_changeset_hashtags, live_ids)
         else:
             conn.execute("INSERT INTO pg_target.users SELECT * FROM users ON CONFLICT DO NOTHING")
-            _push_changesets(conn)
-            _push_changeset_stats(conn)
-            _push_changeset_hashtags(conn)
+            _push_chunked(conn, "changesets", _push_changesets)
+            _push_chunked(conn, "changeset_stats", _push_changeset_stats)
+            _push_chunked(conn, "changesets", _push_changeset_hashtags)
 
         conn.execute(
             """