Turn Fleet and Warehouse Data Into Real-Time Operational Clarity

CREATE SOURCE gps_pings WITH (
  connector = 'kafka',
  topic = 'fleet.gps_pings',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

CREATE MATERIALIZED VIEW segment_travel_time AS
SELECT
  van_id,
  window_start,
  AVG(speed_mph) AS avg_speed_mph,
  h.historical_avg_speed,
  CASE
    WHEN AVG(speed_mph) < 1 THEN 999
    ELSE ROUND(
      h.segment_distance_mi / AVG(speed_mph) * 60, 1
    )
  END AS est_segment_minutes,
  ROUND(
    h.segment_distance_mi / h.historical_avg_speed * 60, 1
  ) AS historical_segment_minutes,
  ROUND(
    (h.segment_distance_mi / AVG(speed_mph) * 60)
    - (h.segment_distance_mi / h.historical_avg_speed * 60), 1
  ) AS delay_vs_historical_min,
  MAX(stops_remaining) AS stops_remaining
FROM TUMBLE(gps_pings, ts, INTERVAL '60 SECONDS') g
JOIN route_segments h
  ON g.van_id = h.van_id
  AND h.segment_seq = (
    SELECT MAX(segment_seq) FROM route_segments r
    WHERE r.van_id = g.van_id
    AND r.segment_start_lat BETWEEN g.lat - 0.005 AND g.lat + 0.005
  )
GROUP BY van_id, window_start, h.historical_avg_speed, h.segment_distance_mi;