From 5255a6ba636d93c137a4d7f57715f829ac789c8b Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 13:18:10 -0500 Subject: [PATCH] perf(weather): batch ASOS observation upserts into a single insert_all IemClient.fetch_asos returns 24-288 rows per station per call, each previously triggering an individual UPDATE-conflict round-trip via Weather.upsert_surface_observation/2. On the Turing Pi 2 Postgres node that per-row latency dominates ingestion time. Add Weather.upsert_surface_observations/2, a bulk variant that collapses the rows into a single Repo.insert_all with the same update-only-when-changed on_conflict predicate and (station_id, observed_at) conflict target. Switch WeatherFetchWorker to use it. --- lib/microwaveprop/weather.ex | 88 +++++++++++++++++++ .../workers/weather_fetch_worker.ex | 10 +-- test/microwaveprop/weather_test.exs | 69 +++++++++++++++ 3 files changed, 162 insertions(+), 5 deletions(-) diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 189f7770..22480f9d 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -86,6 +86,94 @@ defmodule Microwaveprop.Weather do ) end + @doc """ + Bulk-upsert surface observations for a single station via one + `Repo.insert_all` round-trip. ASOS fetches return 24–288 rows per + station per call — collapsing them into a single statement avoids the + per-row UPDATE-conflict round-trip of `upsert_surface_observation/2`, + which is expensive against the Turing Pi 2 Postgres node. + + Rows missing `observed_at` are dropped (they can't satisfy the + `(station_id, observed_at)` unique index). Returns the + `{count, nil}` tuple from `Repo.insert_all/3`; `count` reflects + affected rows (new + updated-when-changed). Returns `{0, nil}` for + an empty input without touching the DB. + """ + @spec upsert_surface_observations(Station.t(), [map()]) :: {non_neg_integer(), nil} + def upsert_surface_observations(%Station{} = station, rows) when is_list(rows) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + entries = + rows + |> Enum.filter(&row_has_observed_at?/1) + |> Enum.map(&surface_observation_entry(&1, station.id, now)) + + case entries do + [] -> + {0, nil} + + _ -> + Repo.insert_all(SurfaceObservation, entries, + on_conflict: + from(s in SurfaceObservation, + update: [ + set: [ + temp_f: fragment("EXCLUDED.temp_f"), + dewpoint_f: fragment("EXCLUDED.dewpoint_f"), + relative_humidity: fragment("EXCLUDED.relative_humidity"), + wind_speed_kts: fragment("EXCLUDED.wind_speed_kts"), + wind_direction_deg: fragment("EXCLUDED.wind_direction_deg"), + sea_level_pressure_mb: fragment("EXCLUDED.sea_level_pressure_mb"), + altimeter_setting: fragment("EXCLUDED.altimeter_setting"), + sky_condition: fragment("EXCLUDED.sky_condition"), + precip_1h_in: fragment("EXCLUDED.precip_1h_in"), + wx_codes: fragment("EXCLUDED.wx_codes"), + updated_at: fragment("EXCLUDED.updated_at") + ] + ], + where: + s.temp_f != fragment("EXCLUDED.temp_f") or + s.dewpoint_f != fragment("EXCLUDED.dewpoint_f") or + s.relative_humidity != fragment("EXCLUDED.relative_humidity") or + s.wind_speed_kts != fragment("EXCLUDED.wind_speed_kts") or + s.sea_level_pressure_mb != fragment("EXCLUDED.sea_level_pressure_mb") + ), + conflict_target: [:station_id, :observed_at] + ) + end + end + + defp row_has_observed_at?(%{observed_at: %DateTime{}}), do: true + defp row_has_observed_at?(%{"observed_at" => %DateTime{}}), do: true + defp row_has_observed_at?(_), do: false + + defp surface_observation_entry(row, station_id, now) do + %{ + id: Ecto.UUID.generate(), + station_id: station_id, + observed_at: fetch_row(row, :observed_at), + temp_f: fetch_row(row, :temp_f), + dewpoint_f: fetch_row(row, :dewpoint_f), + relative_humidity: fetch_row(row, :relative_humidity), + wind_speed_kts: fetch_row(row, :wind_speed_kts), + wind_direction_deg: fetch_row(row, :wind_direction_deg), + sea_level_pressure_mb: fetch_row(row, :sea_level_pressure_mb), + altimeter_setting: fetch_row(row, :altimeter_setting), + sky_condition: fetch_row(row, :sky_condition), + precip_1h_in: fetch_row(row, :precip_1h_in), + wx_codes: fetch_row(row, :wx_codes), + inserted_at: now, + updated_at: now + } + end + + defp fetch_row(row, key) when is_atom(key) do + case Map.fetch(row, key) do + {:ok, value} -> value + :error -> Map.get(row, Atom.to_string(key)) + end + end + @spec upsert_sounding(Station.t(), map()) :: {:ok, Sounding.t()} | {:error, Ecto.Changeset.t()} def upsert_sounding(%Station{} = station, attrs) do attrs = Map.put(attrs, :station_id, station.id) diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index 33d81326..0282f58f 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -88,11 +88,11 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do defp ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) do count = Enum.count(rows, & &1.observed_at) - Enum.each(rows, fn row -> - if row.observed_at, do: Weather.upsert_surface_observation(station, row) - end) - - if count == 0, do: store_asos_stub(station, start_dt, end_dt) + if count > 0 do + Weather.upsert_surface_observations(station, rows) + else + store_asos_stub(station, start_dt, end_dt) + end Logger.info("WeatherFetch ASOS: #{station_code} ingested #{count} observations") diff --git a/test/microwaveprop/weather_test.exs b/test/microwaveprop/weather_test.exs index 0aa9565c..507077bd 100644 --- a/test/microwaveprop/weather_test.exs +++ b/test/microwaveprop/weather_test.exs @@ -5,6 +5,7 @@ defmodule Microwaveprop.WeatherTest do alias Microwaveprop.Weather.HrrrNativeProfile alias Microwaveprop.Weather.HrrrProfile alias Microwaveprop.Weather.IemreObservation + alias Microwaveprop.Weather.SurfaceObservation @station_attrs %{ station_code: "KDFW", @@ -76,6 +77,74 @@ defmodule Microwaveprop.WeatherTest do end end + describe "upsert_surface_observations/2" do + test "bulk inserts all rows in a single call" do + {:ok, station} = Weather.find_or_create_station(@station_attrs) + + rows = + for i <- 0..99 do + %{ + observed_at: DateTime.add(~U[2026-03-28 00:00:00Z], i * 300, :second), + temp_f: 40.0 + i / 10, + dewpoint_f: 30.0 + i / 10, + sky_condition: "CLR" + } + end + + assert {count, _} = Weather.upsert_surface_observations(station, rows) + assert count == 100 + assert Repo.aggregate(SurfaceObservation, :count) == 100 + end + + test "overlapping observed_at updates instead of duplicating" do + {:ok, station} = Weather.find_or_create_station(@station_attrs) + + rows1 = + for i <- 0..9 do + %{ + observed_at: DateTime.add(~U[2026-03-28 00:00:00Z], i * 300, :second), + temp_f: 40.0, + dewpoint_f: 30.0 + } + end + + {10, _} = Weather.upsert_surface_observations(station, rows1) + + rows2 = + for i <- 0..9 do + %{ + observed_at: DateTime.add(~U[2026-03-28 00:00:00Z], i * 300, :second), + temp_f: 80.0, + dewpoint_f: 60.0 + } + end + + {_, _} = Weather.upsert_surface_observations(station, rows2) + + assert Repo.aggregate(SurfaceObservation, :count) == 10 + obs = Repo.all(SurfaceObservation) + assert Enum.all?(obs, &(&1.temp_f == 80.0)) + end + + test "returns {0, _} for empty list without hitting the DB" do + {:ok, station} = Weather.find_or_create_station(@station_attrs) + assert {0, _} = Weather.upsert_surface_observations(station, []) + end + + test "skips rows without observed_at" do + {:ok, station} = Weather.find_or_create_station(@station_attrs) + + rows = [ + %{observed_at: ~U[2026-03-28 18:00:00Z], temp_f: 70.0}, + %{observed_at: nil, temp_f: 99.0}, + %{temp_f: 99.0} + ] + + assert {1, _} = Weather.upsert_surface_observations(station, rows) + assert Repo.aggregate(SurfaceObservation, :count) == 1 + end + end + describe "upsert_sounding/2" do @sample_profile [ %{"pres" => 1013.0, "hght" => 171, "tmpc" => 25.0, "dwpc" => 15.0, "drct" => 180, "sknt" => 10},