Simple Strominka

CREATE TABLE sandbox.yurbasov_staromynka5 AS

WITH base
AS (SELECT
  from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd') AS DAY,
  v_regno,
  concat(codd_kafka.v_gps_y, '_', codd_kafka.v_gps_x) AS camera_id,
  v_time_check
FROM codd_data.codd_kafka_batch codd_kafka
WHERE from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') < '2017-12-01'
  UNION ALL
SELECT
  from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd') AS DAY,
  v_regno,
  concat(codd_kafka.v_gps_y, '_', codd_kafka.v_gps_x) AS camera_id,
  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') < '2017-12-01'
),
pre
AS (SELECT
  v_regno,
  camera_id,
  weekofyear(v_time_check) AS w_no,
  COUNT(*) AS cnt
FROM base
WHERE camera_id IN ('55.784832_37.696632', '55.789967_37.68305', '37.696632_55.784832', '37.68305_55.789967')
GROUP BY v_regno,
         camera_id,
         weekofyear(v_time_check)),
xxxx
AS (SELECT DISTINCT
  v_regno
FROM pre
WHERE cnt >= 2),
par
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), extra AS (
SELECT
    xxxx.v_regno AS v_regno
    , sec.regno_hash AS regno_hash
    , sec.regno AS regno
FROM xxxx
 JOIN secret.regno AS sec ON xxxx.v_regno = sec.regno_hash
 GROUP BY
 v_regno, regno_hash, regno
), res AS
(SELECT
  extra.v_regno,
  extra.regno,
  par.phoneno
FROM extra
JOIN par
  ON extra.v_regno = par.carno
WHERE par.carno IS NOT NULL
GROUP BY
extra.v_regno, extra.regno,
par.phoneno) SELECT s.phone, res.regno, res.v_regno FROM res
JOIN secret.phones s ON s.phone_hash = res.phoneno