diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 6f9504d..bd9e725 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -277,8 +277,9 @@ defmodule Aprsme.PacketConsumer do insert_opts = [ # Don't return IDs for better performance returning: false, - # Skip conflicts to avoid blocking + # Skip conflicts on the composite PK; any other error surfaces to the rescue below on_conflict: :nothing, + conflict_target: [:id, :received_at], # Increased timeout for large batches timeout: 60_000 ] @@ -314,7 +315,7 @@ defmodule Aprsme.PacketConsumer do try do changeset = Aprsme.Packet.changeset(%Aprsme.Packet{}, packet_attrs) - case Repo.insert(changeset, on_conflict: :nothing) do + case Repo.insert(changeset, on_conflict: :nothing, conflict_target: [:id, :received_at]) do {:ok, _record} -> {count + 1, [packet_attrs | inserted]} @@ -425,6 +426,7 @@ defmodule Aprsme.PacketConsumer do |> truncate_datetimes_to_second() # Create PostGIS geometry for location field BEFORE filtering |> create_location_geometry() + |> set_has_weather() # Remove all non-schema fields - do this LAST to ensure all processing is done |> remove_non_schema_fields() rescue @@ -433,6 +435,16 @@ defmodule Aprsme.PacketConsumer do nil end + # Compute has_weather from any populated weather field (replaces DB trigger) + defp set_has_weather(attrs) do + has_weather? = + Enum.any?(Aprsme.EncodingUtils.weather_fields(), fn field -> + not is_nil(Map.get(attrs, field)) + end) + + Map.put(attrs, :has_weather, has_weather?) + end + # Create PostGIS geometry from lat/lon coordinates defp create_location_geometry(attrs) do lat = attrs[:lat] diff --git a/priv/repo/migrations/20260417134047_drop_redundant_packet_indexes_and_triggers.exs b/priv/repo/migrations/20260417134047_drop_redundant_packet_indexes_and_triggers.exs new file mode 100644 index 0000000..bc9ded2 --- /dev/null +++ b/priv/repo/migrations/20260417134047_drop_redundant_packet_indexes_and_triggers.exs @@ -0,0 +1,102 @@ +defmodule Aprsme.Repo.Migrations.DropRedundantPacketIndexesAndTriggers do + use Ecto.Migration + + @disable_ddl_transaction true + @disable_migration_lock true + + @redundant_indexes [ + # BRIN on partitioned table where each partition is small — B-tree covers this. + "idx_packets_received_brin", + # Standalone lat/lon — GiST on location handles all spatial queries. + "idx_packets_lat", + "idx_packets_lon", + # Redundant with idx_packets_sender_received (same leading column). + "idx_packets_sender_id_desc", + # Geometry GiST — app uses geography casts (ST_DWithin ::geography). + "idx_packets_location", + # text_pattern_ops — callsign queries use upper() equality, not LIKE 'xxx%'. + "idx_packets_sender_pattern", + # Weather filters are always by sender/time; no query filters by raw metric values. + "idx_packets_temperature", + "idx_packets_humidity", + "idx_packets_pressure", + "idx_packets_wind_speed", + "idx_packets_received_temp", + "idx_packets_sender_temp", + # idx_packets_has_weather + idx_packets_sender_weather_lookup cover these. + "idx_packets_weather_history", + "idx_packets_weather_selective", + "idx_packets_weather_symbol" + ] + + def up do + execute("SET LOCAL statement_timeout = '0'") + + for idx <- @redundant_indexes do + execute("DROP INDEX IF EXISTS #{idx}") + end + + # has_weather is now computed in Elixir (packet_consumer.ex prepare_packet_for_insert). + execute("DROP TRIGGER IF EXISTS update_has_weather_trigger ON packets") + end + + def down do + execute("SET LOCAL statement_timeout = '0'") + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_received_brin ON packets USING brin (received_at)" + ) + + execute("CREATE INDEX IF NOT EXISTS idx_packets_lat ON packets (lat)") + execute("CREATE INDEX IF NOT EXISTS idx_packets_lon ON packets (lon)") + execute("CREATE INDEX IF NOT EXISTS idx_packets_sender_id_desc ON packets (sender, id DESC)") + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_location ON packets USING gist (location) WHERE has_position = true" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_sender_pattern ON packets (sender text_pattern_ops)" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_temperature ON packets (temperature) WHERE temperature IS NOT NULL" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_humidity ON packets (humidity) WHERE humidity IS NOT NULL" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_pressure ON packets (pressure) WHERE pressure IS NOT NULL" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_wind_speed ON packets (wind_speed) WHERE wind_speed IS NOT NULL" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_received_temp ON packets (received_at, temperature) WHERE temperature IS NOT NULL" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_sender_temp ON packets (sender, temperature) WHERE temperature IS NOT NULL" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_weather_history ON packets (sender, data_type, received_at) WHERE data_type = 'weather'" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_weather_selective ON packets (received_at DESC) WHERE data_type IN ('weather', 'Weather', 'WX', 'wx')" + ) + + execute( + "CREATE INDEX IF NOT EXISTS idx_packets_weather_symbol ON packets (received_at DESC, lat, lon) WHERE (symbol_table_id = '/' AND symbol_code = '_') OR (symbol_table_id = E'\\\\' AND symbol_code = '_')" + ) + + execute( + "CREATE TRIGGER update_has_weather_trigger BEFORE INSERT OR UPDATE ON packets FOR EACH ROW EXECUTE FUNCTION update_has_weather()" + ) + end +end