COPY를 이용한 데이터 적재

oracledb.defaults.fetch_lobs = False      # CLOB/BLOB 을 str/bytes 로 바로 받는다
oracledb.defaults.fetch_decimals = True   # NUMBER 를 Decimal 로 받아 소수점 손실 방지

def copy_table(ora, pg, owner, t, target, batch=5000, progress=None):
    cols = [c.name for c in t.columns]
    cur = ora.cursor()
    cur.arraysize = batch
    cur.prefetchrows = batch + 1
    cur.execute(f'select {", ".join(chr(34) + c + chr(34) for c in cols)} from "{owner}"."{t.name}"')
...
    with pg.transaction():
        with pg.cursor().copy(copy_sql) as cp:
            while True:
                rows = cur.fetchmany()
                if not rows:
                    break
                for r in rows:
                    cp.write_row([_clean(v) for v in r])
                done += len(rows)
                if progress:
                    progress(done, time.time() - t0)

mkdir -p '<WORK_DIR>'
cd '<WORK_DIR>'

SELECT MAX(DBMS_LOB.GETLENGTH(memo)) AS max_memo_chars FROM orders;
SHOW LONG

set markup csv on quote on
set feedback off termout off
alter session set nls_date_format = 'YYYY-MM-DD HH24:MI:SS';
spool <WORK_DIR>/cust.csv
select cust_id, cust_nm, gender, tel, join_dt, grade_cd from cust order by cust_id;
spool off
exit

set markup csv on quote on
set feedback off termout off
set long <LOB_OUTPUT_BYTES>
alter session set nls_date_format = 'YYYY-MM-DD HH24:MI:SS';
spool <WORK_DIR>/orders.csv
select order_id, cust_id, order_dt, amount, status, memo from orders order by order_id;
spool off
exit

time sqlplus -s <SOURCE_SCHEMA>/<PASSWORD>@//<HOST>:1521/<SERVICE_NAME> @exp_cust.sql
time sqlplus -s <SOURCE_SCHEMA>/<PASSWORD>@//<HOST>:1521/<SERVICE_NAME> @exp_orders.sql

head -n 2 cust.csv orders.csv
wc -l cust.csv orders.csv

sed -i '1{/^\r\?$/d;}' cust.csv orders.csv
export PGPASSWORD=<PASSWORD>
time /usr/pgsql-15/bin/psql -h <HOST> -U <USERNAME> -d <TARGET_DB_NAME> -v ON_ERROR_STOP=1 \
  -c "\copy <SCHEMA>.cust from '<WORK_DIR>/cust.csv' with (format csv, header true)"
time /usr/pgsql-15/bin/psql -h <HOST> -U <USERNAME> -d <TARGET_DB_NAME> -v ON_ERROR_STOP=1 \
  -c "\copy <SCHEMA>.orders from '<WORK_DIR>/orders.csv' with (format csv, header true)"

\set ON_ERROR_STOP on
alter table <SCHEMA>.cust   add constraint cust_pk   primary key (cust_id);
alter table <SCHEMA>.orders add constraint orders_pk primary key (order_id);
alter table <SCHEMA>.orders add constraint orders_cust_fk foreign key (cust_id) references <SCHEMA>.cust (cust_id);
create sequence <SCHEMA>.orders_seq;
select setval('<SCHEMA>.orders_seq', (select max(order_id) from <SCHEMA>.orders));
create view <SCHEMA>.v_cust_sales as
select c.cust_id, c.cust_nm, coalesce(sum(o.amount), 0) as total_amt
  from <SCHEMA>.cust c left join <SCHEMA>.orders o on c.cust_id = o.cust_id
 group by c.cust_id, c.cust_nm;
create procedure <SCHEMA>.p_upgrade_grade(p_min_amt numeric)
language plpgsql as $$
begin
  update <SCHEMA>.cust c set grade_cd = '01'
   where c.cust_id in (select o.cust_id from <SCHEMA>.orders o
                        group by o.cust_id having sum(o.amount) >= p_min_amt);
  commit;
end;
$$;
analyze <SCHEMA>.cust;
analyze <SCHEMA>.orders;

set heading off feedback off pages 0 tab off lines 200
alter session set nls_date_format = 'YYYY-MM-DD HH24:MI:SS';
select 'cust      '||count(*)||' | '||sum(cust_id)||' | '||min(join_dt)||' ~ '||max(join_dt) from cust;
select 'orders    '||count(*)||' | '||to_char(sum(amount),'FM999999999999.00')||' | '||count(case when memo is not null then 1 end)||' | '||min(order_dt)||' ~ '||max(order_dt) from orders;
select 'grade     '||listagg(grade_cd||'='||cnt, ', ') within group (order by grade_cd) from (select grade_cd, count(*) cnt from cust group by grade_cd);
select 'view      '||count(*)||' | '||to_char(sum(total_amt),'FM999999999999.00') from v_cust_sales;
exit

\pset tuples_only on
\pset format unaligned
select 'cust      '||count(*)||' | '||sum(cust_id)||' | '||min(join_dt)||' ~ '||max(join_dt) from <SCHEMA>.cust;
select 'orders    '||count(*)||' | '||sum(amount)||' | '||count(memo)||' | '||min(order_dt)||' ~ '||max(order_dt) from <SCHEMA>.orders;
select 'grade     '||string_agg(grade_cd||'='||cnt, ', ' order by grade_cd) from (select grade_cd, count(*) cnt from <SCHEMA>.cust group by grade_cd) t;
select 'view      '||count(*)||' | '||sum(total_amt) from <SCHEMA>.v_cust_sales;

from itertools import zip_longest

with ora.cursor() as src, pg.cursor(name="memo_check") as dst:
    src.execute("select order_id, memo, dbms_lob.getlength(memo) from orders order by order_id")
    dst.execute("select order_id, memo, char_length(memo) from <SCHEMA>.orders order by order_id")
    missing = object()
    for left, right in zip_longest(src, dst, fillvalue=missing):
        if left is missing or right is missing:
            raise ValueError("행 수 불일치")
        order_id, memo, chars = left
        if hasattr(memo, "read"):
            memo = memo.read()
        if (order_id, memo, chars) != right:
            raise ValueError(f"메모 길이 또는 내용 불일치: order_id={order_id}")