oracle转greenplum功能实现

sky

项目中遇到的，对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;
    /