oracle转greenplum功能实现

项目中遇到的，对greenplum不是很熟悉，所以花了一天完成的东西，主要用来将
oracle的表定义信息转换成greenplum形式，可以实现批量化，不用powerdisger一
个个点了。

功能说明：

  将oracle的表（表结构定义、约束、默认值及分区信息）定义信息转换成greenplum
格式。

包说明：

create or replace package metadata_ora2gp as
  function split(in_tablist in varchar2) return dbms_sql.varchar2a;
  procedure convert_tab_create_ddl(in_tablist varchar2);
  procedure convert_index_create_ddl(in_tablist varchar2);
  procedure convert_constraint_ddl(in_tablist varchar2);
end;
/

包体：

create or replace package body metadata_ora2gp as

  function split(in_tablist in varchar2) return dbms_sql.varchar2a as
    v_tablist dbms_sql.varchar2a;
  begin

    select distinct trim(regexp_substr(regexp_replace(in_tablist,
                                                       '([^ .
]+) +|[.]([^ .
]+)',
                                                       '\1.\2'),
                                        '[^
]+',
                                        1,
                                        level)) bulk collect
      into v_tablist
      from dual
    connect by level <= regexp_count(in_tablist, chr(10)) - 1;

    return v_tablist;
  end split;

  procedure convert_tab_create_ddl(in_tablist varchar2) as
    /*
      Author:         Zhangyu@BI
      CreateDate:     2014/05/14
      Version:        1.0Beta
      Function:       convert oracle table create ddl to greenplum table create ddl.

      11g支持分区模式：RANGE-RANGE、LIST-RANGE、LIST-HASH和LIST-LIST。
    */
    v_dk varchar2(4000); --greenplum dk键
    v_ddl_stmt clob; --ddl语句变量
    v_tablist  dbms_sql.varchar2a; --表名数组，格式为owner.table_name

    --列信息游标，包括字段类型、默认值及非空值等
    cursor cur_collist(in_tab varchar2) is
      select mtc.OWNER,
             mtc.TABLE_NAME,
             mtc.COLUMN_NAME,
             mtc.DATA_TYPE,
             mtc.DATA_LENGTH,
             mtc.DATA_PRECISION,
             mtc.DATA_SCALE,
             mtc.NULLABLE,
             mtc.COLUMN_ID,
             replace(lower(mtc.DATA_DEFAULT), 'sysdate', 'current_date') as DATA_DEFAULT, --greenplum中没有sysdate函数
             t.gp_type_name
        from sys.my_tab_columns mtc, ORA2GP_TYPE_MAPPING t
       where mtc.OWNER || '.' || mtc.TABLE_NAME in (upper(in_tab))
         and mtc.DATA_TYPE = t.ora_type_name(+)
       order by mtc.COLUMN_ID;

    type collist_tab is table of cur_collist%rowtype;
    v_collist collist_tab;

    v_finalcol varchar2(32767); --存放单列信息的变量

    type part_rec is record(
      col1 varchar2(4000),
      col2 varchar2(4000));
    type part_tab is table of part_rec;
    v_part part_tab; --存放单个分区信息的数组

    v_partsql clob; --存放最终分区语句的变量

  begin
    /************************************************************************************/
    --拆分参数存放入表清单数组
    /*
      格式必须为：
      owner.table_name
      或者
      owner table_name
    */
    v_tablist := metadata_ora2gp.split(in_tablist);

    /************************************************************************************/
    --生成表定义语句

    for loop_idx in 1 .. v_tablist.count loop

      open cur_collist(v_tablist(loop_idx));

      if cur_collist%notfound then
        v_collist := null;
        dbms_output.put_line('Error: no ' || v_tablist(loop_idx) ||
                             ' table.');

      else

        fetch cur_collist bulk collect
          into v_collist;
        close cur_collist;

      end if;

      for inner_idx in 1 .. v_collist.count loop

        --列映射转换
        select v_collist(inner_idx).column_name || ' ' || case
                  when v_collist(inner_idx)
                   .data_type in ('VARCHAR2', 'NVARCHAR2', 'CHAR', 'NCHAR') then
                   v_collist(inner_idx)
                   .gp_type_name || '(' || v_collist(inner_idx).data_length || ')'
                  when v_collist(inner_idx).data_type like 'TIMESTAMP%' then
                   v_collist(inner_idx).data_type
                  else
                   v_collist(inner_idx).gp_type_name
                end || case
                  when v_collist(inner_idx).nullable = 'N' then
                   ' NOT NULL '
                  else
                   ' '
                end || case
                  when v_collist(inner_idx).data_default is not null then
                   ' DEFAULT ' || v_collist(inner_idx).data_default
                  else
                   ' '
                end
          into v_finalcol
          from dual;

        --拼接create table sql
        if inner_idx = 1 then
          if v_collist.count = 1 then
            v_ddl_stmt := 'create table ' || v_collist(inner_idx).owner || '.' || v_collist(inner_idx)
                         .table_name || '(' || v_finalcol || ')' || chr(10);
          else
            v_ddl_stmt := 'create table ' || v_collist(inner_idx).owner || '.' || v_collist(inner_idx)
                         .table_name || '(' || v_finalcol || ',' || chr(10);
          end if;
        elsif inner_idx = v_collist.last then

          /************************************************************************************/
          --生成distrubited key
          <<DKGeneratedBlock>>
          begin
            with t as
             (select listagg(dic.COLUMN_NAME, ',') within group(order by dic.COLUMN_POSITION) as collist,
                     count(*) as col_cnt,
                     min(count(*)) over() min_col_cnt
                from dba_indexes dc, dba_ind_columns dic
               where dc.table_owner || '.' || dc.table_name in
                     (upper(v_collist(inner_idx)
                            .owner || '.' || v_collist(inner_idx).table_name))
                 and dc.uniqueness = 'UNIQUE'
                 and dc.owner = dic.INDEX_OWNER
                 and dc.index_name = dic.INDEX_NAME
               group by dic.INDEX_OWNER, dic.INDEX_NAME)
            select ' DISTRIBUTED BY(' || collist || ') '
              into v_dk
              from t
             where col_cnt = min_col_cnt;

          exception
            when others then
              v_dk := '';
          end;
          <<DKGeneratedBlock>>

          v_ddl_stmt := v_ddl_stmt || v_finalcol || ')' || chr(10) || v_dk;
        else
          v_ddl_stmt := v_ddl_stmt || v_finalcol || ',' || chr(10);
        end if;

      end loop;
      /************************************************************************************/
      --生成分区语句
      <<PartitionGeneratedBlock>>
      begin

        with t1 as
         (select dpt.partitioning_type, dpt.subpartitioning_type
            from dba_part_tables dpt
           where dpt.partitioning_type not in ('HASH')
             and dpt.subpartitioning_type not in ('HASH')
                --gp默认是hash分区
                --缺陷：list-hash分区也被排除
             and dpt.owner || '.' || dpt.table_name in
                 (upper(v_tablist(loop_idx)))),
        t2 as
         (select listagg(dpkc.column_name, ',') within group(order by dpkc.column_position) pkcol
            from dba_part_key_columns dpkc
           where dpkc.owner || '.' || dpkc.name = upper(v_tablist(loop_idx))),
        t3 as
         (select listagg(dskc.column_name, ',') within group(order by dskc.column_position) spkcol
            from dba_subpart_key_columns dskc
           where dskc.owner || '.' || dskc.name = upper(v_tablist(loop_idx))),
        t4 as
         (select ' partition by ' || partitioning_type || '(' || pkcol || ')' ||
                 chr(10) || ' subpartition by ' || subpartitioning_type || '(' ||
                 spkcol || ')' part_col,
                 t1.*,
                 t2.*,
                 t3.*
            from t1
            left outer join t2
              on 1 = 1
            left outer join t3
              on 1 = 1),
        t5 as
         (select dtp.partition_name,
                 dtp.partition_position,
                 case
                   when regexp_like(dtp.high_value, '^TO_DATE *[(]') then
                    'timestamp ''' ||
                    trim(regexp_substr(dtp.high_value, '[^'']+', 1, 2)) || ''''
                   else
                    dtp.high_value
                 end as high_value, --修
                 dtsp.partition_name as ppname,
                 dtsp.subpartition_name,
                 dtsp.subpartition_position,
                 case
                   when regexp_like(dtsp.high_value, '^TO_DATE *[(]') then
                    'timestamp ''' ||
                    trim(regexp_substr(dtsp.high_value, '[^'']+', 1, 2)) || ''''
                   else
                    dtsp.high_value
                 end as sphval --修
            from sys.my_tab_partitions dtp, sys.my_tab_subpartitions dtsp
           where dtp.table_owner || '.' || dtp.table_name =
                 upper(v_tablist(loop_idx))
             and dtp.table_name = dtsp.table_name(+)
             and dtp.table_owner = dtsp.table_owner(+)
             and dtp.partition_name = dtsp.partition_name(+)),
        t6 as
         (select case
                   when upper(high_value) in ('DEFAULT', 'MAXVALUE') then
                    ' DEFAULT PARTITION other '
                   else
                    ' partition ' || partition_name || case
                      when partitioning_type = 'RANGE' then
                       ' start (' || high_value || ') INCLUSIVE '
                      when partitioning_type = 'LIST' then
                       ' values(' || high_value || ')'
                      else
                       null
                    end
                 end as p_key,
                 case
                   when upper(sphval) in ('DEFAULT', 'MAXVALUE') then
                    ' DEFAULT SUBPARTITION other '
                   else
                    ' subpartition ' || subpartition_name || case
                      when subpartitioning_type = 'RANGE' then
                       ' start(' || sphval || ') INCLUSIVE'
                      when subpartitioning_type = 'LIST' then
                       ' values(' || sphval || ')'
                      else
                       null
                    end
                 end as sp_key,
                 t4.*,
                 t5.*
            from t4, t5),
        t7 as
         (select lag(p_key) over(order by partition_position, subpartition_position) as lg_pkey,
                 sp_key,
                 partition_position,
                 count(*) over() as cnt,
                 row_number() over(order by partition_position, subpartition_position) row_cnt,
                 count(subpartition_position) over(partition by partition_position) spcnt,
                 subpartition_position,
                 p_key,
                 part_col
            from t6),
        t8 as
         (select case
                   when p_key = lg_pkey then
                    null
                   else
                    p_key
                 end as p_key,
                 t7.sp_key,
                 t7.cnt,
                 t7.row_cnt,
                 t7.spcnt,
                 t7.partition_position,
                 t7.subpartition_position,
                 t7.part_col
            from t7)
        select regexp_replace(regexp_replace(case
                                               when p_key is not null then
                                                p_key || chr(10) || '('
                                               else
                                                null
                                             end || case
                                               when row_cnt = cnt then
                                                sp_key || ')' || chr(10)
                                               when spcnt = subpartition_position then
                                                sp_key || '),' || chr(10)
                                               else
                                                sp_key || ',' || chr(10)
                                             end,
                                             '[(] subpartition ,$',
                                             ','),
                              '[(] subpartition [)]$') key_list,
               t8.part_col bulk collect
          into v_part
          from t8
         order by partition_position, subpartition_position;
      exception
        when others then
          v_part := null;
      end;
      <<PartitionGeneratedBlock>>

      for loop_idx in 1 .. v_part.count loop
        v_partsql := v_partsql || v_part(loop_idx).col1;
      end loop;

      if v_part.count <> 0 then

        v_partsql := regexp_replace(v_part(1).col2,
                                    'subpartition by NONE[(][)]') ||
                     chr(10) || '(' || chr(10) || v_partsql || ')';
      else
        v_part := null;
      end if;

      /************************************************************************************/
      --将分区语句拼接到sql中
      v_ddl_stmt := v_ddl_stmt || v_partsql;

      dbms_output.put_line(v_ddl_stmt || ';');
      v_partsql := '';
    end loop;

  end convert_tab_create_ddl;

  procedure convert_index_create_ddl(in_tablist in varchar2) as
    /*
      Author:         Zhangyu@BI
      CreateDate:     2014/05/14
      Version:        1.0Beta
      Function:       convert oracle index create ddl to greenplum index create ddl.
    */
    v_idxlist dbms_sql.varchar2a;
    v_tablist dbms_sql.varchar2a;

  begin
    /************************************************************************************/
    --拆分参数存放入表清单数组
    /*
      格式必须为：
      owner.table_name
      或者
      owner table_name
    */
    v_tablist := metadata_ora2gp.split(in_tablist);

    for loop_idx in 1 .. v_tablist.count loop

      with t1 as
       (select di.owner,
               di.index_name,
               di.uniqueness,
               di.index_type,
               dic.COLUMN_NAME,
               partitioned,
               di.table_owner,
               di.table_name,
               dic.COLUMN_POSITION,
               ' ' DESCEND --gp中无升降概念
          from dba_indexes di, dba_ind_columns dic
         where di.owner = dic.INDEX_OWNER
           and di.index_name = dic.INDEX_NAME
           and di.table_owner || '.' || di.table_name in
               (upper(v_tablist(loop_idx)))
           and di.index_type not like 'FUNCTION-BASED%'),
      t2 as
       (select owner,
               index_name,
               uniqueness,
               table_owner,
               partitioned,
               table_name,
               index_type,
               listagg(COLUMN_NAME || ' ' || DESCEND, ',') within group(order by COLUMN_POSITION) collist
          from t1
         group by owner,
                  index_name,
                  uniqueness,
                  index_type,
                  partitioned,
                  table_owner,
                  table_name)
      select 'create ' || case
                when index_type = 'BITMAP' then

                 index_type
                when uniqueness = 'UNIQUE' then
                 uniqueness
                else
                 null
              end || ' index ' || t2.index_name || ' on ' ||
              table_owner || '.' || t2.table_name || '(' || collist || ') ' ||
             --dpi.locality
              ';' as idx_crtsql bulk collect
        into v_idxlist
        from t2, dba_part_indexes dpi
       where t2.owner = dpi.owner(+)
         and t2.index_name = dpi.index_name(+);

      for loop_idx in 1 .. v_idxlist.count loop
        dbms_output.put_line(v_idxlist(loop_idx));
      end loop;

    end loop;

  end convert_index_create_ddl;

  procedure convert_constraint_ddl(in_tablist varchar2) as
    /*
      Author:         Zhangyu@BI
      CreateDate:     2014/05/14
      Version:        1.0Beta
      Function:       convert oracle index create ddl to greenplum index create ddl.
    */
    v_conslist dbms_sql.varchar2a;
    v_tablist  dbms_sql.varchar2a;

  begin
    /************************************************************************************/
    --拆分参数存放入表清单数组
    /*
      格式必须为：
      owner.table_name
      或者
      owner table_name
    */
    v_tablist := metadata_ora2gp.split(in_tablist);

    for loop_idx in 1 .. v_tablist.count loop

      with t as
       (
        --check约束
        select 'alter table ' || dc.owner || '.' || dc.table_name ||
                ' add constraint ' || dc.constraint_name || ' check (' ||
                replace(dc.search_condition,'"') || ');' as cons_ddl
          from sys.my_constraints dc
         where dc.constraint_type in ('C')
           and dc.status = 'ENABLED'
           and dc.owner || '.' || dc.table_name = upper(v_tablist(loop_idx))
        union all
        --reference约束
        select 'alter table ' || dc.owner || '.' || dc.table_name ||
                ' add constraint ' || dc.constraint_name || ' foreign key(' ||
                collist || ') references ' || dc.r_owner || '.' ||
                dc1.table_name || ';'
          from sys.my_constraints dc,
                (select dcc.owner,
                        dcc.constraint_name,
                        dcc.table_name,
                        listagg(column_name, ',') within group(order by dcc.position) as collist
                   from dba_cons_columns dcc
                  group by dcc.owner, dcc.constraint_name, dcc.table_name) cons_col,
                sys.my_constraints dc1
         where dc.constraint_type in ('R')
           and dc.status = 'ENABLED'
           and dc.owner = cons_col.owner
           and dc.table_name = cons_col.table_name
           and dc.constraint_name = cons_col.constraint_name
           and dc.r_owner = dc1.owner
           and dc.r_constraint_name = dc1.constraint_name
           and dc.owner || '.' || dc.table_name = upper(v_tablist(loop_idx))
        union all
        --unique约束
        select 'alter table ' || dc.owner || '.' || dc.table_name ||
                ' add constraint ' || dc.constraint_name || ' unique(' ||
                collist || ');'
          from sys.my_constraints dc,
                (select dcc.owner,
                        dcc.constraint_name,
                        dcc.table_name,
                        listagg(column_name, ',') within group(order by dcc.position) as collist
                   from dba_cons_columns dcc
                  group by dcc.owner, dcc.constraint_name, dcc.table_name) cons_col
         where dc.constraint_type in ('U')
           and dc.status = 'ENABLED'
           and dc.owner = cons_col.owner
           and dc.table_name = cons_col.table_name
           and dc.constraint_name = cons_col.constraint_name
           and dc.owner || '.' || dc.table_name = upper(v_tablist(loop_idx))
        union all
        --primary约束
        select 'alter table ' || dc.owner || '.' || dc.table_name ||
                ' add constraint ' || dc.constraint_name || ' primary key(' ||
                collist || ');'
          from sys.my_constraints dc,
                (select dcc.owner,
                        dcc.constraint_name,
                        dcc.table_name,
                        listagg(column_name, ',') within group(order by dcc.position) as collist
                   from dba_cons_columns dcc
                  group by dcc.owner, dcc.constraint_name, dcc.table_name) cons_col
         where dc.constraint_type in ('P')
           and dc.status = 'ENABLED'
           and dc.owner = cons_col.owner
           and dc.table_name = cons_col.table_name
           and dc.owner || '.' || dc.table_name = upper(v_tablist(loop_idx))
           and dc.constraint_name = cons_col.constraint_name)
      select * bulk collect into v_conslist from t;

      for loop_idx in 1 .. v_conslist.count loop
        dbms_output.put_line(v_conslist(loop_idx));
      end loop;
    end loop;

  end convert_constraint_ddl;
end metadata_ora2gp;
/