Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés. # Migración de tablas externas de Oracle a Amazon Aurora compatible con PostgreSQL *Anuradha Chintha y Rakesh Raghav, Amazon Web Services* ## Resumen Las tablas externas permiten a Oracle consultar los datos almacenados fuera de la base de datos en archivos planos. Puede usar el controlador ORACLE\$1LOADER para acceder a cualquier dato almacenado en cualquier formato que pueda cargar la utilidad SQL\$1Loader. No puede usar el Lenguaje de Manipulación de Datos (DML) en tablas externas, pero puede usar las tablas externas para operaciones de consulta, unión y clasificación. Amazon Aurora compatible con PostgreSQL no proporciona una funcionalidad similar a las tablas externas de Oracle. En su lugar, debe adoptar la modernización para desarrollar una solución escalable que cumpla con los requisitos funcionales y sea eficiente. Este patrón proporciona los pasos para migrar diferentes tipos de tablas externas de Oracle a la edición de Aurora compatible con PostgreSQL en la nube de Amazon Web Services (AWS) mediante la extensión `aws_s3`. Recomendamos probar exhaustivamente esta solución antes de implementarla en un entorno de producción. ## Requisitos previos y limitaciones **Requisitos previos ** + Una cuenta de AWS activa + Interfaz de la línea de comandos de AWS (AWS CLI) + Una instancia disponible de base de datos Aurora compatible con PostgreSQL. + Una base de datos de Oracle en las instalaciones con una tabla externa + API de pg.Client + Archivos de datos  **Limitaciones** + Este patrón no proporciona la funcionalidad necesaria para sustituir a las tablas externas de Oracle. Sin embargo, los pasos y el código de muestra se pueden mejorar aún más para lograr sus objetivos de modernización de la base de datos. + Los archivos no deben contener el carácter que se emplea como delimitador en las funciones de exportación e importación de `aws_s3`. **Versiones de producto** + Para realizar la importación de Amazon S3 en RDS para PostgreSQL, la base de datos debe ejecutar la versión PostgreSQL 10.7 o posterior. ## Arquitectura **Pila de tecnología de origen** + Oracle **Arquitectura de origen ** ![\[Diagrama de los archivos de datos que van a un directorio y una tabla de la base de datos de Oracle en las instalaciones.\]](http://docs.aws.amazon.com/es_es/prescriptive-guidance/latest/patterns/images/pattern-img/555e69af-36fc-4ff5-b66c-af22b4cf262a/images/3fbc507d-b0fa-4e05-b999-043dc7327ed7.png) **Pila de tecnología de destino** + Amazon Aurora compatible con PostgreSQL + Amazon CloudWatch + AWS Lambda + AWS Secrets Manager + Amazon Simple Notification Service (Amazon SNS) + Amazon Simple Storage Service (Amazon S3) **Arquitectura de destino** En el siguiente diagrama se muestra una representación de alto nivel de la solución. ![\[La descripción se encuentra después del diagrama.\]](http://docs.aws.amazon.com/es_es/prescriptive-guidance/latest/patterns/images/pattern-img/555e69af-36fc-4ff5-b66c-af22b4cf262a/images/5421540e-d2e3-4361-89cc-d8415fcb21fd.png) 1. Los archivos se cargan en el bucket de S3. 1. Se inicia la función de Lambda. 1. La función de Lambda inicia la llamada a la función de base de datos. 1. Secrets Manager proporciona las credenciales para acceder a la base de datos. 1. Según la función de la base de datos, se crea una alarma de SNS. **Automatizar y escalar** Cualquier adición o cambio en las tablas externas se puede gestionar mediante el mantenimiento de los metadatos. ## Tools (Herramientas) + [Amazon Aurora compatible con PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_AuroraOverview.html): la edición de Amazon Aurora compatible con PostgreSQL es un motor de bases de datos relacionales, completamente administrado, compatible con PostgreSQL y conforme a ACID, que combina la velocidad y la fiabilidad de las bases de datos comerciales de tecnología avanzada con la rentabilidad de las bases de datos de código abierto. + [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-welcome.html): la interfaz de la línea de comandos de AWS (AWS CLI) es una herramienta unificada para administrar los servicios de AWS. Con una única herramienta para descargar y configurar, puede controlar varios servicios de AWS desde la línea de comando y automatizarlos mediante scripts. + [Amazon CloudWatch](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/WhatIsCloudWatch.html): Amazon CloudWatch supervisa los recursos y la utilización de Amazon S3. + [AWS Lambda](https://docs.aws.amazon.com/lambda/latest/dg/welcome.html): AWS Lambda es un servicio de computación sin servidor que permite ejecutar código sin aprovisionar ni administrar servidores, crear una lógica de escalado de clústeres adaptada a las cargas de trabajo, mantener las integraciones de eventos o gestionar los tiempos de ejecución. En este patrón, Lambda ejecuta la función de base de datos cada vez que se carga un archivo en Amazon S3. + [AWS Secrets Manager](https://docs.aws.amazon.com/secretsmanager/latest/userguide/intro.html): AWS Secrets Manager es un servicio de almacenamiento y recuperación de credenciales. Con Secrets Manager puede reemplazar las credenciales codificadas en el código, incluidas las contraseñas, con una llamada a la API de Secrets Manager para recuperar el secreto mediante programación. + [Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/Welcome.html): Amazon Simple Storage Service (Amazon S3) proporciona una capa de almacenamiento que permite recibir y almacenar archivos para su consumo y transmisión hacia y desde el clúster de Aurora compatible con PostgreSQL. + [aws\$1s3](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/PostgreSQL.Procedural.Importing.html#aws_s3.table_import_from_s3): la extensión `aws_s3` integra Amazon S3 y Aurora compatible con PostgreSQL. + [Amazon SNS](https://docs.aws.amazon.com/sns/latest/dg/welcome.html): Amazon Simple Notification Service (Amazon SNS) coordina y administra la entrega o el envío de mensajes entre publicadores y clientes. En este patrón, Amazon SNS se usa para enviar notificaciones. **Código** Siempre que se ubique un archivo en el bucket de S3, se debe crear una función de base de datos a la que llamar desde la aplicación de procesamiento o la función de Lambda. Para obtener más información, consulte el código (adjunto). ## Epics ### Cree un archivo externo | Tarea | Descripción | Habilidades requeridas | | --- | --- | --- | | Añada un archivo externo a la base de datos de origen. | Cree un archivo externo y trasládelo al directorio `oracle`. | Administrador de base de datos | ### Configure el objetivo (Aurora compatible con PostgreSQL) | Tarea | Descripción | Habilidades requeridas | | --- | --- | --- | | Cree una base de datos de Aurora PostgreSQL. | Cree una instancia de base de datos en su clúster de Amazon Aurora compatible con PostgreSQL. | Administrador de base de datos | | Cree un esquema, una extensión aws\$1s3 y tablas. | Use el código que aparece en la sección `ext_tbl_scripts` de *Información adicional*. Las tablas incluyen tablas reales, tablas de ensayo, tablas de errores y registros y una metatabla. | Administrador de base de datos, desarrollador | | Cree la función de base de datos. | Para crear la función de base de datos, use el código que aparece bajo la función `load_external_table_latest` en la sección de *Información adicional*. | Administrador de base de datos, desarrollador | ### Creación y configuración de la función de Lambda | Tarea | Descripción | Habilidades requeridas | | --- | --- | --- | | Crear un rol. | Cree un rol con permisos para acceder a Amazon S3 y a Amazon Relational Database Service (Amazon RDS). Esta función se asignará a Lambda para ejecutar el patrón. | Administrador de base de datos | | Crear la función de Lambda. | Cree una función de Lambda que lea el nombre del archivo de Amazon S3 (por ejemplo, `file_key = info.get('object', {}).get('key')`) y llame a la función de base de datos (por ejemplo, `curs.callproc("load_external_tables", [file_key])`) con el nombre del archivo como parámetro de entrada.Según el resultado de la llamada a la función, se iniciará una notificación de SNS (por ejemplo, `client.publish(TopicArn='arn:',Message='fileloadsuccess',Subject='fileloadsuccess')`).En función de las necesidades de su empresa, puede crear una función de Lambda con código adicional si es necesario. Para más información, consulte la [documentación de Lambda](https://docs.aws.amazon.com/lambda/latest/dg/welcome.html). | Administrador de base de datos | | Configure un desencadenante de eventos en el bucket de S3. | Configure un mecanismo para llamar a la función de Lambda en todos los eventos de creación de objetos en el bucket de S3. | Administrador de base de datos | | Cree un secreto. | Cree un nombre secreto para las credenciales de la base de datos mediante Secrets Manager. Pase el secreto a la función de Lambda. | Administrador de base de datos | | Cargue los archivos de soporte de Lambda. | Cargue un archivo .zip que contenga los paquetes de soporte de Lambda y el script de Python adjunto para conectar a Aurora compatible con PostgreSQL. El código Python llamará a la función que creó en la base de datos. | Administrador de base de datos | | Cree un tema de SNS. | Cree un tema de SNS para enviar un correo si la carga de datos se ha realizado correctamente o no. | Administrador de base de datos | ### Cómo añadir integración con Amazon S3 | Tarea | Descripción | Habilidades requeridas | | --- | --- | --- | | Cree un bucket de S3. | En la consola de Amazon S3, cree un bucket de S3 con un nombre único que no contenga barras diagonales en el inicio. Un nombre de bucket de S3 es globalmente único y todas las cuentas de AWS comparten el espacio de nombres. | Administrador de base de datos | | Cree políticas de IAM . | Para crear las políticas de AWS Identity and Access Management (IAM), use el código que se describe en la sección `s3bucketpolicy_for_import` de *Información adicional*. | Administrador de base de datos | | Cree roles. | Cree dos roles para Aurora compatible con PostgreSQL, uno para Importar y otro para Exportar. Asigne las políticas correspondientes a los roles. | Administrador de base de datos | | Adjunte los roles al clúster de Aurora compatible con PostgreSQL. | En **Administrar roles**, adjunte los roles de Importar y Exportar al clúster de Aurora PostgreSQL. | Administrador de base de datos | | Cree objetos de apoyo para Aurora compatible con PostgreSQL. | Para las tablas de scripts, utilice el código de `ext_tbl_scripts` en la sección *Información adicional*.Para la función personalizada, utilice el código de `load_external_Table_latest` en la sección *Información adicional*. | Administrador de base de datos | ### Procese un archivo de prueba | Tarea | Descripción | Habilidades requeridas | | --- | --- | --- | | Cargar un archivo en el bucket de S3. | Para cargar un archivo de prueba en el bucket de S3, use la consola o ejecute el siguiente comando en la CLI de AWS. 
aws s3 cp /Users/Desktop/ukpost/exttbl/"testing files"/aps s3://s3importtest/inputext/aps
En cuanto se carga el archivo, el evento de bucket inicia la función de Lambda, que ejecuta la función Aurora compatible con PostgreSQL. | Administrador de base de datos | | Compruebe los datos y los archivos de registro y error. | La función de Aurora compatible con PostgreSQL carga los archivos en la tabla principal y crea los archivos `.log` y `.bad` en el bucket de S3. | Administrador de base de datos | | Supervise la solución. | En la CloudWatch consola de Amazon, supervise la función Lambda. | Administrador de base de datos | ## Recursos relacionados + [Integración de Amazon S3](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/oracle-s3-integration.html) + [Amazon S3](https://aws.amazon.com/s3/) + [Trabajar con la versión de Amazon Aurora compatible con PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Aurora.AuroraPostgreSQL.html) + [AWS Lambda](https://aws.amazon.com/lambda/) + [Amazon CloudWatch](https://aws.amazon.com/cloudwatch/) + [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/) + [Configuración de notificaciones de Amazon SNS](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/US_SetupSNS.html) ## Información adicional **ext\$1table\$1scripts** ``` CREATE EXTENSION aws_s3 CASCADE; CREATE TABLE IF NOT EXISTS meta_EXTERNAL_TABLE ( table_name_stg character varying(100) , table_name character varying(100) , col_list character varying(1000) , data_type character varying(100) , col_order numeric, start_pos numeric, end_pos numeric, no_position character varying(100) , date_mask character varying(100) , delimeter character(1) , directory character varying(100) , file_name character varying(100) , header_exist character varying(5) ); CREATE TABLE IF NOT EXISTS ext_tbl_stg ( col1 text ); CREATE TABLE IF NOT EXISTS error_table ( error_details text, file_name character varying(100), processed_time timestamp without time zone ); CREATE TABLE IF NOT EXISTS log_table ( file_name character varying(50) COLLATE pg_catalog."default", processed_date timestamp without time zone, tot_rec_count numeric, proc_rec_count numeric, error_rec_count numeric ); sample insert scripts of meta data: INSERT INTO meta_EXTERNAL_TABLE (table_name_stg, table_name, col_list, data_type, col_order, start_pos, end_pos, no_position, date_mask, delimeter, directory, file_name, header_exist) VALUES ('F_EX_APS_TRANSACTIONS_STG', 'F_EX_APS_TRANSACTIONS', 'source_filename', 'character varying', 2, 8, 27, NULL, NULL, NULL, 'databasedev', 'externalinterface/loaddir/APS', 'NO'); INSERT INTO meta_EXTERNAL_TABLE (table_name_stg, table_name, col_list, data_type, col_order, start_pos, end_pos, no_position, date_mask, delimeter, directory, file_name, header_exist) VALUES ('F_EX_APS_TRANSACTIONS_STG', 'F_EX_APS_TRANSACTIONS', 'record_type_identifier', 'character varying', 3, 28, 30, NULL, NULL, NULL, 'databasedev', 'externalinterface/loaddir/APS', 'NO'); INSERT INTO meta_EXTERNAL_TABLE (table_name_stg, table_name, col_list, data_type, col_order, start_pos, end_pos, no_position, date_mask, delimeter, directory, file_name, header_exist) VALUES ('F_EX_APS_TRANSACTIONS_STG', 'F_EX_APS_TRANSACTIONS', 'fad_code', 'numeric', 4, 31, 36, NULL, NULL, NULL, 'databasedev', 'externalinterface/loaddir/APS', 'NO'); INSERT INTO meta_EXTERNAL_TABLE (table_name_stg, table_name, col_list, data_type, col_order, start_pos, end_pos, no_position, date_mask, delimeter, directory, file_name, header_exist) VALUES ('F_EX_APS_TRANSACTIONS_STG', 'F_EX_APS_TRANSACTIONS', 'session_sequence_number', 'numeric', 5, 37, 42, NULL, NULL, NULL, 'databasedev', 'externalinterface/loaddir/APS', 'NO'); INSERT INTO meta_EXTERNAL_TABLE (table_name_stg, table_name, col_list, data_type, col_order, start_pos, end_pos, no_position, date_mask, delimeter, directory, file_name, header_exist) VALUES ('F_EX_APS_TRANSACTIONS_STG', 'F_EX_APS_TRANSACTIONS', 'transaction_sequence_number', 'numeric', 6, 43, 48, NULL, NULL, NULL, 'databasedev', 'externalinterface/loaddir/APS', 'NO'); ``` **s3bucketpolicy\$1for import** ``` ---Import role policy --Create an IAM policy to allow, Get, and list actions on S3 bucket { "Version": "2012-10-17", "Statement": [ { "Sid": "s3import", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Effect": "Allow", "Resource": [ "arn:aws:s3:::s3importtest", "arn:aws:s3:::s3importtest/*" ] } ] } --Export Role policy --Create an IAM policy to allow, put, and list actions on S3 bucket { "Version": "2012-10-17", "Statement": [ { "Sid": "s3export", "Action": [ "S3:PutObject", "s3:ListBucket" ], "Effect": "Allow", "Resource": [ "arn:aws:s3:::s3importtest/*" ] } ] } ``` **Ejemplo de función de base de datos load\$1external\$1tables\$1latest** ``` CREATE OR REPLACE FUNCTION public.load_external_tables(pi_filename text) RETURNS character varying LANGUAGE plpgsql AS $function$ /* Loading data from S3 bucket into a APG table */ DECLARE v_final_sql TEXT; pi_ext_table TEXT; r refCURSOR; v_sqlerrm text; v_chunk numeric; i integer; v_col_list TEXT; v_postion_list CHARACTER VARYING(1000); v_len integer; v_delim varchar; v_file_name CHARACTER VARYING(1000); v_directory CHARACTER VARYING(1000); v_table_name_stg CHARACTER VARYING(1000); v_sql_col TEXT; v_sql TEXT; v_sql1 TEXT; v_sql2 TEXT; v_sql3 TEXT; v_cnt integer; v_sql_dynamic TEXT; v_sql_ins TEXT; proc_rec_COUNT integer; error_rec_COUNT integer; tot_rec_COUNT integer; v_rec_val integer; rec record; v_col_cnt integer; kv record; v_val text; v_header text; j integer; ERCODE VARCHAR(5); v_region text; cr CURSOR FOR SELECT distinct DELIMETER, FILE_NAME, DIRECTORY FROM meta_EXTERNAL_TABLE WHERE table_name = pi_ext_table AND DELIMETER IS NOT NULL; cr1 CURSOR FOR SELECT col_list, data_type, start_pos, END_pos, concat_ws('',' ',TABLE_NAME_STG) as TABLE_NAME_STG, no_position,date_mask FROM meta_EXTERNAL_TABLE WHERE table_name = pi_ext_table order by col_order asc; cr2 cursor FOR SELECT distinct table_name,table_name_stg FROM meta_EXTERNAL_TABLE WHERE upper(file_name) = upper(pi_filename); BEGIN -- PERFORM utl_file_utility.init(); v_region := 'us-east-1'; /* find tab details from file name */ --DELETE FROM ERROR_TABLE WHERE file_name= pi_filename; -- DELETE FROM log_table WHERE file_name= pi_filename; BEGIN SELECT distinct table_name,table_name_stg INTO strict pi_ext_table,v_table_name_stg FROM meta_EXTERNAL_TABLE WHERE upper(file_name) = upper(pi_filename); EXCEPTION WHEN NO_DATA_FOUND THEN raise notice 'error 1,%',sqlerrm; pi_ext_table := null; v_table_name_stg := null; RAISE USING errcode = 'NTFIP' ; when others then raise notice 'error others,%',sqlerrm; END; j :=1 ; for rec in cr2 LOOP pi_ext_table := rec.table_name; v_table_name_stg := rec.table_name_stg; v_col_list := null; IF pi_ext_table IS NOT NULL THEN --EXECUTE concat_ws('','truncate table ' ,pi_ext_table) ; EXECUTE concat_ws('','truncate table ' ,v_table_name_stg) ; SELECT distinct DELIMETER INTO STRICT v_delim FROM meta_EXTERNAL_TABLE WHERE table_name = pi_ext_table; IF v_delim IS NOT NULL THEN SELECT distinct DELIMETER, FILE_NAME, DIRECTORY , concat_ws('',' ',table_name_stg), case header_exist when 'YES' then 'CSV HEADER' else 'CSV' end as header_exist INTO STRICT v_delim,v_file_name,v_directory,v_table_name_stg,v_header FROM meta_EXTERNAL_TABLE WHERE table_name = pi_ext_table AND DELIMETER IS NOT NULL; IF upper(v_delim) = 'CSV' THEN v_sql := concat_ws('','SELECT aws_s3.table_import_FROM_s3 ( ''', v_table_name_stg,''','''', ''DELIMITER '''','''' CSV HEADER QUOTE ''''"'''''', aws_commons.create_s3_uri ( ''', v_directory,''',''',v_file_name,''', ''',v_region,'''))'); ELSE v_sql := concat_ws('','SELECT aws_s3.table_import_FROM_s3(''', v_table_name_stg, ''','''', ''DELIMITER AS ''''^''''',''',',' aws_commons.create_s3_uri ( ''',v_directory, ''',''', v_file_name, ''',', '''',v_region,''') )'); raise notice 'v_sql , %',v_sql; begin EXECUTE v_sql; EXCEPTION WHEN OTHERS THEN raise notice 'error 1'; RAISE USING errcode = 'S3IMP' ; END; select count(col_list) INTO v_col_cnt from meta_EXTERNAL_TABLE where table_name = pi_ext_table; -- raise notice 'v_sql 2, %',concat_ws('','update ',v_table_name_stg, ' set col1 = col1||''',v_delim,''''); execute concat_ws('','update ',v_table_name_stg, ' set col1 = col1||''',v_delim,''''); i :=1; FOR rec in cr1 loop v_sql1 := concat_ws('',v_sql1,'split_part(col1,''',v_delim,''',', i,')',' as ',rec.col_list,','); v_sql2 := concat_ws('',v_sql2,rec.col_list,','); -- v_sql3 := concat_ws('',v_sql3,'rec.',rec.col_list,'::',rec.data_type,','); case WHEN upper(rec.data_type) = 'NUMERIC' THEN v_sql3 := concat_ws('',v_sql3,' case WHEN length(trim(split_part(col1,''',v_delim,''',', i,'))) =0 THEN null ELSE coalesce((trim(split_part(col1,''',v_delim,''',', i,')))::NUMERIC,0)::',rec.data_type,' END as ',rec.col_list,',') ; WHEN UPPER(rec.data_type) = 'TIMESTAMP WITHOUT TIME ZONE' AND rec.date_mask = 'YYYYMMDD' THEN v_sql3 := concat_ws('',v_sql3,' case WHEN length(trim(split_part(col1,''',v_delim,''',', i,'))) =0 THEN null ELSE to_date(coalesce((trim(split_part(col1,''',v_delim,''',', i,'))),''99990101''),''YYYYMMDD'')::',rec.data_type,' END as ',rec.col_list,','); WHEN UPPER(rec.data_type) = 'TIMESTAMP WITHOUT TIME ZONE' AND rec.date_mask = 'MM/DD/YYYY hh24:mi:ss' THEN v_sql3 := concat_ws('',v_sql3,' case WHEN length(trim(split_part(col1,''',v_delim,''',', i,'))) =0 THEN null ELSE to_date(coalesce((trim(split_part(col1,''',v_delim,''',', i,'))),''01/01/9999 0024:00:00''),''MM/DD/YYYY hh24:mi:ss'')::',rec.data_type,' END as ',rec.col_list,','); ELSE v_sql3 := concat_ws('',v_sql3,' case WHEN length(trim(split_part(col1,''',v_delim,''',', i,'))) =0 THEN null ELSE coalesce((trim(split_part(col1,''',v_delim,''',', i,'))),'''')::',rec.data_type,' END as ',rec.col_list,',') ; END case; i :=i+1; end loop; -- raise notice 'v_sql 3, %',v_sql3; SELECT trim(trailing ' ' FROM v_sql1) INTO v_sql1; SELECT trim(trailing ',' FROM v_sql1) INTO v_sql1; SELECT trim(trailing ' ' FROM v_sql2) INTO v_sql2; SELECT trim(trailing ',' FROM v_sql2) INTO v_sql2; SELECT trim(trailing ' ' FROM v_sql3) INTO v_sql3; SELECT trim(trailing ',' FROM v_sql3) INTO v_sql3; END IF; raise notice 'v_delim , %',v_delim; EXECUTE concat_ws('','SELECT COUNT(*) FROM ',v_table_name_stg) INTO v_cnt; raise notice 'stg cnt , %',v_cnt; /* if upper(v_delim) = 'CSV' then v_sql_ins := concat_ws('', ' SELECT * from ' ,v_table_name_stg ); else -- v_sql_ins := concat_ws('',' SELECT ',v_sql1,' from (select col1 from ' ,v_table_name_stg , ')sub '); v_sql_ins := concat_ws('',' SELECT ',v_sql3,' from (select col1 from ' ,v_table_name_stg , ')sub '); END IF;*/ v_chunk := v_cnt/100; for i in 1..101 loop BEGIN -- raise notice 'v_sql , %',v_sql; -- raise notice 'Chunk number , %',i; v_sql_ins := concat_ws('',' SELECT ',v_sql3,' from (select col1 from ' ,v_table_name_stg , ' offset ',v_chunk*(i-1), ' limit ',v_chunk,') sub '); v_sql := concat_ws('','insert into ', pi_ext_table ,' ', v_sql_ins); -- raise notice 'select statement , %',v_sql_ins; -- v_sql := null; -- EXECUTE concat_ws('','insert into ', pi_ext_table ,' ', v_sql_ins, 'offset ',v_chunk*(i-1), ' limit ',v_chunk ); --v_sql := concat_ws('','insert into ', pi_ext_table ,' ', v_sql_ins ); -- raise notice 'insert statement , %',v_sql; raise NOTICE 'CHUNK START %',v_chunk*(i-1); raise NOTICE 'CHUNK END %',v_chunk; EXECUTE v_sql; EXCEPTION WHEN OTHERS THEN -- v_sql_ins := concat_ws('',' SELECT ',v_sql1, ' from (select col1 from ' ,v_table_name_stg , ' )sub '); -- raise notice 'Chunk number for cursor , %',i; raise NOTICE 'Cursor - CHUNK START %',v_chunk*(i-1); raise NOTICE 'Cursor - CHUNK END %',v_chunk; v_sql_ins := concat_ws('',' SELECT ',v_sql3, ' from (select col1 from ' ,v_table_name_stg , ' )sub '); v_final_sql := REPLACE (v_sql_ins, ''''::text, ''''''::text); -- raise notice 'v_final_sql %',v_final_sql; v_sql :=concat_ws('','do $a$ declare r refcursor;v_sql text; i numeric;v_conname text; v_typ ',pi_ext_table,'[]; v_rec ','record','; begin open r for execute ''select col1 from ',v_table_name_stg ,' offset ',v_chunk*(i-1), ' limit ',v_chunk,'''; loop begin fetch r into v_rec; EXIT WHEN NOT FOUND; v_sql := concat_ws('''',''insert into ',pi_ext_table,' SELECT ',REPLACE (v_sql3, ''''::text, ''''''::text) , ' from ( select '''''',v_rec.col1,'''''' as col1) v''); execute v_sql; exception when others then v_sql := ''INSERT INTO ERROR_TABLE VALUES (concat_ws('''''''',''''Error Name: '''',$$''||SQLERRM||''$$,''''Error State: '''',''''''||SQLSTATE||'''''',''''record : '''',$$''||v_rec.col1||''$$),'''''||pi_filename||''''',now())''; execute v_sql; continue; end ; end loop; close r; exception when others then raise; end ; $a$'); -- raise notice ' inside excp v_sql %',v_sql; execute v_sql; -- raise notice 'v_sql %',v_sql; END; END LOOP; ELSE SELECT distinct DELIMETER,FILE_NAME,DIRECTORY ,concat_ws('',' ',table_name_stg), case header_exist when 'YES' then 'CSV HEADER' else 'CSV' end as header_exist INTO STRICT v_delim,v_file_name,v_directory,v_table_name_stg,v_header FROM meta_EXTERNAL_TABLE WHERE table_name = pi_ext_table ; v_sql := concat_ws('','SELECT aws_s3.table_import_FROM_s3(''', v_table_name_stg, ''','''', ''DELIMITER AS ''''#'''' ',v_header,' '',',' aws_commons.create_s3_uri ( ''',v_directory, ''',''', v_file_name, ''',', '''',v_region,''') )'); EXECUTE v_sql; FOR rec in cr1 LOOP IF rec.start_pos IS NULL AND rec.END_pos IS NULL AND rec.no_position = 'recnum' THEN v_rec_val := 1; ELSE case WHEN upper(rec.data_type) = 'NUMERIC' THEN v_sql1 := concat_ws('',' case WHEN length(trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))) =0 THEN null ELSE coalesce((trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1)))::NUMERIC,0)::',rec.data_type,' END as ',rec.col_list,',') ; WHEN UPPER(rec.data_type) = 'TIMESTAMP WITHOUT TIME ZONE' AND rec.date_mask = 'YYYYMMDD' THEN v_sql1 := concat_ws('','case WHEN length(trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))) =0 THEN null ELSE to_date(coalesce((trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))),''99990101''),''YYYYMMDD'')::',rec.data_type,' END as ',rec.col_list,','); WHEN UPPER(rec.data_type) = 'TIMESTAMP WITHOUT TIME ZONE' AND rec.date_mask = 'YYYYMMDDHH24MISS' THEN v_sql1 := concat_ws('','case WHEN length(trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))) =0 THEN null ELSE to_date(coalesce((trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))),''9999010100240000''),''YYYYMMDDHH24MISS'')::',rec.data_type,' END as ',rec.col_list,','); ELSE v_sql1 := concat_ws('',' case WHEN length(trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))) =0 THEN null ELSE coalesce((trim(substring(COL1, ',rec.start_pos ,',', rec.END_pos,'-',rec.start_pos ,'+1))),'''')::',rec.data_type,' END as ',rec.col_list,',') ; END case; END IF; v_col_list := concat_ws('',v_col_list ,v_sql1); END LOOP; SELECT trim(trailing ' ' FROM v_col_list) INTO v_col_list; SELECT trim(trailing ',' FROM v_col_list) INTO v_col_list; v_sql_col := concat_ws('',trim(trailing ',' FROM v_col_list) , ' FROM ',v_table_name_stg,' WHERE col1 IS NOT NULL AND length(col1)>0 '); v_sql_dynamic := v_sql_col; EXECUTE concat_ws('','SELECT COUNT(*) FROM ',v_table_name_stg) INTO v_cnt; IF v_rec_val = 1 THEN v_sql_ins := concat_ws('',' select row_number() over(order by ctid) as line_number ,' ,v_sql_dynamic) ; ELSE v_sql_ins := concat_ws('',' SELECT' ,v_sql_dynamic) ; END IF; BEGIN EXECUTE concat_ws('','insert into ', pi_ext_table ,' ', v_sql_ins); EXCEPTION WHEN OTHERS THEN IF v_rec_val = 1 THEN v_final_sql := ' select row_number() over(order by ctid) as line_number ,col1 from '; ELSE v_final_sql := ' SELECT col1 from'; END IF; v_sql :=concat_ws('','do $a$ declare r refcursor;v_rec_val numeric := ',coalesce(v_rec_val,0),';line_number numeric; col1 text; v_typ ',pi_ext_table,'[]; v_rec ',pi_ext_table,'; begin open r for execute ''',v_final_sql, ' ',v_table_name_stg,' WHERE col1 IS NOT NULL AND length(col1)>0 '' ; loop begin if v_rec_val = 1 then fetch r into line_number,col1; else fetch r into col1; end if; EXIT WHEN NOT FOUND; if v_rec_val = 1 then select line_number,',trim(trailing ',' FROM v_col_list) ,' into v_rec; else select ',trim(trailing ',' FROM v_col_list) ,' into v_rec; end if; insert into ',pi_ext_table,' select v_rec.*; exception when others then INSERT INTO ERROR_TABLE VALUES (concat_ws('''',''Error Name: '',SQLERRM,''Error State: '',SQLSTATE,''record : '',v_rec),''',pi_filename,''',now()); continue; end ; end loop; close r; exception when others then raise; end ; $a$'); execute v_sql; END; END IF; EXECUTE concat_ws('','SELECT COUNT(*) FROM ' ,pi_ext_table) INTO proc_rec_COUNT; EXECUTE concat_ws('','SELECT COUNT(*) FROM error_table WHERE file_name =''',pi_filename,''' and processed_time::date = clock_timestamp()::date') INTO error_rec_COUNT; EXECUTE concat_ws('','SELECT COUNT(*) FROM ',v_table_name_stg) INTO tot_rec_COUNT; INSERT INTO log_table values(pi_filename,now(),tot_rec_COUNT,proc_rec_COUNT, error_rec_COUNT); raise notice 'v_directory, %',v_directory; raise notice 'pi_filename, %',pi_filename; raise notice 'v_region, %',v_region; perform aws_s3.query_export_to_s3('SELECT replace(trim(substring(error_details,position(''('' in error_details)+1),'')''),'','','';''),file_name,processed_time FROM error_table WHERE file_name = '''||pi_filename||'''', aws_commons.create_s3_uri(v_directory, pi_filename||'.bad', v_region), options :='FORmat csv, header, delimiter $$,$$' ); raise notice 'v_directory, %',v_directory; raise notice 'pi_filename, %',pi_filename; raise notice 'v_region, %',v_region; perform aws_s3.query_export_to_s3('SELECT * FROM log_table WHERE file_name = '''||pi_filename||'''', aws_commons.create_s3_uri(v_directory, pi_filename||'.log', v_region), options :='FORmat csv, header, delimiter $$,$$' ); END IF; j := j+1; END LOOP; RETURN 'OK'; EXCEPTION WHEN OTHERS THEN raise notice 'error %',sqlerrm; ERCODE=SQLSTATE; IF ERCODE = 'NTFIP' THEN v_sqlerrm := concat_Ws('',sqlerrm,'No data for the filename'); ELSIF ERCODE = 'S3IMP' THEN v_sqlerrm := concat_Ws('',sqlerrm,'Error While exporting the file from S3'); ELSE v_sqlerrm := sqlerrm; END IF; select distinct directory into v_directory from meta_EXTERNAL_TABLE; raise notice 'exc v_directory, %',v_directory; raise notice 'exc pi_filename, %',pi_filename; raise notice 'exc v_region, %',v_region; perform aws_s3.query_export_to_s3('SELECT * FROM error_table WHERE file_name = '''||pi_filename||'''', aws_commons.create_s3_uri(v_directory, pi_filename||'.bad', v_region), options :='FORmat csv, header, delimiter $$,$$' ); RETURN null; END; $function$ ```