Optimize historical packet loading by selecting only needed columns
Historical map loading previously fetched all 73 columns (SELECT *) from the packets table (~2068 bytes/row). Added get_recent_packets_for_map/1 with select_map_fields/1 that fetches only the 22 columns needed for map rendering (~200 bytes/row), yielding a ~3x query speedup. Also removes unused all_packets tracking from PacketProcessor and simplifies visible_packets cleanup in Index.
This commit is contained in:
parent
1832a8ef04
commit
8755bba9f0
6 changed files with 188 additions and 46 deletions
|
|
@ -538,6 +538,45 @@ defmodule Aprsme.Packets do
|
||||||
Repo.all(query)
|
Repo.all(query)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Gets recent packets for map display with only the columns needed for rendering.
|
||||||
|
Returns maps (not full Packet structs) to avoid loading all 73 columns.
|
||||||
|
"""
|
||||||
|
@spec get_recent_packets_for_map(map()) :: [map()]
|
||||||
|
def get_recent_packets_for_map(opts \\ %{}) do
|
||||||
|
hours_back = Map.get(opts, :hours_back, 24)
|
||||||
|
time_ago = DateTime.add(DateTime.utc_now(), -hours_back * 3600, :second)
|
||||||
|
limit = Map.get(opts, :limit, 200)
|
||||||
|
offset = Map.get(opts, :offset, 0)
|
||||||
|
|
||||||
|
callsign = opts[:callsign]
|
||||||
|
normalized_callsign = if is_binary(callsign), do: String.trim(callsign), else: ""
|
||||||
|
|
||||||
|
base_query =
|
||||||
|
if normalized_callsign == "" do
|
||||||
|
from(p in Packet,
|
||||||
|
where: p.has_position == true,
|
||||||
|
where: p.received_at >= ^time_ago
|
||||||
|
)
|
||||||
|
else
|
||||||
|
from(p in Packet,
|
||||||
|
where: p.received_at >= ^time_ago
|
||||||
|
)
|
||||||
|
end
|
||||||
|
|
||||||
|
bounds = Map.get(opts, :bounds)
|
||||||
|
|
||||||
|
base_query
|
||||||
|
|> QueryBuilder.maybe_filter_region(opts)
|
||||||
|
|> maybe_filter_by_callsign(opts)
|
||||||
|
|> maybe_filter_by_bounds(bounds)
|
||||||
|
|> order_by(desc: :received_at)
|
||||||
|
|> limit(^limit)
|
||||||
|
|> offset(^offset)
|
||||||
|
|> QueryBuilder.select_map_fields()
|
||||||
|
|> Repo.all()
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Gets the closest stations to a given point.
|
Gets the closest stations to a given point.
|
||||||
Uses PostGIS spatial indexes for efficient querying.
|
Uses PostGIS spatial indexes for efficient querying.
|
||||||
|
|
|
||||||
|
|
@ -96,6 +96,39 @@ defmodule Aprsme.Packets.QueryBuilder do
|
||||||
select: %{p | lat: fragment("ST_Y(?)", p.location), lon: fragment("ST_X(?)", p.location)}
|
select: %{p | lat: fragment("ST_Y(?)", p.location), lon: fragment("ST_X(?)", p.location)}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Selects only the fields needed for map display as a plain map.
|
||||||
|
Returns ~22 columns instead of all 73, reducing I/O significantly.
|
||||||
|
"""
|
||||||
|
@spec select_map_fields(Ecto.Query.t()) :: Ecto.Query.t()
|
||||||
|
def select_map_fields(query) do
|
||||||
|
from p in query,
|
||||||
|
select: %{
|
||||||
|
id: p.id,
|
||||||
|
sender: p.sender,
|
||||||
|
object_name: p.object_name,
|
||||||
|
item_name: p.item_name,
|
||||||
|
lat: fragment("?::float8", p.lat),
|
||||||
|
lon: fragment("?::float8", p.lon),
|
||||||
|
received_at: p.received_at,
|
||||||
|
symbol_table_id: p.symbol_table_id,
|
||||||
|
symbol_code: p.symbol_code,
|
||||||
|
comment: p.comment,
|
||||||
|
path: p.path,
|
||||||
|
temperature: p.temperature,
|
||||||
|
humidity: p.humidity,
|
||||||
|
wind_direction: p.wind_direction,
|
||||||
|
wind_speed: p.wind_speed,
|
||||||
|
wind_gust: p.wind_gust,
|
||||||
|
pressure: p.pressure,
|
||||||
|
rain_1h: p.rain_1h,
|
||||||
|
rain_24h: p.rain_24h,
|
||||||
|
rain_since_midnight: p.rain_since_midnight,
|
||||||
|
snow: p.snow,
|
||||||
|
luminosity: p.luminosity
|
||||||
|
}
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Filters by region if provided in options.
|
Filters by region if provided in options.
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -231,7 +231,7 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
|
||||||
params = build_query_params(query_params)
|
params = build_query_params(query_params)
|
||||||
Logger.debug("HistoricalLoader: Querying packets with params: #{inspect(params)}")
|
Logger.debug("HistoricalLoader: Querying packets with params: #{inspect(params)}")
|
||||||
|
|
||||||
recent_packets = Packets.get_recent_packets(params)
|
recent_packets = Packets.get_recent_packets_for_map(params)
|
||||||
Logger.debug("HistoricalLoader: Got #{length(recent_packets)} packets")
|
Logger.debug("HistoricalLoader: Got #{length(recent_packets)} packets")
|
||||||
|
|
||||||
maybe_include_latest_packet(recent_packets, query_params.callsign, batch_offset)
|
maybe_include_latest_packet(recent_packets, query_params.callsign, batch_offset)
|
||||||
|
|
|
||||||
|
|
@ -259,7 +259,6 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
defp assign_defaults(socket, one_hour_ago) do
|
defp assign_defaults(socket, one_hour_ago) do
|
||||||
assign(socket,
|
assign(socket,
|
||||||
packets: [],
|
packets: [],
|
||||||
all_packets: %{},
|
|
||||||
visible_packets: %{},
|
visible_packets: %{},
|
||||||
historical_packets: %{},
|
historical_packets: %{},
|
||||||
page_title: "APRS Map",
|
page_title: "APRS Map",
|
||||||
|
|
@ -330,15 +329,10 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
|
|
||||||
@impl true
|
@impl true
|
||||||
def handle_event("clear_and_reload_markers", _params, socket) do
|
def handle_event("clear_and_reload_markers", _params, socket) do
|
||||||
# Get current visible packets from PacketManager
|
|
||||||
current_packets = PacketManager.get_visible_packets(socket.assigns.packet_state)
|
|
||||||
|
|
||||||
# Filter by time and bounds
|
# Filter by time and bounds
|
||||||
filtered_packets =
|
filtered_packets =
|
||||||
filter_packets_by_time_and_bounds_with_tracked(
|
filter_packets_by_time_and_bounds_with_tracked(
|
||||||
Map.new(current_packets, fn packet ->
|
socket.assigns.visible_packets,
|
||||||
{get_callsign_key(packet), packet}
|
|
||||||
end),
|
|
||||||
socket.assigns.map_bounds,
|
socket.assigns.map_bounds,
|
||||||
socket.assigns.packet_age_threshold,
|
socket.assigns.packet_age_threshold,
|
||||||
socket.assigns.tracked_callsign,
|
socket.assigns.tracked_callsign,
|
||||||
|
|
@ -1749,32 +1743,21 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
# Schedule next cleanup
|
# Schedule next cleanup
|
||||||
Process.send_after(self(), :cleanup_old_packets, 60_000)
|
Process.send_after(self(), :cleanup_old_packets, 60_000)
|
||||||
|
|
||||||
# Use the current packet_age_threshold instead of hardcoded one hour
|
|
||||||
threshold = socket.assigns.packet_age_threshold
|
threshold = socket.assigns.packet_age_threshold
|
||||||
|
|
||||||
# Remove expired packets using PacketManager predicate
|
# Partition visible_packets into kept and expired
|
||||||
updated_packet_state =
|
{kept, expired_keys} =
|
||||||
PacketManager.remove_packets_where(socket.assigns.packet_state, fn packet ->
|
Enum.reduce(socket.assigns.visible_packets, {%{}, []}, fn {key, packet}, {kept_acc, expired_acc} ->
|
||||||
SharedPacketUtils.packet_within_time_threshold?(packet, threshold)
|
if SharedPacketUtils.packet_within_time_threshold?(packet, threshold) do
|
||||||
|
{Map.put(kept_acc, key, packet), expired_acc}
|
||||||
|
else
|
||||||
|
{kept_acc, [key | expired_acc]}
|
||||||
|
end
|
||||||
end)
|
end)
|
||||||
|
|
||||||
# Get expired packet IDs that need to be removed from client
|
|
||||||
current_packets = PacketManager.get_visible_packets(socket.assigns.packet_state)
|
|
||||||
remaining_packets = PacketManager.get_visible_packets(updated_packet_state)
|
|
||||||
|
|
||||||
remaining_keys = MapSet.new(remaining_packets, &get_callsign_key/1)
|
|
||||||
|
|
||||||
expired_keys =
|
|
||||||
current_packets
|
|
||||||
|> Enum.reject(fn current_packet ->
|
|
||||||
MapSet.member?(remaining_keys, get_callsign_key(current_packet))
|
|
||||||
end)
|
|
||||||
|> Enum.map(&get_callsign_key/1)
|
|
||||||
|
|
||||||
# Only update the client if there are expired markers
|
|
||||||
socket = DisplayManager.remove_markers_batch(socket, expired_keys)
|
socket = DisplayManager.remove_markers_batch(socket, expired_keys)
|
||||||
|
|
||||||
{:noreply, assign(socket, packet_state: updated_packet_state)}
|
{:noreply, assign(socket, :visible_packets, kept)}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp handle_update_time_display(socket) do
|
defp handle_update_time_display(socket) do
|
||||||
|
|
@ -2008,11 +1991,8 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
)
|
)
|
||||||
|
|
||||||
# Remove out-of-bounds packets and markers immediately
|
# Remove out-of-bounds packets and markers immediately
|
||||||
current_packets = PacketManager.get_visible_packets(socket.assigns.packet_state)
|
new_visible_packets = BoundsUtils.filter_packets_by_bounds(socket.assigns.visible_packets, map_bounds)
|
||||||
current_packets_map = Map.new(current_packets, fn packet -> {get_callsign_key(packet), packet} end)
|
packets_to_remove = BoundsUtils.reject_packets_by_bounds(socket.assigns.visible_packets, map_bounds)
|
||||||
|
|
||||||
new_visible_packets = BoundsUtils.filter_packets_by_bounds(current_packets_map, map_bounds)
|
|
||||||
packets_to_remove = BoundsUtils.reject_packets_by_bounds(current_packets_map, map_bounds)
|
|
||||||
|
|
||||||
# Remove markers for out-of-bounds packets
|
# Remove markers for out-of-bounds packets
|
||||||
socket = remove_markers_batch(socket, packets_to_remove)
|
socket = remove_markers_batch(socket, packets_to_remove)
|
||||||
|
|
|
||||||
|
|
@ -15,7 +15,6 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
alias Phoenix.LiveView.Socket
|
alias Phoenix.LiveView.Socket
|
||||||
|
|
||||||
@max_visible_packets 1000
|
@max_visible_packets 1000
|
||||||
@max_all_packets 2000
|
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Process a packet for display on the map.
|
Process a packet for display on the map.
|
||||||
|
|
@ -25,18 +24,6 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
{lat, lon, _data_extended} = CoordinateUtils.get_coordinates(packet)
|
{lat, lon, _data_extended} = CoordinateUtils.get_coordinates(packet)
|
||||||
callsign_key = SharedPacketUtils.get_callsign_key(packet)
|
callsign_key = SharedPacketUtils.get_callsign_key(packet)
|
||||||
|
|
||||||
# Update all_packets with memory limit
|
|
||||||
all_packets = Map.put(socket.assigns.all_packets, callsign_key, packet)
|
|
||||||
|
|
||||||
all_packets =
|
|
||||||
if map_size(all_packets) > @max_all_packets do
|
|
||||||
SharedPacketUtils.prune_oldest_packets(all_packets, @max_all_packets)
|
|
||||||
else
|
|
||||||
all_packets
|
|
||||||
end
|
|
||||||
|
|
||||||
socket = assign(socket, :all_packets, all_packets)
|
|
||||||
|
|
||||||
# Handle packet visibility logic
|
# Handle packet visibility logic
|
||||||
socket = handle_packet_visibility(packet, lat, lon, callsign_key, socket)
|
socket = handle_packet_visibility(packet, lat, lon, callsign_key, socket)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -305,6 +305,109 @@ defmodule Aprsme.PacketsTest do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
describe "get_recent_packets_for_map/1" do
|
||||||
|
setup do
|
||||||
|
now = DateTime.utc_now()
|
||||||
|
|
||||||
|
PacketsFixtures.packet_fixture(%{
|
||||||
|
sender: "MAP1",
|
||||||
|
received_at: DateTime.add(now, -1800, :second),
|
||||||
|
lat: 39.8,
|
||||||
|
lon: -98.5,
|
||||||
|
comment: "test comment",
|
||||||
|
symbol_table_id: "/",
|
||||||
|
symbol_code: ">",
|
||||||
|
temperature: 72.0,
|
||||||
|
humidity: 45.0
|
||||||
|
})
|
||||||
|
|
||||||
|
PacketsFixtures.packet_fixture(%{
|
||||||
|
sender: "MAP2",
|
||||||
|
received_at: DateTime.add(now, -3600, :second),
|
||||||
|
lat: 39.9,
|
||||||
|
lon: -98.4,
|
||||||
|
object_name: "TestObj"
|
||||||
|
})
|
||||||
|
|
||||||
|
PacketsFixtures.packet_fixture(%{
|
||||||
|
sender: "OLD_MAP",
|
||||||
|
received_at: DateTime.add(now, -48 * 3600, :second),
|
||||||
|
lat: 40.0,
|
||||||
|
lon: -98.3
|
||||||
|
})
|
||||||
|
|
||||||
|
:ok
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns maps with only needed fields" do
|
||||||
|
results = Packets.get_recent_packets_for_map(%{hours_back: 24})
|
||||||
|
|
||||||
|
assert length(results) >= 2
|
||||||
|
first = hd(results)
|
||||||
|
|
||||||
|
# Should be a map, not a Packet struct
|
||||||
|
refute is_struct(first, Packet)
|
||||||
|
assert is_map(first)
|
||||||
|
|
||||||
|
# Should have the essential fields
|
||||||
|
assert Map.has_key?(first, :id)
|
||||||
|
assert Map.has_key?(first, :sender)
|
||||||
|
assert Map.has_key?(first, :lat)
|
||||||
|
assert Map.has_key?(first, :lon)
|
||||||
|
assert Map.has_key?(first, :received_at)
|
||||||
|
assert Map.has_key?(first, :symbol_table_id)
|
||||||
|
assert Map.has_key?(first, :symbol_code)
|
||||||
|
assert Map.has_key?(first, :comment)
|
||||||
|
assert Map.has_key?(first, :path)
|
||||||
|
assert Map.has_key?(first, :object_name)
|
||||||
|
assert Map.has_key?(first, :item_name)
|
||||||
|
|
||||||
|
# Should have weather fields
|
||||||
|
assert Map.has_key?(first, :temperature)
|
||||||
|
assert Map.has_key?(first, :humidity)
|
||||||
|
|
||||||
|
# Should NOT have heavy fields
|
||||||
|
refute Map.has_key?(first, :raw_packet)
|
||||||
|
refute Map.has_key?(first, :information_field)
|
||||||
|
refute Map.has_key?(first, :body)
|
||||||
|
refute Map.has_key?(first, :data_extended)
|
||||||
|
end
|
||||||
|
|
||||||
|
test "respects time filtering" do
|
||||||
|
results = Packets.get_recent_packets_for_map(%{hours_back: 1})
|
||||||
|
callsigns = Enum.map(results, & &1.sender)
|
||||||
|
|
||||||
|
assert "MAP1" in callsigns
|
||||||
|
refute "OLD_MAP" in callsigns
|
||||||
|
end
|
||||||
|
|
||||||
|
test "respects bounds filtering" do
|
||||||
|
bounds = [-99.0, 39.5, -98.0, 40.5]
|
||||||
|
results = Packets.get_recent_packets_for_map(%{hours_back: 24, bounds: bounds})
|
||||||
|
callsigns = Enum.map(results, & &1.sender)
|
||||||
|
|
||||||
|
assert "MAP1" in callsigns
|
||||||
|
refute "OLD_MAP" in callsigns
|
||||||
|
end
|
||||||
|
|
||||||
|
test "respects limit and offset" do
|
||||||
|
results = Packets.get_recent_packets_for_map(%{hours_back: 24, limit: 1})
|
||||||
|
assert length(results) == 1
|
||||||
|
end
|
||||||
|
|
||||||
|
test "filters by callsign" do
|
||||||
|
results = Packets.get_recent_packets_for_map(%{hours_back: 24, callsign: "MAP1"})
|
||||||
|
assert length(results) == 1
|
||||||
|
assert hd(results).sender == "MAP1"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "includes object_name for APRS objects" do
|
||||||
|
results = Packets.get_recent_packets_for_map(%{hours_back: 24})
|
||||||
|
map2 = Enum.find(results, &(&1.sender == "MAP2"))
|
||||||
|
assert map2.object_name == "TestObj"
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
describe "get_nearby_stations/4" do
|
describe "get_nearby_stations/4" do
|
||||||
setup do
|
setup do
|
||||||
# Create stations at various distances
|
# Create stations at various distances
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue