packages feed

transfer-db-0.3.1.0: src/TransferDB.hs

module TransferDB where

import Prelude hiding (fail, log)

import System.IO (hPutStr, hPutStrLn, stderr, putStrLn, hFlush, stdout)

import Control.Concurrent (forkIO, ThreadId)
import Control.Concurrent.STM (TVar, newTVar, modifyTVar, readTVar, writeTVar,
                               TQueue, newTQueue, readTQueue, writeTQueue,
                               STM, atomically, check, orElse, retry)
       
import Control.Monad (replicateM_, join)
import Control.Monad.Trans.Maybe (MaybeT, runMaybeT)
import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask, asks, withReaderT)
import Control.Monad.Trans.Class (lift)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Control.Monad.Fail (MonadFail, fail)

import Control.Logging (log, withStderrLogging)

import Foreign.Marshal.Alloc (alloca, allocaBytes)
import Foreign.Ptr (Ptr, castPtr, nullPtr, wordPtrToPtr, ptrToWordPtr)
import Foreign.Storable (peek, poke)

import Data.String(IsString(fromString))
import Data.Char (toLower, toUpper)
import Data.List (find)

import SQL.CLI (SQLHDBC,
                SQLSMALLINT,
                SQLINTEGER,
                SQLPOINTER,
                SQLULEN,
                SQLLEN,
                sql_handle_dbc,
                sql_handle_stmt,
                sql_null_data,
                sql_all_types,
                sql_default,
                sql_char,
                sql_datetime,
                sql_code_date,
                sql_code_time,
                sql_code_timestamp,
                sql_type_date,
                sql_type_time,
                sql_type_timestamp,
                sql_varchar,
                sql_attr_imp_row_desc,
                sql_data_at_exec,
                sql_desc_type,
                sql_desc_length,
                sql_desc_octet_length,
                sql_desc_precision,
                sql_desc_scale,
                sql_desc_datetime_interval_code,
                sql_commit,
                sql_rollback)

import SQL.ODBC (sql_longvarchar, sql_binary, sql_longvarbinary, sql_varbinary, sql_interval, sql_attr_autocommit, sql_autocommit_off)
import SQL.CLI.Utils (SQLConfig,
                      ColumnInfo (ci_ColumnName, ci_TableSchem, ci_TableName),
                      allocHandle,
                      freeHandle,
                      tableExists,
                      forAllRecords,
                      collectColumnsInfo,
                      execDirect,
                      prepare,
                      execute,
                      getStorableStmtAttr,
                      numResultCols,
                      paramData,
                      forAllData,
                      putData,
                      getDescField,
                      bindParam,
                      setConnectAttr,
                      endTran,
                      toCLIType)

import Data.List (intercalate)

import TransferPlan (Batch, BatchItem,
                     ColumnName, WhereCondition,
                     scope_Db,
                     scope_UserName,
                     scope_Password,
                     scope_Schema,
                     plan_Source,
                     plan_Destination,
                     plan_Batches,
                     batch_Name,
                     batch_Items,
                     batch_Table,
                     batch_Where,
                     batch_OrderBy)
import Options (Options(option_Plan, option_Threads, option_Count, option_Drop))
import Database.TransferDB.Commons (finally, withConnection)

-- | the size of the buffer used to transfer data
transferBufferSize :: SQLLEN
transferBufferSize = 8192 * 8


data TransferOptions = TransferOptions {
  to_Options   :: Options,
  to_SQLConfig :: SQLConfig}


transferDB :: ReaderT TransferOptions (MaybeT IO) ()
transferDB = withStderrLogging $ do
  threads <- asks (option_Threads . to_Options)

  if threads == 0
    then do liftIO $ log $ fromString "Transfering database in single thread mode"
            schema1 <- asks (scope_Schema . plan_Source . option_Plan . to_Options)
            schema2 <- asks (scope_Schema . plan_Destination . option_Plan . to_Options)
            batches <- asks (plan_Batches . option_Plan . to_Options)
            sqlConfig <- asks to_SQLConfig
            dropBatches  <- asks (option_Drop . to_Options)
            countBatches <- asks (option_Count . to_Options)
            let batches'  = if dropBatches > 0 then drop dropBatches batches else batches
                batches'' = if countBatches > 0 then take countBatches batches' else batches'

            liftIO $ log $ fromString $ "Number of batches: " ++ (show $ length batches'') ++ "/" ++ (show $ length batches)
            result <- withSourceAndDest
              (\ srcDBC dstDBC -> liftIO $ allocaBytes (fromIntegral transferBufferSize)
                (\ p_transfer_buf ->
                    alloca
                    (\ p_transfer_len_or_ind ->
                        runMaybeT $ runReaderT (sequence_ $
                                                 map (transferBatch srcDBC schema1 dstDBC schema2 p_transfer_buf p_transfer_len_or_ind) batches'') sqlConfig)))
            maybe (fail "Transfer tables failed") return result
    else do liftIO $ log $ fromString $ "Transfering database using " ++ (show threads) ++ " threads"
            batchChan           <- liftIO $ atomically newTQueue
            allBatchesPublished <- liftIO $ atomically $ newTVar False
            workerThreads       <- liftIO $ atomically $ newTVar 0
            withReaderT (ThreadedTransferSetup batchChan allBatchesPublished workerThreads) $ do
              startWorkerThreads
              publishBatches
              waitForWorkToEnd


-- | Runs an action with 2 connections, to the source and destination dbs
withSourceAndDest :: (MonadIO m) => (SQLHDBC -> SQLHDBC -> ReaderT TransferOptions (MaybeT m) a) -> ReaderT TransferOptions (MaybeT m) a
withSourceAndDest f = do
  s1      <- asks (scope_Db       . plan_Source . option_Plan . to_Options)
  u1      <- asks (scope_UserName . plan_Source . option_Plan . to_Options)
  p1      <- asks (scope_Password . plan_Source . option_Plan . to_Options)
  s2      <- asks (scope_Db       . plan_Destination . option_Plan . to_Options)
  u2      <- asks (scope_UserName . plan_Destination . option_Plan . to_Options)
  p2      <- asks (scope_Password . plan_Destination . option_Plan . to_Options)
  withConnection s1 u1 p1
    (\ _ srcDBC -> do
        setConnectAttr srcDBC sql_attr_autocommit (wordPtrToPtr sql_autocommit_off) 0
        withConnection s2 u2 p2
          (\ _ dstDBC -> do
              setConnectAttr dstDBC sql_attr_autocommit (wordPtrToPtr sql_autocommit_off) 0
              f srcDBC dstDBC ))

-- | Executes one transfer batch. It gets the source db conection, the source schema, the destination db connection, the
-- destination schema, the pointers to allocated
-- buffers for receiving the data and length/indicator value, the batch and returns a 'ReaderT' action expecting
-- the 'SQLConfig' of the current CLI implementation.
transferBatch :: SQLHDBC -> String -> SQLHDBC -> String -> SQLPOINTER -> Ptr SQLLEN -> Batch -> ReaderT SQLConfig (MaybeT IO) ()
transferBatch srcDBC schema1 dstDBC schema2 p_transfer_buf p_transfer_len_or_ind batch = 
  let
    transfer_table' :: BatchItem -> ReaderT SQLConfig (MaybeT IO) ()
    transfer_table' batchItem = do
      let tableName      =  batch_Table   batchItem
          whereCondition =  batch_Where   batchItem
          orderBy        =  batch_OrderBy batchItem
      srcExists <- tableExists srcDBC schema1 tableName
      destTableName <- let lowerTableName = map toLower tableName
                           upperTableName = map toUpper tableName
                       in
                         findM (lift . (tableExists dstDBC schema2)) [tableName, lowerTableName, upperTableName]

      let (destExists, destTableName') = maybe (False, "") ((,) True) destTableName
      liftIO $ log $ fromString $ "------> Source table " ++ tableName ++ (if srcExists then " exists" else " does not exist")
      liftIO $ log $ fromString $ "------> Source table " ++ tableName ++  (if destExists
                                                                            then (" has destination: " ++ destTableName')
                                                                            else " has no destination.")
      if srcExists && destExists
        then transferTable srcDBC dstDBC (batch_Name batch) schema1 tableName schema2 destTableName' orderBy whereCondition p_transfer_buf p_transfer_len_or_ind
        else return ()
  in do
    liftIO $ log $ fromString "---------------------------------"
    liftIO $ log $ fromString $ "Batch: " ++ (batch_Name batch)
    liftIO $ log $ fromString "---------------------------------"
    sequence_ $ map transfer_table' $ batch_Items batch

data ThreadedTransferSetup = ThreadedTransferSetup {
  threaded_BatchChan           :: TQueue Batch,         -- ^ the queue for sending the batch information to the worker threads
  threaded_AllBatchesPublished :: TVar Bool,            -- ^ if false, there still are more batches to be published
  threaded_WorkerThreads       :: TVar Int,             -- ^ the number of active worker threads
  threaded_TransferOptions     :: TransferOptions       -- ^ the transfere options, read from the command line
  }

-- | Create the requested number of worker threads; the requested number of threads is read from the
-- 'threaded_TransferOptions' environment variable.
--
-- Each worker thread will evaluate the worker function ('transferBatch') for each batch read from the 'threaded_BatchChan' queue.
startWorkerThreads :: (MonadIO m) => ReaderT ThreadedTransferSetup (MaybeT m) ()
startWorkerThreads = do
  threads <- asks (option_Threads . to_Options . threaded_TransferOptions)
  replicateM_ threads runWorkerThread


-- | Start a worker thread responsible with evaluating the worker action function for each
-- batch read from the 'threaded_BatchChan'. It exists when all batches have been published and
-- the 'threaded_BatchChan' queue is empty
runWorkerThread :: (MonadIO m) => ReaderT ThreadedTransferSetup (MaybeT m) ThreadId
runWorkerThread = do
  workerThreadsVar      <- asks threaded_WorkerThreads
  setup                 <- ask
  
  liftIO $ forkIO $ do
    atomically $ modifyTVar workerThreadsVar (+ 1)
    log $ fromString "Thread started"
    result <- runMaybeT $ runReaderT processBatches setup
    log $ fromString $ maybe "Thread failed" (\ _ -> "Thread ended") result
    atomically $ modifyTVar workerThreadsVar (subtract 1)

-- | implements the action that runs in a thread, processing batches enqued
-- in a queue, one by one
processBatches :: ReaderT ThreadedTransferSetup (MaybeT IO) ()
processBatches = do
  queue                 <- asks threaded_BatchChan
  allBatchesPublishedVar<- asks threaded_AllBatchesPublished
  sqlConfig             <- asks (to_SQLConfig . threaded_TransferOptions)
  schema1               <- asks (scope_Schema   . plan_Source . option_Plan . to_Options . threaded_TransferOptions)
  schema2               <- asks (scope_Schema   . plan_Destination . option_Plan . to_Options . threaded_TransferOptions)
  
  withReaderT threaded_TransferOptions $ withSourceAndDest
    (\ srcDBC dstDBC ->
        liftIO $ allocaBytes (fromIntegral transferBufferSize)
        (\ p_transfer_buf ->
            alloca
            (\ p_transfer_len_or_ind ->
                let processBatches' :: IO ()
                    processBatches' = join $ atomically $ processNextBatch `orElse` checkForEnd
                    processNextBatch :: STM (IO ())
                    processNextBatch = do
                      batch <- readTQueue queue
                      return $ do
                        result <- runMaybeT $ runReaderT (transferBatch srcDBC schema1 dstDBC schema2 p_transfer_buf p_transfer_len_or_ind batch) sqlConfig
                        maybe (log $ fromString $ "batch " ++ (batch_Name batch) ++ " failed") (\ _ -> log $ fromString $ "batch " ++ (batch_Name batch) ++ " finished") result
                        processBatches'
                    checkForEnd = do
                      allBatchesPublished <- readTVar allBatchesPublishedVar
                      check allBatchesPublished
                      return $ return ()
                      
                in processBatches')))

-- | publish all batches to 'threaded_BatchChan' queue
publishBatches :: (MonadIO m) => ReaderT ThreadedTransferSetup m ()
publishBatches = do
  batches      <- asks (plan_Batches . option_Plan . to_Options . threaded_TransferOptions)
  dropBatches  <- asks (option_Drop . to_Options . threaded_TransferOptions)
  countBatches <- asks (option_Count . to_Options . threaded_TransferOptions)
  queue                 <- asks threaded_BatchChan
  allBatchesPublishedVar<- asks threaded_AllBatchesPublished

  let batches'  = if dropBatches > 0 then drop dropBatches batches else batches
      batches'' = if countBatches > 0 then take countBatches batches' else batches'

  liftIO $ log $ fromString $ "Number of batches: " ++ (show $ length batches'') ++ "/" ++ (show $ length batches)
  liftIO $ mapM_ (atomically . (writeTQueue queue)) batches''
  liftIO $ atomically $ writeTVar allBatchesPublishedVar True

-- | wait for worker threads to complete work
waitForWorkToEnd :: (MonadIO m) => ReaderT ThreadedTransferSetup m ()
waitForWorkToEnd = do
  allBatchesPublishedVar<- asks threaded_AllBatchesPublished
  workerThreadsVar      <- asks threaded_WorkerThreads
  liftIO $ atomically $ do
    allBatchesPublished <- readTVar allBatchesPublishedVar
    workerThreads       <- readTVar workerThreadsVar
    check (workerThreads <= 0 && allBatchesPublished) -- don't stop if there still are batches to be published

  
-- | Transfer data from one table in one database to another table in
-- another database. If all goes well, a line with the source table name
-- destination table name, number of records and total transferred size,
-- separated by comma is displayed on standard out.
transferTable :: SQLHDBC        -- ^ source connection handler
  -> SQLHDBC                    -- ^ destination connection handler
  -> String                     -- ^ batch name
  -> String                     -- ^ source schema name
  -> String                     -- ^ source table name
  -> String                     -- ^ destination schema name
  -> String                     -- ^ destination table name
  -> Maybe [ColumnName]         -- ^ optional order by fields; it should either be Nothing or Just a list with at least one column
  -> Maybe WhereCondition       -- ^ optional where condition
  -> SQLPOINTER                 -- ^ a pointer to a transfer buffer of size 'transferBufferSize'; it will be used to read data from the source fields
  -> Ptr SQLLEN                 -- ^ a pointer to a buffer used to store column size info
  -> ReaderT SQLConfig (MaybeT IO) ()
transferTable srcDBC dstDBC batchName srcSName srcTName dstSName dstTName orderBy whereCondition p_transfer_buf p_transfer_len_or_ind = do
  liftIO $ log $ fromString $ "Transferring table from source " ++ srcTName ++ " to destination " ++ dstTName

  -- 1. execute the select statemtn from source table
  liftIO $ log $ fromString "---------------------------- 1. SOURCE TABLE -------------------------------"
  colss <- collectColumnsInfo srcDBC srcSName srcTName
  select <- makeSelectSql colss orderBy whereCondition
  liftIO $ log $ fromString $ batchName ++ " (source-db) executing: " ++ select
  srcStmt <- allocHandle sql_handle_stmt srcDBC
  finally (freeHandle sql_handle_stmt srcStmt) $ do 
    execDirect srcStmt select (fail $ batchName ++ " parameter data expected on source select statement")

    -- 2. prepare the insert statement to destination table
    liftIO $ log $ fromString "---------------------------- 2. DESTINATION TABLE -------------------------------"
    colsd <- collectColumnsInfo dstDBC dstSName dstTName
    insert <- makeInsertSql colss colsd
    liftIO $ log $ fromString $ batchName ++ " (destination-db) preparing: " ++ insert
    dstStmt <- allocHandle sql_handle_stmt dstDBC
    finally (freeHandle sql_handle_stmt dstStmt) $ do
      prepare dstStmt insert

      -- 3. set dynamic parameters for the insert statement based on columns
      -- information from the select statement
      liftIO $ log $ fromString "---------------------------- 3. DYNAMIC PARAMETERS -------------------------------"
      numCols    <- numResultCols srcStmt
      srcDesc    <- getStorableStmtAttr srcStmt sql_attr_imp_row_desc

      let whereConditionInfo' = maybe "(-)" (\ w -> "(" ++ w ++ ")") whereCondition
          whereConditionInfo  = " " ++ whereConditionInfo'

      result     <- liftIO $ alloca
        (\ p_transferred ->
           -- p_transferred is an Ptr Integer to keep track of
           -- transferred size
           alloca
           (\ p_data_at_exec -> runMaybeT $ do
               -- p_data_at_exec is the address of a buffer used to set OCTET_LENGTH_PTR field
               -- of parameters descriptor for the insert into destination table statement. This
               -- buffer should contain the value 'SQL_DATA_AT_EXEC', so the value of the parameter
               -- will be asked when executing the insert statement.
               liftIO $ poke (p_data_at_exec :: Ptr SQLLEN) sql_data_at_exec
               liftIO $ poke p_transferred 0
               result <- liftIO $ allocaBytes 64
                 (\ p_buf -> let setParamFromStmt recno = do
                                   typeField <- liftIO $ runMaybeT $ do
                                     getDescField srcDesc recno sql_desc_type p_buf 64 nullPtr
                                     liftIO (peek (castPtr p_buf) :: IO SQLSMALLINT)

                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": type = " ++ (show typeField)

                                   subTypeField <-  liftIO $ runMaybeT $ do
                                     getDescField srcDesc recno sql_desc_datetime_interval_code p_buf 64 nullPtr
                                     liftIO (peek (castPtr p_buf) :: IO SQLSMALLINT)
                                     
                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": subtype = " ++ (show subTypeField)

                                   lengthField <- liftIO $ runMaybeT $ do
                                     liftIO $ poke ((castPtr p_buf) :: Ptr SQLULEN) 0
                                     getDescField srcDesc recno sql_desc_length p_buf 64 nullPtr
                                     liftIO (peek (castPtr p_buf) :: IO SQLULEN)
                                     
                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": length = " ++ (show lengthField)
                                   
                                   octetLenField <- liftIO $ runMaybeT $ do
                                     liftIO $ poke ((castPtr p_buf) :: Ptr SQLLEN) 0
                                     getDescField srcDesc recno sql_desc_octet_length p_buf 64 nullPtr
                                     liftIO (peek (castPtr p_buf) :: IO SQLLEN)
                                     
                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": octet length = " ++ (show octetLenField)

                                   precisionField <- liftIO $ runMaybeT $ do
                                     getDescField srcDesc recno sql_desc_precision p_buf 64 nullPtr
                                     liftIO (peek p_buf :: IO SQLSMALLINT)
                                     
                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": precision = " ++ (show precisionField)

                                   scaleField <- liftIO $ runMaybeT $ do
                                     getDescField srcDesc recno sql_desc_scale p_buf 64 nullPtr
                                     liftIO (peek p_buf :: IO SQLSMALLINT)
                                     
                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": scale = " ++ (show scaleField)
                                   
                                   let typeField'       = maybe sql_all_types id typeField
                                       translatedType   = toCLIType typeField'
                                       subTypeField'    = maybe sql_all_types id subTypeField
                                       lengthField'     = maybe 0 id lengthField
                                       precisionField'  = maybe 0 id precisionField
                                       scaleField'      = maybe 0 id scaleField
                                       lenprec = if typeField' `elem` [sql_char, sql_varchar, sql_datetime,
                                                                      sql_longvarchar, sql_binary, sql_longvarbinary,
                                                                      sql_varbinary, sql_interval]
                                                 then lengthField'
                                                 else fromIntegral $ precisionField'

                                   let translatedType' = case translatedType of
                                         x | x == sql_datetime -> case subTypeField' of
                                               x' | x' == sql_code_date -> sql_type_date
                                                  | x' == sql_code_time -> sql_type_time
                                                  | x' == sql_code_timestamp -> sql_type_timestamp
                                                  | otherwise -> sql_datetime
                                           | otherwise -> x
                                   --liftIO $ log $ fromString $ "recno " ++ (show recno) ++ ": lenprec = " ++ (show lenprec)
                                   bindParam dstStmt recno sql_default translatedType' lenprec scaleField' (wordPtrToPtr (fromIntegral recno)) p_data_at_exec
                                                                                                                            
                                   --liftIO $ log $ fromString "Implementation descriptor fields:"

                                   return ()
                             in
                               runMaybeT $ sequence_ [setParamFromStmt i | i <- [1..numCols]])
               maybe (fail $ batchName ++ ": set dynamic parameters for insert into the destination table failed") return result
               -- 4. for each row in the result set of select statement, execute the
               -- insert statement
               liftIO $ log $ fromString "---------------------------- 4. TRANSFER -------------------------------"
               let transferRow :: (MonadIO m, MonadFail m) => Int -> m Int
                   transferRow count = seq count $ do
                     if count `mod` 10000 == 0 then liftIO $ hPutStr stderr "." else return ()
                     execute dstStmt transferCols
                     return $! (count + 1)
                   transferCols :: (MonadIO m, MonadFail m) => m ()
                   transferCols = paramData dstStmt transferCol
                   transferCol :: (MonadIO m, MonadFail m) => SQLPOINTER -> m ()
                   transferCol p_data = do
                     let colno = fromIntegral $ ptrToWordPtr p_data
                     transferredSize <- seq colno $ forAllData srcStmt colno sql_default p_transfer_buf transferBufferSize p_transfer_len_or_ind transferChunk 0
                     crtTransferred <- liftIO $ peek p_transferred
                     seq crtTransferred $ seq transferredSize $ liftIO $ poke p_transferred $! (transferredSize + crtTransferred)
                   transferChunk :: (MonadIO m, MonadFail m) => Int -> m Int
                   transferChunk crtsize = seq crtsize $ do
                     size <- liftIO $ peek p_transfer_len_or_ind
                     let chunksize = if size < transferBufferSize then size else transferBufferSize
                     seq chunksize $ putData dstStmt p_transfer_buf chunksize
                     return $! if size == sql_null_data then 0 else (fromIntegral chunksize) + crtsize

               liftIO $ log $ fromString $ batchName ++ " - Start transfering table: " ++ srcTName
               count <- forAllRecords srcStmt transferRow 0
               transferred <- liftIO $ peek p_transferred
               
               liftIO $ log $ fromString $ "\n" ++ batchName ++ " - Transfer completed from " ++ srcTName ++ " to " ++ dstTName ++ whereConditionInfo
               liftIO $ log $ fromString $ batchName ++ " - Transferred " ++ (show count) ++ " records" ++ ", " ++ (show transferred) ++ " bytes" 
               liftIO $ putStrLn $ batchName ++ "," ++ srcTName ++ "," ++ dstTName ++ "," ++ (show count) ++ "," ++ (show transferred) ++ "," ++ whereConditionInfo'
               liftIO $ hFlush stdout
               commit))
      maybe (rollback >> (fail $ batchName ++ " - Transferred failed between " ++ srcTName ++ " and " ++ dstTName ++ whereConditionInfo)) return result 
        where
          rollback :: (MonadIO m) => m ()
          rollback = liftIO $ do
            liftIO $ log $ fromString $ batchName ++ " - Rollback transaction"
            _ <- runMaybeT $ endTran sql_handle_dbc dstDBC sql_rollback
            _ <- runMaybeT $ endTran sql_handle_dbc srcDBC sql_rollback
            return ()
          commit :: (MonadIO m, MonadFail m) => m ()
          commit = do
            result <- runMaybeT $ do
              liftIO $ log $ fromString $ batchName ++ " - Commit transaction"
              endTran sql_handle_dbc dstDBC sql_commit
              endTran sql_handle_dbc srcDBC sql_commit
            maybe (rollback >> fail (batchName ++ " - commit failed")) return result

makeSelectSql :: (MonadIO m, MonadFail m) => [ColumnInfo] -> Maybe [ColumnName] -> Maybe WhereCondition -> m String
makeSelectSql cs orderBy whereCondition = do
  tableName <- extractQualifiedTableName cs
  let select        = "select " ++ (fieldsList cs) ++ " from " ++ tableName
      select'       = maybe select addWhere whereCondition
      select''      = maybe select' addOrderBy orderBy
      addWhere w    = select ++ " where " ++ w
      addOrderBy fs = select' ++ " order by " ++ (intercalate ", " fs)
  return select''

makeInsertSql :: (MonadIO m, MonadFail m) => [ColumnInfo] -- ^ source columns info
  -> [ColumnInfo]                                         -- ^ destination columns info
  -> m String
makeInsertSql cs' cs = do
  tableName <- extractQualifiedTableName cs
  matchedFields <- findDstFields
  return $ "insert into " ++ tableName ++ " (" ++ (fieldsList' matchedFields) ++ ") values (" ++ values ++ ")"
  
  where dstfields = fields cs
        srcfields = fields cs'
        values    = intercalate ", " $ replicate n "?"
        n         = length cs
        findDstFields = sequence $ map findDstField srcfields
        findDstField f = do
          let dstField = find (`elem` dstfields) [f, lowerF, upperF]
          maybe failf return dstField            
            where upperF = map toUpper f
                  lowerF = map toLower f
                  failf = do let err = "source field " ++ f ++ " not found in destination table"
                             liftIO $ log $ fromString err
                             fail err
          
fieldsList :: [ColumnInfo] -> String
fieldsList cs = fieldsList' fieldNames
  where fieldNames = [field | field <- fields cs]

fieldsList' :: [String] -> String
fieldsList' fieldNames = intercalate ", " fieldNames

fields :: [ColumnInfo] -> [String]
fields cs = map ci_ColumnName cs

extractQualifiedTableName :: (MonadIO m, MonadFail m) => [ColumnInfo] -> m String
extractQualifiedTableName [] = do
  liftIO $ log $ fromString "extractQualifiedTableName called with no column info"
  fail "extractQualifiedTableName failed: columns list is empty"
extractQualifiedTableName (c:cs) =
  let schemaName  = ci_TableSchem c
      tableName   = ci_TableName c
      otherSchema = find ((schemaName /= ) . ci_TableSchem) cs
      otherTable  = find ((tableName /= ) . ci_TableName) cs
  in
    case otherSchema of
      Just s -> do
        let err = "Columns info contain different schema names: " ++ schemaName ++ ", " ++ (ci_TableSchem s)
        liftIO $ log $ fromString err
        fail err
      Nothing -> case otherTable of
        Just t -> do
          let err = "Columns info contain different table names: " ++ tableName ++ ", " ++ (ci_TableName t)
          liftIO $ log $ fromString err
          fail err
        Nothing -> case schemaName of
          [] -> return tableName
          s' -> return $ s' ++ "." ++ tableName

findM :: (Monad m) => (a -> m Bool) -> [a] -> m (Maybe a)
findM _ [] = return Nothing
findM f (x:xs) = do
  found <- f x
  if found then return $ Just x else findM f xs