-- ==================================================================
/**
  Purpose:

    Purpose of this procedure to rollup the stats data at first interval
            to the second interval which defined at vpx_interval_def table
            It happen every 30min which is the length of the second
            interval.Out of the box the interval is defined as following
            Interval Name       Interval Value    Interval Length
            5 mins Interval       300                86400
            30 mins Interval      1800               604800
            2 hrs Interval        6400               2592000
            1 day Interval        86400              31536000

  R1,R2 &R3 job Schedule in the below order.
  21     R1 runs every 30minutes(processes HS1 data) 00:30:00 to 00:29:59
  22     R2 runs every 2 hours and initial start @ 1:15AM(processes HS2)01:15:00 to 01:14:59
  23     R3 run once a day @2:15AM(processes HS3 data)
  24
  25   e.g:
  26   0:30 R1 processes HS1 23:00-23:25 to HS2 23:00
  27   1:00 R1 processes HS1 23:30-23:55 to HS2 23:30
  28   1:15 R2 processes HS2 22:00-23:30 to HS3 22:00
  29   1:30 R1 processes HS1 00:00-00:25 to HS2 00:00
  30   2:00 R1 processes HS1 00:30-00:55 to HS2 00:30
  31   2:15 R3 processes HS3 00:00-22:00 to HS4 00:00
*/
--
-- ==================================================================

/** For efficiency, we create a custom aggregate which handles all the
 * ROLLUP_TYPE aggregates.  This simplifies the stored function query and avoid
 * to repeatadly read the same data for each ROLLUP_TYPE.
 */

-- ROLLUP based on the ROLLUP_TYPE:
        -- when 0 then 'average'
        -- when 1 then 'maximum'
        -- when 2 then 'minimum'
        -- when 3 then 'latest'
        -- when 4 then 'summation'
        -- when 5 then 'none'
DO
$$ BEGIN
IF NOT EXISTS (SELECT 1
           FROM information_schema.data_type_privileges
           WHERE object_schema = 'vc'
           AND object_name = 'vc_stat_agg_state'
           )
THEN
   CREATE TYPE vc_stat_agg_state AS (val1 numeric, val2 numeric, type numeric);
END IF;
END $$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION vc_stat_agg_transition(agg_state   vc_stat_agg_state,
                                                  val         numeric,
                                                  type        numeric,
                                                  sample_time timestamp)
  RETURNS vc_stat_agg_state
  LANGUAGE plpgsql
  IMMUTABLE
AS $$
BEGIN
  IF agg_state.type <> type THEN
    raise exception 'type must have a consistent value for each group';
  END IF;
  IF val IS NULL THEN
    RETURN agg_state;
  END IF;
  IF type < 0 or type > 4 THEN
    RETURN agg_state;
  END IF;
  agg_state.type = type;
  CASE type
    WHEN 0 THEN
      agg_state.val2 := agg_state.val2 + 1;
      agg_state.val1 := coalesce(agg_state.val1, 0) + val;
    WHEN 1 THEN
      IF agg_state.val1 IS NULL OR val > agg_state.val1 THEN agg_state.val1 := val; END IF;
    WHEN 2 THEN
      IF agg_state.val1 IS NULL OR val < agg_state.val1 then agg_state.val1 := val; END IF;
    WHEN 3 THEN
      IF sample_time IS NULL THEN
        RETURN agg_state;
      END IF;
      DECLARE
        etime numeric;
      BEGIN
        etime := extract(epoch FROM sample_time)::numeric;
        IF agg_state.val1 IS NULL OR etime > agg_state.val2 THEN
          agg_state.val1 := val;
          agg_state.val2 := etime;
        END IF;
      END;
    WHEN 4 then
      agg_state.val1 := coalesce(agg_state.val1, 0) + val;
  END case;
  RETURN agg_state;
END;
$$;

CREATE OR REPLACE FUNCTION vc_stat_agg_final(agg_state vc_stat_agg_state)
  RETURNS numeric
  LANGUAGE plpgsql
  IMMUTABLE
AS $$
BEGIN
  RETURN
    case agg_state.type
      WHEN 0 THEN case when agg_state.val2 >= 1 THEN agg_state.val1 / agg_state.val2 ELSE NULL::numeric END
      WHEN 1 THEN agg_state.val1
      WHEN 2 THEN agg_state.val1
      WHEN 3 THEN agg_state.val1
      WHEN 4 THEN agg_state.val1
      ELSE NULL::numeric
    END;
End;
$$;

DO
$$ BEGIN
IF NOT EXISTS (SELECT 1
           FROM information_schema.routines
           WHERE routine_schema = 'vc'
           AND routine_name = 'vc_stat_agg'
           )
THEN
    CREATE AGGREGATE vc_stat_agg(val numeric, type numeric, sample_time timestamp)
    (
      sfunc = vc_stat_agg_transition,
      stype = vc_stat_agg_state,
      finalfunc = vc_stat_agg_final,
      initcond = '(,0,)'
    );
END IF;
END $$ LANGUAGE plpgsql;

/** end of new custom aggregate part */

CREATE OR REPLACE FUNCTION stats_rollup1_proc()
RETURNS VOID
LANGUAGE plpgsql
VOLATILE
AS
$BODY$
DECLARE
  l_time_cur cursor (current_time_in timestamp without time zone,cursor_limit int) for
          select sample_time, partition_index
            from vpx_sample_time1
           where rollup_counter is null
             and sample_time < current_time_in - '1 hour'::interval
           order by 1 asc
           limit cursor_limit;

  l_time                      vpx_sample_time1.sample_time%type;
  l_time_rowid                vpx_sample_time2.time_id%type;
  l_cnt                       int;
  l_rollup_counter            vpx_sample_time2.rollup_counter%type;
  l_start_time_id             vpx_sample_time1.time_id%type;
  l_rollup_start_time         vpx_sample_time1.sample_time%type;
  l_rollup_flag               numeric(1):=1;
  l_next_interval             numeric(10,0);
  l_current_interval          numeric(10,0);
  l_stat_rollup_level         numeric(10,0);
  l_null_device_id            int;
  l_last_sample_time          vpx_sample_time1.sample_time%type;
  l_source_interval_seq_num   numeric(1) := 1;

  l_hs1_partition_index        vpx_sample_time1.partition_index%type;
  l_hs1_frompartitionidxclause varchar(255);
  l_hs2_frompartitionidxclause varchar(255);
  l_previous_time              vpx_sample_time1.sample_time%type;
  l_next_time                  vpx_sample_time1.sample_time%type;
  l_last_part2                 vpx_sample_time2.partition_index%type;
  l_hour_mod                   int;
  l_cursor_limit               int;
  v_errmsg                     vpx_proc_log.description%type;
  v_err_log                    CHAR(3);
  v_sqlstate                   text;
  v_message                    text;
  v_context                    text;

BEGIN

l_rollup_start_time := timezone('UTC', clock_timestamp());

-- Get DB procedures loging
select COALESCE(value,'OFF') into v_err_log
   from vpx_parameter
   where name = 'DBProc.Log.Debug.Info';

-- Calculation cursor limit depending from statistic interval
-- (rollup time in minutes) / (statistic interval in minutes)
--  + 0.5 (rollup time in minutes) / (statistic interval in minutes)
-- rollup time in minutes is always 30 min
SELECT CAST( 30/(INTERVAL_VAL/60) * 3/2  AS INTEGER) INTO l_cursor_limit
       FROM VPX_STAT_INTERVAL_DEF
       WHERE INTERVAL_SEQ_NUM = 1;
--check if rollup has been turnoff by user.
--hardcode the interval_seq_num here as there will be 3 differenct procedures
--rolup on vpx_hist_stat1, vpx_hist_stat2, vpx_hist_stat3. The table name will be hardcoded in these
--procedures so as the interval seq number.
select rollup_enabled_flg into l_rollup_flag
  from vpx_stat_interval_def
 where interval_seq_num = 2;

select max(sample_time) into l_last_sample_time
  from vpx_sample_time1
 where rollup_counter is null;

if (l_rollup_flag = 1) and (l_last_sample_time is not null) then
  select interval_val into l_current_interval
    from vpx_stat_interval_def
   where interval_seq_num = l_source_interval_seq_num;

  select stats_level, interval_val into l_stat_rollup_level, l_next_interval
    from vpx_stat_interval_def
   where interval_seq_num = l_source_interval_seq_num + 1;

  SELECT DEVICE_ID
     INTO l_null_device_id
     FROM VPX_DEVICE
     WHERE DEVICE_NAME = ''
     OR DEVICE_NAME IS NULL
     OR DEVICE_NAME = ' ';
-----------------
  -- Initializing HS2_PI(runs only once - first run of this procedure)
   select count(*) into l_cnt
     from vpx_sample_time1
    where rollup_counter is not null and
      partition_index_hs2 is not null;

   -- If no row found for RC=NULL and HS2PI=NULL,
   -- then set PARTITION_INDEX_HS2 =1
   IF l_cnt = 0 then
      --Update HS2_PI=1, for all the HS1_PI values within the min(sample_time)
      UPDATE VPX_SAMPLE_TIME1
         SET PARTITION_INDEX_HS2 = 1
       WHERE PARTITION_INDEX =
               (SELECT PARTITION_INDEX
                  FROM VPX_SAMPLE_TIME1
                 WHERE SAMPLE_TIME =
                        (SELECT MIN(SAMPLE_TIME)
                           FROM VPX_SAMPLE_TIME1
                          WHERE ROLLUP_COUNTER IS NULL
                            AND SAMPLE_TIME < l_rollup_start_time - '1 hour'::interval));
         -- later INSERT query relies on up-do-date statistics on that table
         ANALYZE vpx_sample_time1;
         l_last_part2 := 1;
   END IF;

   l_cnt:=1;
   open l_time_cur(l_rollup_start_time,l_cursor_limit);

   loop
    fetch l_time_cur into l_time, l_hs1_partition_index;
    if not found then
      exit;
    end if;

    -- Execute for all runs:
    -- Figure out which is the next destination partition index
    -- comparing the difference between the "first sample_time for last partition index2"
    -- and the "next sample_time we will have to rollup now"

    -- Get max(sample_time) from previous run, and get the previous PI# for it
    select partition_index_hs2 into l_last_part2
      from vpx_sample_time1
     where sample_time = (select max(sample_time)
                            from vpx_sample_time1
                           where partition_index_hs2 is not null);

    --Using the previous PI#, get the min(sample_time).
    select min(sample_time) into l_previous_time
      from vpx_sample_time1
     where partition_index_hs2 = l_last_part2;

    select min(sample_time) into l_next_time
      from vpx_sample_time1
     where rollup_counter is null
       and sample_time < l_rollup_start_time - '60 min'::interval;

    -- make sure each HS2 partition ends with odd hour and 30 minutes pattern(e.g. 01:30)
    l_previous_time := date_trunc('hour', l_previous_time);
    l_hour_mod := extract(hour from l_previous_time)::int;
    l_hour_mod := l_hour_mod % 2;
    l_previous_time := l_previous_time - (l_hour_mod || ' hour')::interval;

    if l_next_time - '120 min'::interval >= l_previous_time then
       l_last_part2 := l_last_part2 + 1;
    end if;

    -- VPX_HIST_STAT2 table is splited into 85 tables (custom partitions)
    -- for every 2 hours window and a view over them.
    -- So we have to cycle destination partitions during rollup on time basis
    if l_last_part2 > 85 then
       l_last_part2 := 1;
    end if;
     --record the rollup_counter for this time sample and its time_id
     SELECT rollup_counter, time_id
       into l_rollup_counter, l_start_time_id
       FROM vpx_sample_time1
      where sample_time = l_time;

     l_time:= date_trunc('hour', l_time) + (case when extract(minute from l_time) < 30 then 0 else 30 end || ' minute')::interval;

     -- in case of longer than 2 hours non-rolluped HS1 (e.g. for some reasons the job will have consecutive failures)
     -- apply the same cycle logic for the destination HS2 table
     if l_time - '120 min'::interval >= l_next_time then
        l_last_part2 := l_last_part2 + 1;
        l_next_time := l_time;
        if l_last_part2 > 85 then
           l_last_part2 := 1;
        end if;
     end if;

     l_hs1_fromPartitionIdxClause := 'VPX_HIST_STAT1_' || TO_CHAR(l_hs1_partition_index, 'FM9999');
     l_hs2_fromPartitionIdxClause := 'VPX_HIST_STAT2_' || TO_CHAR(l_last_part2, 'FM9999');

    --check if this sample time has been updated by previous bulk rollup_counter update.
    --check between the time which is 1 hour prior to this rollup and the time of this cursor sample time
    -- the difference is more than 30mins in this case which is next interval, if not, it is time to
    --end the rollup and commit previous transactions.
    if  l_rollup_counter is null
        and (l_last_sample_time + (l_current_interval || ' sec')::interval) >= (l_time + (l_next_interval || ' sec')::interval)  then

        --tag the sample time data start from currect cursor sample time record and rest of sample time record which is within 30mins(l_next_interval)
        update VPX_SAMPLE_TIME1
           set rollup_counter = (l_start_time_id::text || l_cnt)::numeric(38,0),
               partition_index_hs2 = l_last_part2
         where sample_time < l_time + (l_next_interval || ' sec')::interval
           and sample_time >= l_time;
         -- later INSERT query relies on up-do-date statistics on that table
         ANALYZE vpx_sample_time1;

        select time_id
          into l_time_rowid
          from vpx_sample_time2
         where sample_time = l_time
           and sample_interval=l_next_interval;


        if not found then
          select NEXTVAL('VPX_SAMPLE_TIME2_SEQ')
            into l_time_rowid;
          insert into vpx_sample_time2
                 (time_id, sample_time, sample_interval, partition_index)
          values
                 (l_time_rowid, l_time, l_next_interval, l_last_part2);
        end if;


        --ROLLUP based on the ROLLUP_TYPE:
        --when 0 then 'average'
        --when 1 then 'maximum'
        --when 2 then 'minimum'
        --when 3 then 'latest'
        --when 4 then 'summation'
        --when 5 then 'none'
        BEGIN
          EXECUTE '
          INSERT INTO '||l_hs2_fromPartitionIdxClause||'(counter_id, time_id, stat_val)
          SELECT st.counter_id, $1,
             vc_stat_agg(st.stat_val, sd.rollup_type, sm.sample_time)
           FROM vpx_sample_time1 sm
           JOIN  '||l_hs1_fromPartitionIdxClause||' st
              ON st.time_id = sm.time_id
           JOIN vpx_stat_counter sc ON st.counter_id = SC.COUNTER_ID
           JOIN vpx_stat_def sd ON SC.STAT_ID = SD.ID
           WHERE sm.rollup_counter = ($2::text || $3)::numeric(38, 0)
             AND ((SC.DEVICE_ID = $4 and SD.STAT_LEVEL <= $5)
                  or (SC.DEVICE_ID <> $6 and SD.PERDEVICE_STAT_LEVEL <= $7))
             AND SD.ROLLUP_TYPE BETWEEN 0 AND 4
           GROUP BY st.counter_id'
          USING l_time_rowid,
                l_start_time_id, l_cnt,
                l_null_device_id, l_stat_rollup_level,
                l_null_device_id, l_stat_rollup_level;
        EXCEPTION
          WHEN OTHERS THEN
               v_errmsg := 'Error:' || cast(sqlstate as VARCHAR(30)) || ' :' || coalesce(SQLERRM, '');
              IF v_err_log = 'ON' THEN
                 INSERT INTO VPX_PROC_LOG (PROC_NAME, EVENT, EVENT_DT, ERR_FLAG, DESCRIPTION)
                      VALUES ('STATS_ROLLUP1_PROC','ERR', TIMEZONE('UTC',clock_timestamp()), -1, SUBSTR(v_errmsg, 1, 1000));
              END IF;
        END;

        l_cnt:=l_cnt+1;

      END IF;

      END LOOP;
    close l_time_cur;


ELSE
     NULL;
END IF;
--
EXCEPTION WHEN OTHERS THEN
    GET STACKED DIAGNOSTICS
        v_sqlstate = returned_sqlstate,
        v_message = message_text,
        v_context = pg_exception_context;
    RAISE LOG 'sqlstate: %', v_sqlstate;
    RAISE LOG 'message: %', v_message;
    RAISE LOG 'context: %', v_context;
  WHEN OTHERS THEN
    PERFORM 1
    FROM    pg_cursors
    WHERE   name = 'l_time_cur';

    IF FOUND THEN
      CLOSE l_time_cur; -- close the cursor no matter what happens
    END IF;
    v_errmsg := 'Error:' || cast(sqlstate as VARCHAR(30)) || ' :' || coalesce(SQLERRM, '');
    if v_err_log = 'ON' then
        INSERT INTO VPX_PROC_LOG (PROC_NAME, EVENT, EVENT_DT, ERR_FLAG, DESCRIPTION)
           VALUES ('STATS_ROLLUP1_PROC','ERR', TIMEZONE('UTC',clock_timestamp()), -1, SUBSTR(v_errmsg, 1, 1000));
    end if;
end;
$BODY$
