graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reduce dead tuples by using conditional upserts and skip-on-conflict

Browse source 
- 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
This commit is contained in:
Graham McIntire 2026-04-04 17:32:57 -05:00
parent 383c18e50a
commit dc66252fdc
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
4 changed files with 93 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/microwaveprop/radio.ex
View file

@ -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

5
lib/microwaveprop/terrain.ex
View file

@ -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

92
lib/microwaveprop/weather.ex
View file

@ -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

27
test/microwaveprop/weather_test.exs
View file

@ -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