import { log , generateError } from './FsOps' ; import Conversion from './Conversion' ; import DBAccess from './DBAccess' ; import DBAccessQueryResult from './DBAccessQueryResult' ; import DBVendors from './DBVendors' ; import MessageToMaster from './MessageToMaster' ; import MessageToDataLoader from './MessageToDataLoader' ; import { dataTransferred } from './ConsistencyEnforcer' ; import IDBAccessQueryParams from './IDBAccessQueryParams' ; import * as extraConfigProcessor from './ExtraConfigProcessor' ; import { getDataPoolTableName } from './DataPoolManager' ; import * as path from 'path' ; import { PoolClient , QueryResult } from 'pg' ; import { PoolConnection } from 'mysql' ; const { from } = require ( 'pg-copy-streams' ) ; const { Transform : Json2CsvTransform } = require ( 'json2csv' ) ; process . on ( 'message' , async ( signal ) => { const { config , chunk } = signal ; const conv = new Conversion ( config ) ; log ( conv , `\t--[loadData] Loading the data into "${ conv . _schema }"."${ chunk . _tableName }" table...` ) ; const isRecoveryMode = await dataTransferred ( conv , chunk . _id ) ; if ( ! isRecoveryMode ) { await populateTableWorker ( conv , chunk . _tableName , chunk . _selectFieldList , chunk . _rowsCnt , chunk . _id ) ; return ; } const client = await DBAccess . getPgClient ( conv ) ; return deleteChunk ( conv , chunk . _id , client ) ; } ) ; /**
 * Wraps "process.send" method to avoid "cannot invoke an object which is possibly undefined" warning.
 */ const processSend = ( x ) : void => { if ( process . send ) { process . send ( x ) ; } } ; /**
 * Deletes given record from the data-pool.
 */ const deleteChunk = async ( conversion , dataPoolId , client , originalSessionReplicationRole = null ) : Promise < void > => { const sql = `DELETE FROM ${ getDataPoolTableName ( conversion ) } WHERE id = ${ dataPoolId };` ; try { await client . query ( sql ) ; if ( originalSessionReplicationRole ) { await enableTriggers ( conversion , client , < string > originalSessionReplicationRole ) ; } } catch ( error ) { await generateError ( conversion , `\t--[DataLoader::deleteChunk] ${ error }` , sql ) ; } finally { await DBAccess . releaseDbClient ( conversion , client ) ; } } ; /**
 * Processes data-loading error.
 */ const processDataError = async ( conv , streamError , sql , sqlCopy , tableName , dataPoolId , client , originalSessionReplicationRole ) : Promise < void > => { await generateError ( conv , `\t--[populateTableWorker] ${ streamError }` , sqlCopy ) ; const rejectedData = `\t--[populateTableWorker] Error loading table data:\n${ sql }\n` ; log ( conv , rejectedData , path . join ( conv . _logsDirPath , `${ tableName }.log` ) ) ; await deleteChunk ( conv , dataPoolId , client , originalSessionReplicationRole ) ; processSend ( new MessageToMaster ( tableName , 0 ) ) ; } ; /**
 * Loads a chunk of data using "PostgreSQL COPY".
 */ const populateTableWorker = async ( conv , tableName , strSelectFieldList , rowsCnt , dataPoolId ) : Promise < void > => { const originalTableName = extraConfigProcessor . getTableName ( conv , tableName , true ) ; const sql = `SELECT ${ strSelectFieldList } FROM \`${ originalTableName }\`;` ; const mysqlClient = await DBAccess . getMysqlClient ( conv ) ; const sqlCopy = `COPY "${ conv . _schema }"."${ tableName }" FROM STDIN DELIMITER '${ conv . _delimiter }' CSV;` ; const client = await DBAccess . getPgClient ( conv ) ; let originalSessionReplicationRole = null ; if ( conv . shouldMigrateOnlyData ( ) ) { originalSessionReplicationRole = await disableTriggers ( conv , client ) ; } const copyStream = getCopyStream ( conv , client , sqlCopy , sql , tableName , rowsCnt , dataPoolId , originalSessionReplicationRole ) ; const json2csvStream = await getJson2csvStream ( conv , originalTableName , dataPoolId , client , originalSessionReplicationRole ) ; const mysqlClientErrorHandler = async ( err ) => { await processDataError ( conv , err , sql , sqlCopy , tableName , dataPoolId , client , originalSessionReplicationRole ) ; } ; mysqlClient . query ( sql ) . on ( 'error' , mysqlClientErrorHandler ) . stream ( { highWaterMark : conv . _streamsHighWaterMark } ) . pipe ( json2csvStream ) . pipe ( copyStream ) ; } ; /**
 * Returns new PostgreSQL copy stream object.
 */ const getCopyStream = ( conv , client , sqlCopy , sql , tableName , rowsCnt , dataPoolId , originalSessionReplicationRole ) : any => { const copyStream = client . query ( from ( sqlCopy ) ) ; copyStream . on ( 'finish' , async ( ) => { processSend ( new MessageToMaster ( tableName , rowsCnt ) ) ; await deleteChunk ( conv , dataPoolId , client ) ; } ) . on ( 'error' , async ( copyStreamError ) => { await processDataError ( conv , copyStreamError , sql , sqlCopy , tableName , dataPoolId , client , originalSessionReplicationRole ) ; } ) ; return copyStream ; } ; /**
 * Returns new json-to-csv stream-transform object.
 */ const getJson2csvStream = async ( conversion , originalTableName , dataPoolId , client , originalSessionReplicationRole ) : Promise < any > => { const params = { conversion : conversion , caller : 'DataLoader::populateTableWorker' , sql : `SHOW COLUMNS FROM \`${ originalTableName }\`;` , vendor : DBVendors . MYSQL , processExitOnError : true , shouldReturnClient : false } ; const tableColumnsResult = await DBAccess . query ( params ) ; const options = { delimiter : conversion . _delimiter , header : false , fields : tableColumnsResult . data . map ( ( column ) => column . Field ) } ; const streamTransformOptions = { highWaterMark : conversion . _streamsHighWaterMark , objectMode : true , encoding : conversion . _encoding } ; const json2CsvTransformStream = new Json2CsvTransform ( options , streamTransformOptions ) ; json2CsvTransformStream . on ( 'error' , async ( transformError ) => { await processDataError ( conversion , transformError , '' , '' , originalTableName , dataPoolId , client , originalSessionReplicationRole ) ; } ) ; return json2CsvTransformStream ; } ; /**
 * Disables all triggers and rules for current database session.
 * !!!DO NOT release the client, it will be released after current data-chunk deletion.
 */ const disableTriggers = async ( conversion , client ) : Promise < string > => { let sql = `SHOW session_replication_role;` ; let originalSessionReplicationRole = 'origin' ; try { const queryResult = await client . query ( sql ) ; originalSessionReplicationRole = queryResult . rows [ 0 ] . session_replication_role ; sql = 'SET session_replication_role = replica;' ; await client . query ( sql ) ; } catch ( error ) { await generateError ( conversion , `\t--[DataLoader::disableTriggers] ${ error }` , sql ) ; } return originalSessionReplicationRole ; } ; /**
 * Enables all triggers and rules for current database session.
 * !!!DO NOT release the client, it will be released after current data-chunk deletion.
 */ const enableTriggers = async ( conversion , client , originalSessionReplicationRole ) : Promise < void > => { const sql = `SET session_replication_role = ${ originalSessionReplicationRole };` ; try { await client . query ( sql ) ; } catch ( error ) { await generateError ( conversion , `\t--[DataLoader::enableTriggers] ${ error }` , sql ) ; } } ;