package Ġai . plat on . pulsar . ql . h 2 . utils Ċ Ċ import Ġai . plat on . pulsar . common . Strings Ċ import Ġai . plat on . pulsar . common . getLogger Ċ import Ġorg . apache . commons . lang 3 . StringUtils Ċ import Ġjava . sql . Connection Ċ import Ġjava . sql . ResultSet Ċ import Ġjava . sql . SQLException Ċ import Ġjava . sql . Types Ċ Ċ object ĠJdbc Utils Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= Ġget Logger ( this ) ĊĊĠĠĠ Ġ@ Throws ( SQLException :: class ) ĊĠĠĠ Ġfun Ġexecute Insert ( ĊĠĠĠĠĠĠĠ Ġsource Rs : ĠResultSet , ĊĠĠĠĠĠĠĠ Ġsink Connection : ĠConnection , ĊĠĠĠĠĠĠĠ Ġsink Table : ĠString , ĊĠĠĠĠĠĠĠ Ġdry Run : ĠBoolean Ġ= Ġfalse ĊĠĠĠ Ġ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġsink Connection . auto Commit Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġvar Ġaffected Rows Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġval Ġcolumn Count Ġ= Ġsource Rs . metaData . column Count ĊĠĠĠĠĠĠĠ Ġval Ġprefix Ġ= Ġ" insert Ġinto Ġ`$ sink Table ` (" ĊĠĠĠĠĠĠĠ Ġval Ġpostfix Ġ= Ġ") Ġvalues (" Ġ+ ĠStringUtils . repeat (" ?", Ġ", Ġ", Ġcolumn Count ) Ġ+ Ġ")" ĊĠĠĠĠĠĠĠ Ġval Ġinsert Sql Ġ= ĠInt Range ( 1 , Ġcolumn Count ). join ToString (", Ġ", Ġprefix , Ġpostfix ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"` " Ġ+ Ġsource Rs . metaData . getColumn Label ( it ) Ġ+ Ġ"` " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġsink Connection . prepareStatement ( insert Sql ). use Ġ{ Ġstmt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsource Rs . before First () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( source Rs . next ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠInt Range ( 1 , Ġcolumn Count ). forEach Ġ{ Ġk Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( source Rs . metaData . getColumn Type ( k )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTypes . BOOLEAN Ġ-> Ġstmt . set Boolean ( k , Ġsource Rs . getBoolean ( k )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTypes . FLOAT Ġ-> Ġstmt . set Float ( k , Ġsource Rs . get Float ( k )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTypes . INTEGER Ġ-> Ġstmt . setInt ( k , Ġsource Rs . getInt ( k )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġstmt . setString ( k , ĠStrings . strip Non Print able Char ( source Rs . getString ( k ))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠSQLException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Failed Ġto Ġcreate Ġstmt Ġ{}", Ġe . message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( dry Run ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsql Ġ= Ġ" insert " Ġ+ Ġstmt . toString (). substring After (" insert ") Ġ+ Ġ";" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResultSet Utils . sql Log . info ( sql ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaffected Rows Ġ= Ġstmt . executeUpdate () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsink Connection . commit () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġaffected Rows ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( SQLException :: class ) ĊĠĠĠ Ġfun Ġexecute Batch Insert ( ĊĠĠĠĠĠĠĠ Ġsource Result Sets : ĠIterable < ResultSet >, ĊĠĠĠĠĠĠĠ Ġsink Connection : ĠConnection , ĊĠĠĠĠĠĠĠ Ġsink Table : ĠString , ĊĠĠĠĠĠĠĠ Ġdry Run : ĠBoolean Ġ= Ġfalse ĊĠĠĠ Ġ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! source Result Sets . iterator (). hasNext ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ 0 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġsink Connection . auto Commit Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġvar Ġrows Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġaffected Rows : ĠInt Array ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġsample Ġ= Ġsource Result Sets . first () ĊĠĠĠĠĠĠĠ Ġval Ġcolumn Count Ġ= Ġsample . metaData . column Count ĊĠĠĠĠĠĠĠ Ġval Ġprefix Ġ= Ġ" insert Ġinto Ġ`$ sink Table ` (" ĊĠĠĠĠĠĠĠ Ġval Ġpostfix Ġ= Ġ") Ġvalues (" Ġ+ ĠStringUtils . repeat (" ?", Ġ", Ġ", Ġcolumn Count ) Ġ+ Ġ")" ĊĠĠĠĠĠĠĠ Ġval Ġinsert Sql Ġ= ĠInt Range ( 1 , Ġcolumn Count ). join ToString (", Ġ", Ġprefix , Ġpostfix ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"` " Ġ+ Ġsample . metaData . getColumn Label ( it ) Ġ+ Ġ"` " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġsink Connection . prepareStatement ( insert Sql ). use Ġ{ Ġstmt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsource Result Sets . forEach Ġ{ Ġrs Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrs . before First () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( rs . next ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠInt Range ( 1 , Ġcolumn Count ). forEach Ġ{ Ġk Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( rs . metaData . getColumn Type ( k )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTypes . BOOLEAN Ġ-> Ġstmt . set Boolean ( k , Ġrs . getBoolean ( k )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTypes . FLOAT Ġ-> Ġstmt . set Float ( k , Ġrs . get Float ( k )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTypes . INTEGER Ġ-> Ġstmt . setInt ( k , Ġrs . getInt ( k )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġstmt . setString ( k , ĠStrings . strip Non Print able Char ( rs . getString ( k ))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ++ rows ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstmt . add Batch () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠSQLException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Failed Ġto Ġcreate Ġstmt Ġ{}", Ġe . message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( dry Run ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsql Ġ= Ġ" insert " Ġ+ Ġstmt . toString (). substring After (" insert ") Ġ+ Ġ";" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResultSet Utils . sql Log . info ( sql ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaffected Rows Ġ= Ġstmt . execute Batch () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsink Connection . commit () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġaffected Rows ?. sum () Ġ?: Ġ 0 ĊĠĠĠ Ġ} Ċ }