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

fix: batch pskr calibration upsert into 1,000-row chunks to avoid Postgres 65,535 param limit

Browse source
This commit is contained in:
Graham McInitre 2026-07-15 14:18:43 -05:00
parent 38ee285833
commit a2e2776e7c
1 changed files with 33 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
lib/microwaveprop/pskr/calibration_sampler.ex
View file

@ -267,35 +267,44 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
# ── Upsert ────────────────────────────────────────────────────── # ── Upsert ──────────────────────────────────────────────────────
# Each sample has ~20 fields. PostgreSQL's parameter limit is 65,535,
# so a single insert_all with > ~3,200 rows will fail. Batch at 1,000
# to stay well under the limit even at peak hours.
@upsert_batch_size 1_000
defp upsert([]), do: 0 defp upsert([]), do: 0
defp upsert(samples) do defp upsert(samples) do
{count, _} = samples
Repo.insert_all(CalibrationSample, samples, |> Enum.chunk_every(@upsert_batch_size)
on_conflict: |> Enum.reduce(0, fn batch, total ->
from(c in CalibrationSample, {count, _} =
update: [ Repo.insert_all(CalibrationSample, batch,
set: [ on_conflict:
spot_count: fragment("EXCLUDED.spot_count"), from(c in CalibrationSample,
distinct_paths: fragment("EXCLUDED.distinct_paths"), update: [
median_distance_km: fragment("EXCLUDED.median_distance_km"), set: [
modes: fragment("EXCLUDED.modes"), spot_count: fragment("EXCLUDED.spot_count"),
surface_temp_c: fragment("EXCLUDED.surface_temp_c"), distinct_paths: fragment("EXCLUDED.distinct_paths"),
surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"), median_distance_km: fragment("EXCLUDED.median_distance_km"),
pwat_mm: fragment("EXCLUDED.pwat_mm"), modes: fragment("EXCLUDED.modes"),
surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"), surface_temp_c: fragment("EXCLUDED.surface_temp_c"),
min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"), surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"),
hpbl_m: fragment("EXCLUDED.hpbl_m"), pwat_mm: fragment("EXCLUDED.pwat_mm"),
hrrr_ducting_detected: fragment("EXCLUDED.hrrr_ducting_detected"), surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"),
kp_index: fragment("EXCLUDED.kp_index"), min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"),
updated_at: fragment("EXCLUDED.updated_at") hpbl_m: fragment("EXCLUDED.hpbl_m"),
hrrr_ducting_detected: fragment("EXCLUDED.hrrr_ducting_detected"),
kp_index: fragment("EXCLUDED.kp_index"),
updated_at: fragment("EXCLUDED.updated_at")
]
] ]
] ),
), conflict_target: [:hour_utc, :band, :midpoint_lat, :midpoint_lon]
conflict_target: [:hour_utc, :band, :midpoint_lat, :midpoint_lon] )
)
count total + count
end)
end end
defp align_to_hour(%DateTime{} = dt) do defp align_to_hour(%DateTime{} = dt) do