text: CREATE OR REPLACE PACKAGE BODY UTIL.ELO
  AS
  
  
    GV_ERROR    constant varchar2(10) := 'ERROR';
    GV_SUCCESS  constant varchar2(10) := 'SUCCESS';
    GV_RUNNING  constant varchar2(10) := 'RUNNING';
    GV_READY    constant varchar2(10) := 'READY';
  
    gv_job_module   VARCHAR2(50)  := 'ELO';                 -- Job Module Name : Extract Load package
    gv_pck          VARCHAR2(50)  := 'ELO';                 -- PLSQL Package Name
    gv_job_owner    VARCHAR2(50)  := 'UTIL';                -- Owner of the Job
    gv_proc         VARCHAR2(100);                          -- Procedure Name
    gv_delim        VARCHAR2(10)  := ' : ';                 -- Delimiter Used In Logging
  
    gv_sql_errm     VARCHAR2(4000);                         -- SQL Error Message
    gv_sql_errc     NUMBER;                                 -- SQL Error Code
    gv_sql          LONG := '';
    gv_date_format  varchar2(30) := 'yyyy.mm.dd hh24:mi:ss';
  
  
    function fun_get_delta_col_type(
      i_db_link varchar2,
      i_table   varchar2,
      i_column  varchar2) return varchar2;
  
    procedure prc_update_last_delta(
      i_name            varchar2,
      i_table           varchar2,
      i_delta_col       varchar2,
      i_delta_col_type  varchar2
    );
  
    procedure set_status(i_name varchar2, i_status varchar2) as
      v_status varchar2(100);
      status_change_error exception;
      already_running_error exception;
      pragma exception_init(status_change_error, -20001);
      pragma exception_init(already_running_error, -20002);
    begin
  
  	select t.status INTO v_status from util.elo_tables t where upper(t.name) = upper(i_name);
  
      if upper(i_status) = 'RUNNING' and v_status = 'RUNNING' then
        raise_application_error(-20002, i_name || ' is already running!');
      end if;
      update util.elo_tables set
        status = upper(i_status),
        start_time = decode(upper(i_status), 'RUNNING', sysdate, start_time),
        end_time = case when upper(i_status) in ('ERROR', 'SUCCESS') then sysdate else end_time end
      where upper(name) = upper(i_name);
  
      commit;
    end;
  
    function is_excluded(i_name varchar2) return boolean
    is
      v_count number := 0;
    begin
      select count(1) into v_count from util.elo_tables t
      where
        upper(t.name) = upper(i_name) and
        t.excluded = 1;
      return case v_count when 0 then false else true end;
    end;
  
  
    procedure reset is
    begin
      gv_proc := 'RUN';
      pl.logger := util.logtype.init(gv_pck||'.'||gv_proc);
      gv_sql := 'update util.elo_tables set
        start_time = null,
        end_time = null,
        status = upper('''|| GV_READY || ''')
      ';
      execute immediate gv_sql;
      pl.logger.success(SQL%ROWCOUNT || ' : updated', gv_sql);
      commit;
    exception
      when others then
        pl.logger.error(SQLCODE||' : '||SQLERRM, gv_sql);
        raise;
    end;
  
  
    procedure run(i_name varchar2, i_drop_create number default null, i_analyze number default null, i_index_drop number := 1)
    is
      v_db_link         varchar2(60);
      v_source          varchar2(100);
      v_target          varchar2(100);
      v_filter          varchar2(4000);
      v_source_hint     varchar2(4000);
      v_target_hint     varchar2(4000);
      v_analyze         number;
      v_drop_create     number;
      v_create_options  varchar2(4000);
      v_delta_column    varchar2(50);
      v_last_delta      varchar2(1000);
      v_source_cols     long;
      v_target_cols     long;
      v_delta_data_type varchar2(50);
      v_index_ddls      dbms_sql.varchar2_table;
      v_index_created   boolean := false;
      v_name            varchar2(100) := upper(i_name);
    begin
  
      gv_proc := 'RUN';
  
      -- Initialize Log Variables
      pl.logger := util.logtype.init(gv_pck||'.'||gv_proc||'-'||v_name);
  
      if is_excluded(v_name) = true then
        pl.logger.warning(v_name || ' is EXCLUDED!');
        goto end_proc;
      end if;
  
      set_status(v_name, GV_RUNNING);
  
      pl.logger.info('Started extraction ...');
  
      select
        db_link, source, target, filter, source_hint, target_hint, v_delta_column, last_delta,
        drop_create, create_options, ANALYZE
      into
        v_db_link, v_source, v_target, v_filter, v_source_hint, v_target_hint, v_delta_column, v_last_delta,
        v_drop_create, v_create_options, v_analyze
      from util.ELO_TABLES where name = v_name;
  
      if trim(v_target_hint) is not null and instr(v_target_hint,'/*+') = 0
      then
        v_target_hint := '/*+ '||v_target_hint||' */';
      end if;
  
      if trim(v_source_hint) is not null and instr(v_source_hint,'/*+') = 0
      then
        v_source_hint := '/*+ '||v_source_hint||' */';
      end if;
  
      IF v_delta_column IS NOT NULL THEN
        v_delta_data_type := fun_get_delta_col_type(v_db_link, v_source, v_delta_column);
      END IF;
  
      if v_filter is not null or v_delta_column is not null then
        v_filter := 'WHERE ' || v_filter;
        if v_delta_column is not null then
          v_filter := v_filter || ' AND '||v_delta_column||'>'||
          case v_delta_data_type
            when 'DATE'     then pl.date_string(to_date(v_last_delta,gv_date_format))
            when 'NUMBER'   then to_number(v_last_delta)
            when 'CHAR'     then v_last_delta
            when 'VARCHAR'  then v_last_delta
            else v_last_delta
          end;
        end if;
      end if;
  
      begin
        for c in (select source_col, target_col from util.ELO_COLUMNS where excluded = 0 and name = v_name) loop
          v_source_cols := v_source_cols || c.source_col || ',';
          v_target_cols := v_target_cols || c.target_col || ',';
        end loop;
  
        v_source_cols := rtrim(v_source_cols, ',');
        v_target_cols := rtrim(v_target_cols, ',');
      exception when no_data_found then
        pl.logger.warning(v_name || ': All columns excluded or no elo def!');
        goto end_proc;
      end;
  
      if (i_drop_create is not null and i_drop_create != 0) or (i_drop_create is null and v_drop_create != 0) then
  
        pl.drop_table(v_target);
  
        gv_sql := '
          CREATE TABLE '|| v_target ||'
          '||v_create_options||'
          ('||v_target_cols|| ')
          AS
          SELECT '||v_source_hint||'
          '||v_source_cols||'
          FROM
            '||v_source||'@'||v_db_link||'
            '||v_filter||'';
      else
        gv_sql := '
          INSERT '||v_target_hint||' INTO '|| v_target ||'
          ('||v_target_cols|| ')
          SELECT '||v_source_hint||'
          '||v_source_cols||'
          FROM
            '||v_source||'@'||v_db_link||'
            '||v_filter||'';
  
        if v_delta_column is null then
          pl.logger.info('truncate table ' || v_target);
          execute immediate 'truncate table ' || v_target;
        end if;
  
      end if;
  
  
      if i_index_drop != 0 then
        v_index_ddls := pl.index_ddls(v_target);
        pl.drop_indexes(v_target);
      end if;
  
      execute immediate gv_sql;
      pl.logger.success(SQL%ROWCOUNT || ' : inserted', gv_sql);
      commit;
  
      if i_index_drop != 0 then
        pl.exec(v_index_ddls, true);
        v_index_created := true;
      end if;
  
      if v_delta_column is not null then
        prc_update_last_delta(
          i_name  => v_name,
          i_table => v_target,
          i_delta_col => v_delta_column,
          i_delta_col_type => v_delta_data_type
        );
      end if;
  
      if (i_analyze is not null and i_analyze != 0) or (v_analyze != 0)  then
        pl.gather_table_stats(i_table => v_target);
      end if;
  
      <<end_proc>>
      set_status(v_name, GV_SUCCESS);
    exception
      when others then
        pl.logger.error(SQLCODE||' : '||SQLERRM, gv_sql);
        set_status(v_name, GV_ERROR);
        if i_index_drop != 0 and v_index_created = false then
          pl.exec(v_index_ddls, true);
        end if;
        rollback;
        raise;
    end;
  
    function fun_get_delta_col_type(i_db_link varchar2, i_table varchar2, i_column varchar2) return varchar2
    is
      v_owner       varchar2(30) := substr(i_table,1,instr(i_table, '.')-1);
      v_table_name  varchar2(30) := substr(i_table,instr(i_table, '.')+1);
      v_col_type    varchar2(50);
    begin
  
      gv_sql:= '
        select data_type from all_tab_cols@'||i_db_link||'
        where owner = '''||v_owner||''' and table_name='''||v_table_name||''' and column_name = '''|| i_column||'''
      ';
  
      execute immediate gv_sql into v_col_type;
  
      return gv_sql;
  
    end;
  
    procedure prc_update_last_delta(
      i_name            varchar2,
      i_table           varchar2,
      i_delta_col       varchar2,
      i_delta_col_type  varchar2
    )
    is
      v_last_delta varchar2(1000);
    begin
  
      if i_delta_col_type = 'DATE' then
        gv_sql := 'to_char(max('||i_delta_col||'),'''||gv_date_format||''')';
      else
        gv_sql := 'max('||i_delta_col||')';
      end if;
  
      gv_sql := 'select /*+ parallel(16) */ '||gv_sql||' from '||i_table;
  
      execute immediate gv_sql into v_last_delta;
  
      gv_sql := '
        update util.ELO_TABLES set last_delta = '''||v_last_delta||'''
        where name = '''||i_name||'''
      ';
  
      execute immediate gv_sql;
  
      commit;
    end;
  
    -- private
    function get_remote_col_list(i_table varchar2, i_db_link varchar2) return dbms_sql.varchar2_table is
      type source_cursor_type is ref cursor;
      c source_cursor_type;
      v_column_name varchar2(128);
      v_result dbms_sql.varchar2_table;
      v_index number := 1;
      v_table varchar2(128) := i_table;
    begin
  
      if instr(v_table, '"') = 0 then v_table := upper(v_table); end if;
  
      gv_sql := '
        select column_name from all_tab_cols@'||i_db_link||'
        where
          '''||v_table||''' = owner||''.''||table_name
          and data_type not in (''CLOB'', ''LOB'') --! todo: add others
          and hidden_column = ''NO''
        order by column_id
      ';
  
      open c for gv_sql;
      loop
        fetch c into v_column_name; exit when c%notfound;
        v_result(v_index) := v_column_name;
        v_index := v_index + 1;
      end loop;
  
      return v_result;
    exception
      when others then
        pl.logger.error(gv_sql);
        raise;
    end;
  
    function is_in_elo_tables(i_name varchar2) return boolean is
      v_result number;
      v_name   varchar2(128) := i_name;
    begin
      if instr(i_name, '"') = 0 then
        v_name := upper(v_name);
      end if;
  
      select count(1) into v_result from util.elo_tables t
      where t.name = v_name;
  
      return case v_result when 1 then true else false end;
    end;
  
    function is_in_elo_columns(i_name varchar2, i_column varchar2) return boolean is
      v_result number;
      v_name   varchar2(128) := i_name;
      v_column varchar2(128) := i_column;
    begin
      if instr(i_name, '"') = 0 then
        v_name := upper(v_name);
      end if;
  
      if instr(i_column, '"') = 0 then
        v_column := upper(v_column);
      end if;
  
      select count(1) into v_result from util.elo_columns t
      where t.name = v_name and t.target_col = v_column;
  
      return case v_result when 1 then true else false end;
    end;
  
    -- sugar for elo.define
    procedure d(
      i_source  varchar2,
      i_name    varchar2 := null,
      i_target  varchar2 := null,
      i_db_link varchar2 := null,
      i_columns long := null,
      i_filter  varchar2 := null,
      i_source_hint varchar2 := null,
      i_target_hint varchar2 := null
    ) is
    begin
      define(i_source, i_name, i_target, i_db_link, i_columns, i_filter, i_source_hint, i_target_hint);
    end;
  
    -- sugar for elo.define
    procedure def(
      i_source  varchar2,
      i_name    varchar2 := null,
      i_target  varchar2 := null,
      i_db_link varchar2 := null,
      i_columns long := null,
      i_filter  varchar2 := null,
      i_source_hint varchar2 := null,
      i_target_hint varchar2 := null
    ) is
    begin
      define(i_source, i_name, i_target, i_db_link, i_columns, i_filter, i_source_hint, i_target_hint);
    end;
  
  
  
    -- Helper method to define extractions quickly. Inserts definitions to util.elo_tables and
    -- util.elo_columns.
    --
    -- If only i_source is given, it should be in the form of [remote_schema.remote_table@my_db_link]
    -- If i_columns is not given then all columns will be extracted
    -- If i_columns is given then new columns in the list will be inserted to the elo_columns.
    -- Definitions work in insert mode which means if table or column definitions already exists
    -- then only the new ones will be inserted. This method does not override any existing records in
    -- elo_tables or elo_columns
    procedure define(
      i_source  varchar2,
      i_name    varchar2 := null,
      i_target  varchar2 := null,
      i_db_link varchar2 := null,
      i_columns long := null,
      i_filter  varchar2 := null,
      i_source_hint varchar2 := null,
      i_target_hint varchar2 := null
    ) is
      e_source_parameter_error  exception;
      pragma exception_init(e_source_parameter_error,  -20170);
      v_tokens dbms_sql.varchar2_table;
      v_token  varchar2(128);
    begin
      gv_proc := 'define';
      pl.logger := util.logtype.init(gv_pck||'.'||gv_proc||'-'||i_source);
  
      if i_source is null then
        raise_application_error(-20170, 'source parameter can not be null');
      end if;
  
      if substr(i_source, -1) = ' ' or substr(i_source, 1, 1) = ' ' then
        raise_application_error(-20170, 'parameters can not begin or end with a space');
      end if;
  
      if instr(i_source, '@') > 0 then
        v_tokens := pl.split(i_source, '@');
        define(
          i_source  => v_tokens(1),
          i_name    => i_name,
          i_target  => i_target,
          i_db_link => v_tokens(2),
          i_columns => i_columns,
          i_filter  => i_filter,
          i_source_hint => i_source_hint,
          i_target_hint => i_target_hint
        );
        return;
      end if;
  
      if i_target is null then
        define(
          i_source  => i_source,
          i_name    => i_name,
          i_target  => i_source,
          i_db_link => i_db_link,
          i_columns => i_columns,
          i_filter  => i_filter,
          i_source_hint => i_source_hint,
          i_target_hint => i_target_hint
        );
        return;
      end if;
  
      if i_name is null then
        v_tokens := pl.split(i_source, '.');
        define(
          i_source  => i_source,
          i_name    => v_tokens(2),
          i_target  => i_source,
          i_db_link => i_db_link,
          i_columns => i_columns,
          i_filter  => i_filter,
          i_source_hint => i_source_hint,
          i_target_hint => i_target_hint
        );
        return;
      end if;
  
      if i_columns is null then
        define(
          i_source  => i_source,
          i_name    => i_name,
          i_target  => i_source,
          i_db_link => i_db_link,
          i_columns => pl.make_string(get_remote_col_list(i_source, i_db_link)),
          i_filter  => i_filter,
          i_source_hint => i_source_hint,
          i_target_hint => i_target_hint
        );
        return;
      end if;
  
      if is_in_elo_tables(i_name) = false then
        gv_sql := '
          insert into util.elo_tables (
            name, db_link, source, target, filter, drop_create,
            source_hint, target_hint
          ) values (
            '''||i_name||''', '''||i_db_link||''', '''||i_source||''', '''||i_target||''', '''||i_filter||''', 1,
            '''||i_source_hint||''', '''||i_target_hint||'''
          )
        ';
        execute immediate gv_sql;
      end if;
  
      v_tokens := pl.split(i_columns);
      for i in v_tokens.first .. v_tokens.last loop
  
        v_token := trim(translate(v_tokens(i), chr(10)||chr(13), ' '));
        if instr(v_token, '"') = 0 then v_token := upper(v_token); end if;
  
        if is_in_elo_columns(i_name, v_token) = false then
          gv_sql := '
            insert into util.elo_columns (
              name, source_col, target_col
            ) values (
              '''||i_name ||''',
              '''||v_token||''',
              '''||v_token||'''
            )
          ';
          execute immediate gv_sql;
        end if;
      end loop;
  
      commit;
  
    exception
      when others then
        pl.logger.error(gv_sql);
        rollback;
        raise;
    end;
  
  
    procedure add_filter(i_name varchar2, i_filter varchar2) is
    begin
      if i_name is null then return; end if;
      gv_proc   := 'add_filter';
      pl.logger := util.logtype.init(gv_pck||'.'||gv_proc);
  
      gv_sql := '
        update util.elo_tables
        set filter =
          case when filter is null then '''||i_filter||''' else '''||' and '||i_filter||''' end
        where
          upper(name) = '''||upper(i_name)||'''
      ';
  
      execute immediate gv_sql;
      pl.logger.success(gv_sql);
      commit;
  
    exception
      when others then
        pl.logger.error(gv_sql);
        rollback;
        raise;
    end;
  
    -- sugar for delete
    procedure del(i_name varchar2) is
    begin
      elo.delete(i_name);
    end;
  
    -- delete a mapping by name
    procedure delete(i_name varchar2) is
    begin
      gv_proc   := 'delete';
      pl.logger := util.logtype.init(gv_pck||'.'||gv_proc);
  
      gv_sql := 'delete from util.elo_columns where upper(name)='''||upper(i_name)||'''';
      pl.logger.success(SQL%ROWCOUNT || ' : deleted', gv_sql);
      execute immediate gv_sql;
  
      gv_sql := 'delete from util.elo_tables where upper(name)='''||upper(i_name)||'''';
      pl.logger.success(SQL%ROWCOUNT || ' : deleted', gv_sql);
      execute immediate gv_sql;
  
      commit;
  
    exception
      when others then
        pl.logger.error(gv_sql);
        rollback;
        raise;
    end;
  
  
    -- sugar for delete_columns
    procedure delcols(i_name varchar2) is
    begin
      delete_columns(i_name);
    end;
  
  
    procedure delete_columns(i_name varchar2) is
    begin
  
      gv_proc   := 'delete_columns';
      pl.logger := util.logtype.init(gv_pck||'.'||gv_proc);
  
      gv_sql := 'delete from util.elo_columns where name='''||i_name||'''';
      pl.logger.success(SQL%ROWCOUNT || ' : deleted', gv_sql);
      execute immediate gv_sql;
  
      commit;
  
    exception
      when others then
        pl.logger.error(gv_sql);
        rollback;
        raise;
    end;
  
  
  END;