From dc66252fdc2978ec84435b2f3854e9d45c156efc Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 4 Apr 2026 17:32:57 -0500 Subject: [PATCH] Reduce dead tuples by using conditional upserts and skip-on-conflict - HRRR, IEMRE, terrain: on_conflict: :nothing (immutable data) - Surface obs, soundings, solar: conditional WHERE (only update when values differ) - set_enrichment_status!: skip rows already at target status --- lib/microwaveprop/radio.ex | 2 +- lib/microwaveprop/terrain.ex | 5 +- lib/microwaveprop/weather.ex | 92 +++++++++++++++++++++++------ test/microwaveprop/weather_test.exs | 27 +++++---- 4 files changed, 93 insertions(+), 33 deletions(-) diff --git a/lib/microwaveprop/radio.ex b/lib/microwaveprop/radio.ex index 4445856e..dcceb1fd 100644 --- a/lib/microwaveprop/radio.ex +++ b/lib/microwaveprop/radio.ex @@ -138,7 +138,7 @@ defmodule Microwaveprop.Radio do def set_enrichment_status!(ids, field, status) do Contact - |> where([q], q.id in ^ids) + |> where([q], q.id in ^ids and field(q, ^field) != ^status) |> Repo.update_all(set: [{field, status}]) end diff --git a/lib/microwaveprop/terrain.ex b/lib/microwaveprop/terrain.ex index 558052b5..18af37ac 100644 --- a/lib/microwaveprop/terrain.ex +++ b/lib/microwaveprop/terrain.ex @@ -10,9 +10,8 @@ defmodule Microwaveprop.Terrain do %TerrainProfile{} |> TerrainProfile.changeset(attrs) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :contact_id, :inserted_at]}, - conflict_target: [:contact_id], - returning: true + on_conflict: :nothing, + conflict_target: [:contact_id] ) end diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index ac57ddf6..693c632d 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -45,9 +45,31 @@ defmodule Microwaveprop.Weather do %SurfaceObservation{} |> SurfaceObservation.changeset(attrs) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :station_id, :observed_at, :inserted_at]}, + 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"), + sea_level_pressure_mb: fragment("EXCLUDED.sea_level_pressure_mb"), + 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], - returning: true + returning: true, + stale_error_field: :id ) end @@ -57,9 +79,34 @@ defmodule Microwaveprop.Weather do %Sounding{} |> Sounding.changeset(attrs) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :station_id, :observed_at, :inserted_at]}, + on_conflict: + from(s in Sounding, + update: [ + set: [ + profile: fragment("EXCLUDED.profile"), + level_count: fragment("EXCLUDED.level_count"), + surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"), + surface_temp_c: fragment("EXCLUDED.surface_temp_c"), + surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"), + surface_refractivity: fragment("EXCLUDED.surface_refractivity"), + min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"), + boundary_layer_depth_m: fragment("EXCLUDED.boundary_layer_depth_m"), + precipitable_water_mm: fragment("EXCLUDED.precipitable_water_mm"), + k_index: fragment("EXCLUDED.k_index"), + lifted_index: fragment("EXCLUDED.lifted_index"), + ducting_detected: fragment("EXCLUDED.ducting_detected"), + duct_characteristics: fragment("EXCLUDED.duct_characteristics"), + updated_at: fragment("EXCLUDED.updated_at") + ] + ], + where: + s.level_count != fragment("EXCLUDED.level_count") or + s.surface_temp_c != fragment("EXCLUDED.surface_temp_c") or + s.surface_refractivity != fragment("EXCLUDED.surface_refractivity") + ), conflict_target: [:station_id, :observed_at], - returning: true + returning: true, + stale_error_field: :id ) end @@ -67,9 +114,25 @@ defmodule Microwaveprop.Weather do %SolarIndex{} |> SolarIndex.changeset(attrs) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :date, :inserted_at]}, + on_conflict: + from(s in SolarIndex, + update: [ + set: [ + sfi: fragment("EXCLUDED.sfi"), + sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"), + sunspot_number: fragment("EXCLUDED.sunspot_number"), + ap_index: fragment("EXCLUDED.ap_index"), + kp_values: fragment("EXCLUDED.kp_values"), + updated_at: fragment("EXCLUDED.updated_at") + ] + ], + where: + s.sfi != fragment("EXCLUDED.sfi") or + s.ap_index != fragment("EXCLUDED.ap_index") + ), conflict_target: [:date], - returning: true + returning: true, + stale_error_field: :id ) end @@ -293,18 +356,14 @@ defmodule Microwaveprop.Weather do %HrrrProfile{} |> HrrrProfile.changeset(attrs) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :inserted_at]}, - conflict_target: [:lat, :lon, :valid_time], - returning: true + on_conflict: :nothing, + conflict_target: [:lat, :lon, :valid_time] ) end - def upsert_hrrr_profiles_batch(profiles, opts \\ []) do + def upsert_hrrr_profiles_batch(profiles, _opts \\ []) do now = DateTime.truncate(DateTime.utc_now(), :second) - on_conflict = - if Keyword.get(opts, :skip_existing, false), do: :nothing, else: {:replace_all_except, [:id, :inserted_at]} - profiles |> Enum.chunk_every(500) |> Enum.reduce({0, nil}, fn chunk, {total_count, _} -> @@ -319,7 +378,7 @@ defmodule Microwaveprop.Weather do {count, rows} = Repo.insert_all(HrrrProfile, entries, - on_conflict: on_conflict, + on_conflict: :nothing, conflict_target: [:lat, :lon, :valid_time] ) @@ -430,9 +489,8 @@ defmodule Microwaveprop.Weather do %IemreObservation{} |> IemreObservation.changeset(attrs) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :inserted_at]}, - conflict_target: [:lat, :lon, :date], - returning: true + on_conflict: :nothing, + conflict_target: [:lat, :lon, :date] ) end diff --git a/test/microwaveprop/weather_test.exs b/test/microwaveprop/weather_test.exs index 89b5f14f..9e197a42 100644 --- a/test/microwaveprop/weather_test.exs +++ b/test/microwaveprop/weather_test.exs @@ -2,6 +2,8 @@ defmodule Microwaveprop.WeatherTest do use Microwaveprop.DataCase, async: true alias Microwaveprop.Weather + alias Microwaveprop.Weather.HrrrProfile + alias Microwaveprop.Weather.IemreObservation @station_attrs %{ station_code: "KDFW", @@ -455,12 +457,13 @@ defmodule Microwaveprop.WeatherTest do assert profile.hpbl_m == 1500.0 end - test "updates existing profile on conflict" do + test "preserves existing profile on conflict" do {:ok, first} = Weather.upsert_hrrr_profile(@hrrr_profile_attrs) - {:ok, second} = Weather.upsert_hrrr_profile(%{@hrrr_profile_attrs | hpbl_m: 2000.0}) + Weather.upsert_hrrr_profile(%{@hrrr_profile_attrs | hpbl_m: 2000.0}) - assert first.id == second.id - assert second.hpbl_m == 2000.0 + reloaded = Repo.get!(HrrrProfile, first.id) + assert reloaded.hpbl_m == 1500.0 + assert Repo.aggregate(HrrrProfile, :count) == 1 end end @@ -548,17 +551,17 @@ defmodule Microwaveprop.WeatherTest do assert obs.date == ~D[2026-03-28] end - test "updates existing observation on conflict" do + test "preserves existing observation on conflict" do {:ok, first} = Weather.upsert_iemre_observation(@iemre_attrs) - {:ok, second} = - Weather.upsert_iemre_observation(%{ - @iemre_attrs - | hourly: [%{"hour" => 0, "p01m_mm" => 1.5}] - }) + Weather.upsert_iemre_observation(%{ + @iemre_attrs + | hourly: [%{"hour" => 0, "p01m_mm" => 1.5}] + }) - assert first.id == second.id - assert hd(second.hourly)["p01m_mm"] == 1.5 + reloaded = Repo.get!(IemreObservation, first.id) + assert hd(reloaded.hourly)["p01m_mm"] == 0.0 + assert Repo.aggregate(IemreObservation, :count) == 1 end end