Radius-Kotelniki-3

CREATE TABLE sandbox.yurbasov_radius_kotelniki3 AS
-- RADIUS = 0.1 km
-- ,acos(sin((90.0 - v_gps_y) * pi() / 180.0) * sin((90.0 - nextgpsy) * pi() / 180.0)
--       * cos(radians(v_gps_x) - radians(nextgpsx)) +
--           cos((90.0 - v_gps_y) * pi() / 180.0) * cos((90.0 - nextgpsy) * pi() / 180.0)) * 6371000 as distance
-- UNITS: meters
-- Волг 55.690742_37.82169
-- Волг 55.690548_37.821686
-- Волг 55.69926_37.7997
-- Ряз 55.713745_37.809536
WITH base1
AS (SELECT
  from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd') AS DAY,
  v_regno,
  CASE WHEN codd_kafka.v_gps_y > 55 THEN codd_kafka.v_gps_y ELSE codd_kafka.v_gps_x END AS camera_x,
  CASE WHEN codd_kafka.v_gps_y > 55 THEN codd_kafka.v_gps_x ELSE codd_kafka.v_gps_y END AS camera_y,
  lead(v_time_check, 1) OVER (partition BY v_regno ORDER BY v_time_check) AS r_previous_overall,
  v_time_check
FROM codd_data.codd codd_kafka
WHERE (from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') <= '2018-02-28'
AND from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') >= '2018-01-01')
), tgt1 AS
(
SELECT 55.690742 AS v_gps_x, 37.82169 AS v_gps_y
UNION ALL SELECT 55.690548 AS v_gps_x, 37.821686 AS v_gps_y
UNION ALL SELECT 55.69926 AS v_gps_x, 37.7997 AS v_gps_y
UNION ALL SELECT 55.713745 AS v_gps_x, 37.809536 AS v_gps_y
), pre1 AS
(SELECT
  v_regno,
  camera_x, camera_y,
  MAX(abs(round((unix_timestamp(v_time_check) - unix_timestamp(r_previous_overall))/60, 0))) AS t_diff_next,
  weekofyear(v_time_check) AS w_no,
  COUNT(*) AS cnt
FROM base1
JOIN tgt1
ON acos(sin((90.0 - v_gps_y) * pi() / 180.0) * sin((90.0 - camera_y) * pi() / 180.0) * cos(radians(v_gps_x) - radians(camera_x))
       + cos((90.0 - v_gps_y) * pi() / 180.0) * cos((90.0 - camera_y) * pi() / 180.0)) * 6371000 <= 100
GROUP BY v_regno,
         camera_x, camera_y,
         weekofyear(v_time_check)),
f1 AS (SELECT pre1.v_regno, pre1.cnt FROM pre1 WHERE  t_diff_next >= 0),
xxxx1
AS (SELECT DISTINCT
  v_regno
FROM f1
WHERE cnt >= 6),
par1
AS (SELECT
  phoneno,
  carno
FROM (SELECT
  phoneno,
  carno,
  MAX(from_unixtime(unix_timestamp(parkingstart, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd')) AS max_date,
  COUNT(*) AS qty,
  ROW_NUMBER() OVER (PARTITION BY carno ORDER BY COUNT(*) DESC) AS rn
FROM parking_data.parking
WHERE phoneno != 'NULL'
AND from_unixtime(unix_timestamp(parkingstart, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd') >= '2017-01-01'
GROUP BY phoneno,
         carno) xx
WHERE rn = 1), extra1 AS (
SELECT
    xxxx1.v_regno AS v_regno
    , sec.regno_hash AS regno_hash
    , sec.regno AS regno
FROM xxxx1
 JOIN secret.regno AS sec ON xxxx1.v_regno = sec.regno_hash
 GROUP BY
 v_regno, regno_hash, regno
), res1 AS
(SELECT
  extra1.v_regno,
  extra1.regno,
  par1.phoneno
FROM extra1
JOIN par1
  ON extra1.v_regno = par1.carno
WHERE par1.carno IS NOT NULL
GROUP BY
extra1.v_regno, extra1.regno,
par1.phoneno), res AS
(SELECT res1.phoneno, res1.regno, res1.v_regno FROM res1)
SELECT s.phone, res.regno, res.v_regno FROM res JOIN secret.phones s ON s.phone_hash = res.phoneno