transfer-db (empty) → 0.3.1.0
raw patch · 18 files changed
+3152/−0 lines, 18 filesdep +QuickCheckdep +aesondep +basesetup-changed
Dependencies added: QuickCheck, aeson, base, bytestring, cassava, clock, console-program, containers, cpu, hspec, logging, monad-control, sqlcli, sqlcli-odbc, stm, store, store-core, temporary, text, th-utilities, time, transfer-db, transformers, yaml
Files
- ChangeLog +81/−0
- LICENSE +30/−0
- README.md +9/−0
- Setup.hs +2/−0
- lib/Control/Monad/Trans/StringError.hs +107/−0
- lib/Database/TransferDB/Commons.hs +173/−0
- lib/Database/TransferDB/DumpDB.hs +740/−0
- lib/Database/TransferDB/DumpDB/Format.hs +226/−0
- src/CorrectionPlan.hs +215/−0
- src/Generator.hs +219/−0
- src/Main.hs +213/−0
- src/Options.hs +35/−0
- src/TransferDB.hs +530/−0
- src/TransferPlan.hs +124/−0
- tests/Database/TransferDB/DumpDB/FormatSpec.hs +206/−0
- tests/Database/TransferDB/DumpDBSpec.hs +158/−0
- tests/hspec-tests.hs +2/−0
- transfer-db.cabal +82/−0
+ ChangeLog view
@@ -0,0 +1,81 @@+2018-03-10 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * stack.yaml: update dependencies in to work with remote repositories + + * Release 0.3.1.0 + +2017-12-29 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (commands): add correctivePlan command + +2017-12-27 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Options.hs (readPlan): returns the Plan only + +2017-12-19 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/TransferDB.hs (transferDB): new functionality to skip a number of batches from the list of batches and to only process a certain count of batches + + * src/Main.hs (runTransferPlan): add count and drop options to run command + +2017-12-13 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/TransferDB.hs (transferDB): process batches in parallel + + * src/Main.hs (runTransferPlan): add multithreading command line option + +2017-11-26 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * lib/Database/TransferDB/Commons.hs (withEnvConnection): add withEnvConnection and withEnvConnection' + +2017-11-24 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (commands): add parallel threads option to dump command + +2017-11-07 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (commands): add dump and restore commands to transfer-db + executable. + +2017-11-03 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * lib/Database/TransferDB/DumpDB.hs: add database agnostic dump library + +2017-11-02 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/TransferDB.hs: use simple logging framework instead 'hPutStrLn stderr' + +2017-10-27 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * (release 0.3.0.0): subcommands in program interface, generate plans + +2017-09-21 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * (change summary): add functionality to generate plans + +2017-09-19 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (main): change program interface to accept commands (run, help) + +2017-09-19 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * (release 0.2.0.0): transfer plan, work in restartable batches + +2017-09-18 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (transferTable): use CLI types instead of ODBC type to avoid problem with PostgreSQL driver for LONGVARBINARY + (transferBufferSize): increase buffer to 64k to support max 64k PostgreSQL text fields (ODBC driver limitation for PostgreSQL) + +2017-09-17 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (transferDB): set autocommit off + (transferTable): add transaction demarcaion on each transferred table + +2017-09-15 Mihai Giurgeanu <mihai.giurgeanu@gmail.com> + + * src/Main.hs (main): change the command line argumets to get the plan file + (main): use Options module to process command line arguments + (transferDB): read all its arguments from the ReaderT structure TransferOptions + + * (release 0.1.0.0): No batch, "copy all" version of the software +
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author name here (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Author name here nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,9 @@+# transfer-db++Simple ODBC application that transfers data between 2 databases. ++The application uses the SQL/CLI API and only few ODBC extensions +that are absolutely needed to make ODBC work. This means that the +application is easily portable to other SQL/CLI implementations.++It includes a database agnostic dump file format library.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ lib/Control/Monad/Trans/StringError.hs view
@@ -0,0 +1,107 @@+{-- |+Module : Control.Monad.Trans.StringErrorT+Description : A variant of MaybeT transformer that returns an error string in case of failure+Copyright : (c) Mihai Giurgeanu, 2017+License : GPL-3+Maintainer : mihai.giurgeanu@gmail.com+Stability : experimental+Portability : Portable+--}++module Control.Monad.Trans.StringError where++import Control.Monad.Trans.Class (MonadTrans (lift))+import Control.Monad.IO.Class+import Control.Monad (MonadPlus(mzero, mplus))++import Control.Applicative+import Data.Foldable (Foldable(foldMap))+import Data.Traversable (Traversable(traverse))++import qualified Control.Monad.Fail as Fail++newtype StringErrorT m a = StringErrorT {runStringErrorT :: m (Either String a) }++-- | Transform the computation inside a @StringErrorT@.+--+-- * @'runStringErrorT' ('mapStringErrorT' f m) = f ('runStringErrorT' m)@+mapStringErrorT :: (m (Either String a) -> n (Either String b)) -> StringErrorT m a -> StringErrorT n b+mapStringErrorT f = StringErrorT . f . runStringErrorT+{-# INLINE mapStringErrorT #-}+++instance MonadTrans StringErrorT where+ lift m = StringErrorT $ m >>= (\ x -> return $ Right x)++instance (Functor m) => Functor (StringErrorT m) where+ fmap f = mapStringErrorT (fmap (fmap f))+ {-# INLINE fmap #-}++instance (Foldable f) => Foldable (StringErrorT f) where+ foldMap f (StringErrorT a) = foldMap (foldMap f) a+ {-# INLINE foldMap #-}++instance (Traversable f) => Traversable (StringErrorT f) where+ traverse f (StringErrorT a) = StringErrorT <$> traverse (traverse f) a+ {-# INLINE traverse #-}++instance (Functor m, Monad m) => Applicative (StringErrorT m) where+ pure = StringErrorT . return . Right+ {-# INLINE pure #-}+ mf <*> mx = StringErrorT $ do+ mb_f <- runStringErrorT mf+ case mb_f of+ Left s -> return (Left s)+ Right f -> do+ mb_x <- runStringErrorT mx+ case mb_x of+ Left s -> return (Left s)+ Right x -> return (Right (f x))+ {-# INLINE (<*>) #-}+ m *> k = m >>= \_ -> k+ {-# INLINE (*>) #-}++instance (Functor m, Monad m) => Alternative (StringErrorT m) where+ empty = StringErrorT (return $ Left "")+ {-# INLINE empty #-}+ x <|> y = StringErrorT $ do+ v <- runStringErrorT x+ case v of+ Left _ -> runStringErrorT y+ Right _ -> return v+ {-# INLINE (<|>) #-}++++instance (Monad m) => Monad (StringErrorT m) where+ return = StringErrorT . return . Right+ {-# INLINE return #-}++ (>>=) x f = StringErrorT $ do+ v <- runStringErrorT x+ case v of+ Left s -> return (Left s)+ Right y -> runStringErrorT (f y)+ {-# INLINE (>>=) #-}++ fail s = StringErrorT (return $ Left s)+ {-# INLINE fail #-}+++instance (Monad m) => Fail.MonadFail (StringErrorT m) where+ fail s = StringErrorT (return $ Left s)+ {-# INLINE fail #-}++instance (MonadIO m) => MonadIO (StringErrorT m) where+ liftIO = lift . liftIO+ {-# INLINE liftIO #-}++instance (Monad m) => MonadPlus (StringErrorT m) where+ mzero = StringErrorT (return $ Left "")+ {-# INLINE mzero #-}+ mplus x y = StringErrorT $ do+ v <- runStringErrorT x+ case v of+ Left _ -> runStringErrorT y+ Right _ -> return v+ {-# INLINE mplus #-}
+ lib/Database/TransferDB/Commons.hs view
@@ -0,0 +1,173 @@+{-- | +Module : Commons +Description : Common definitions used by other modules +Copyright : (c) Mihai Giurgeanu, 2017 +License : GPL-3 +Maintainer : mihai.giurgeanu@gmail.com +Stability : experimental +Portability : Portable +--} + +module Database.TransferDB.Commons where + +import Prelude hiding (fail, log) + +import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT) +import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask, asks) +import Control.Monad.Trans.Class (lift) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Fail (MonadFail, fail) +import Control.Logging(loggingLogger, LogLevel(LevelError), log) + +import Data.Text (Text) +import Data.String (fromString) + +import SQL.CLI (SQLHENV, SQLHDBC, sql_handle_env, sql_handle_stmt, sql_null_data, sql_char) +import SQL.CLI.Utils (connect, freeHandle, disconnect, tables, allocHandle, getData, forAllRecords) +import SQL.CLI.ODBC (setupEnv) + +import System.IO (hPutStrLn, stderr) + +import Foreign.Marshal.Alloc (alloca, allocaBytes) +import Foreign.Storable (peek, poke) +import Foreign.C.String (peekCString) +import Foreign.Ptr (castPtr) + +-- | runs the 'run' action, then, runs the 'afterRun' action no matter if the +-- 'run' action failed or succeeded +finally :: Monad m => m a -> ReaderT r (MaybeT m) b -> ReaderT r (MaybeT m) b +finally afterRun run = do + env <- ask + lift $ MaybeT $ runMaybeT (runReaderT run env) >>= (\ result -> afterRun >> return result) + + +-- | a 'MaybeT' only variant of 'finally'; it runs the second action and, then +-- the first action and returns the result of the second action +finally' :: (MonadIO m, MonadFail m) => IO a -> MaybeT IO b -> m b +finally' afterRun run = do + result <- liftIO $ runMaybeT run >>= (\result -> afterRun >> return result) + maybe (fail "action failed in finally'") return result + +-- | calls fail on the MonadFail logging an error message +faillog :: (MonadIO m, MonadFail m) => String -> m a +faillog = faillogS $ fromString "" + +-- | the variant of 'faillog' taking a log source as parameter +faillogS :: (MonadIO m, MonadFail m) => Text -> String -> m a +faillogS source msg = (liftIO $ loggingLogger LevelError source msg) >> fail msg + +-- | setup db environment and connect to database +withConnection :: (MonadIO m) + => String -- ^ datasource name + -> String -- ^ user name + -> String -- ^ password + -> (SQLHENV -> SQLHDBC -> (ReaderT r (MaybeT m) a)) -- ^ a function that gets the newly allocated environment and connection handlers + -> ReaderT r (MaybeT m) a +withConnection d u p f = do + liftIO $ log $ fromString $ "connect to " ++ d + henv <- setupEnv + let freeEnvHandle = do + liftIO $ log $ fromString "free environment handle" + liftIO $ freeHandle sql_handle_env henv + in finally freeEnvHandle $ do + + hdbc <- connect henv d u p + let freeHDBC = do + liftIO $ log $ fromString $ "disconnect from " ++ d + liftIO $ disconnect hdbc + in finally freeHDBC $ f henv hdbc + +withConnection' :: (MonadIO m, HasDBInfo r) => (SQLHENV -> SQLHDBC -> (ReaderT r (MaybeT m) a)) -> ReaderT r (MaybeT m) a +withConnection' f = do + dbi <- asks extractDBInfo + let d = dbi_Datasource dbi + u = dbi_User dbi + p = dbi_Password dbi + withConnection d u p f + + +-- | connect to database in an existing db environment +withEnvConnection :: (MonadIO m) + => SQLHENV -- ^ handle to environment + -> String -- ^ datasource name + -> String -- ^ user name + -> String -- ^ password + -> (SQLHENV -> SQLHDBC -> (ReaderT r (MaybeT m) a)) -- ^ a function that gets the newly allocated environment and connection handlers + -> ReaderT r (MaybeT m) a +withEnvConnection henv d u p f = do + liftIO $ log $ fromString $ "withEnvConnection' using environment handle " ++ (show henv) + hdbc <- connect henv d u p + let freeHDBC = do + liftIO $ log $ fromString $ "disconnect from " ++ d + liftIO $ disconnect hdbc + in finally freeHDBC $ f henv hdbc + +-- | call 'withEnvConnect' within a 'ReaderT' environment containing database connnetion info +withEnvConnection' :: (MonadIO m, HasDBInfo r) => SQLHENV -> (SQLHENV -> SQLHDBC -> (ReaderT r (MaybeT m) a)) -> ReaderT r (MaybeT m) a +withEnvConnection' henv f = do + dbi <- asks extractDBInfo + let d = dbi_Datasource dbi + u = dbi_User dbi + p = dbi_Password dbi + withEnvConnection henv d u p f + +-- | call 'connect' within a ReaderT environment containing database connnetion info +connect' :: (MonadIO m, MonadFail m, HasDBInfo r) => SQLHENV -> ReaderT r m SQLHDBC +connect' henv = do + dbi <- asks extractDBInfo + let d = dbi_Datasource dbi + u = dbi_User dbi + p = dbi_Password dbi + connect henv d u p + +-- | the environment used to run the program +data ProgramOptions = ProgramOptions { + po_Source :: DBInfo, + po_Dest :: DBInfo + } + +-- | Information about source or destination db +data DBInfo = DBInfo { + dbi_Datasource :: String, + dbi_User :: String, + dbi_Password :: String, + dbi_Schema :: String + } + +class HasDBInfo a where + extractDBInfo :: a -> DBInfo + +instance HasDBInfo DBInfo where + extractDBInfo = id + +-- | an instance that deals only with source db +instance HasDBInfo ProgramOptions where + extractDBInfo = po_Source + + +-- | run an action in the current environment on each table name from the current schema, +-- passing an accumulator value; returns the value of the accumulor +forAllTables :: (MonadFail m, MonadIO m, HasDBInfo r) => SQLHDBC -> a -> (a -> String -> ReaderT r (MaybeT m) a) -> ReaderT r (MaybeT m) a +forAllTables hdbc arg f = do + tables_stmt <- allocHandle sql_handle_stmt hdbc + liftIO $ log $ fromString $ "forAllTables allocated tables statement: " ++ (show tables_stmt) + schema <- asks $ dbi_Schema.extractDBInfo + + finally (liftIO $ freeHandle sql_handle_stmt tables_stmt) $ do + result <- liftIO $ runMaybeT $ tables tables_stmt Nothing (Just schema) Nothing (Just "TABLE") + let readTableName = liftIO $ allocaBytes 255 + (\ p_tableName -> + alloca + (\ p_tableName_ind -> do + poke p_tableName_ind 0 + _ <- getData tables_stmt 3 sql_char p_tableName 255 p_tableName_ind + tableName_ind <- liftIO $ peek p_tableName_ind + tableName <- if tableName_ind == sql_null_data + then return Nothing + else (liftIO . peekCString . castPtr) p_tableName >>= (return.Just) + return tableName)) + withTableName arg' = do + tableName <- readTableName + maybe (return arg') (f arg') tableName + result <- forAllRecords tables_stmt withTableName arg + return result
+ lib/Database/TransferDB/DumpDB.hs view
@@ -0,0 +1,740 @@+{-- |+Module : Database.TransferDB.DumpDB+Description : Database agnostic dump+Copyright : (c) Mihai Giurgeanu, 2017+License : GPL-3+Maintainer : mihai.giurgeanu@gmail.com+Stability : experimental+Portability : Portable+--}++{-# LANGUAGE FlexibleContexts, BangPatterns #-}+module Database.TransferDB.DumpDB where++import Prelude hiding (fail, log)++import System.IO (Handle, withBinaryFile, IOMode(ReadMode, WriteMode),+ hSeek, SeekMode(AbsoluteSeek, SeekFromEnd), hFlush, hGetBuf, hPutBuf,+ BufferMode(BlockBuffering), hSetBuffering, hSetBinaryMode)+import System.IO.Temp (withTempFile)++import System.Clock (Clock(Monotonic), TimeSpec(sec), getTime, diffTimeSpec, toNanoSecs)++import Control.Concurrent (forkIO)+import Control.Concurrent.STM (TVar, newTVar, modifyTVar, readTVar, writeTVar,+ TQueue, newTQueue, readTQueue, writeTQueue,+ TMVar, newTMVar, newEmptyTMVar, takeTMVar, putTMVar,+ STM, atomically, check, orElse, retry)+import Control.Monad(foldM, replicateM_, join)+import Control.Monad.Trans.Reader (ReaderT, runReaderT, asks, ask, withReaderT)+import Control.Monad.Trans.Maybe (MaybeT, runMaybeT)+import Control.Monad.Trans.StringError (runStringErrorT)++import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Fail (MonadFail, fail)+import Control.Monad.Trans.Control (MonadBaseControl)++import Control.Logging (log, debugS, withStderrLogging)++import Data.Time.Clock (getCurrentTime)+import Data.String (fromString)+import Data.Text (Text)+import Data.List (intercalate)++import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as C++import qualified Data.Map.Strict as Map++import Foreign.Marshal.Alloc (alloca, allocaBytes)+import Foreign.Ptr (Ptr, castPtr)+import Foreign.Storable (Storable(peek, poke))++import Database.TransferDB.DumpDB.Format+import Database.TransferDB.Commons (HasDBInfo(..), DBInfo(DBInfo), withConnection', connect', forAllTables, faillog, finally', finally)++import SQL.CLI (SQLHENV, SQLHDBC, SQLHSTMT, SQLINTEGER, SQLSMALLINT, SQLPOINTER, SQLHANDLE, SQLLEN, sql_handle_env, sql_handle_dbc, sql_handle_stmt, sql_no_nulls, sql_null_data)+import SQL.CLI.Utils (ColumnInfo(..), SQLConfig(..),+ collectColumnsInfo, collectColumnsInfo', allocHandle, columns,+ freeHandle, execDirect, forAllRecordsWithEndAndFail, forAllData, getData, disconnect)++import SQL.CLI.ODBC (odbcImplementation, setupEnv)+import SQL.ODBC (sql_char,+ sql_varchar,+ sql_longvarchar,+ sql_wchar,+ sql_wvarchar,+ sql_wlongvarchar,+ sql_decimal,+ sql_numeric,+ sql_bit,+ sql_tinyint,+ sql_smallint,+ sql_integer,+ sql_bigint,+ sql_real,+ sql_float,+ sql_double,+ sql_binary,+ sql_varbinary,+ sql_longvarbinary,+ sql_type_date,+ sql_type_time,+ sql_type_timestamp,+ sql_interval_year,+ sql_interval_month,+ sql_interval_day,+ sql_interval_hour,+ sql_interval_minute,+ sql_interval_second,+ sql_interval_year_to_month,+ sql_interval_day_to_hour,+ sql_interval_day_to_minute,+ sql_interval_day_to_second,+ sql_interval_hour_to_minute,+ sql_interval_hour_to_second,+ sql_interval_minute_to_second,+ sql_guid,+ sql_c_char,+ sql_c_wchar,+ sql_c_bit,+ sql_c_tinyint,+ sql_c_short,+ sql_c_long,+ sql_c_sbigint,+ sql_c_float,+ sql_c_double,+ sql_c_binary,+ sql_c_type_date,+ sql_c_type_time,+ sql_c_type_timestamp,+ sql_c_interval_year,+ sql_c_interval_month,+ sql_c_interval_day,+ sql_c_interval_hour,+ sql_c_interval_minute,+ sql_c_interval_second,+ sql_c_interval_year_to_month,+ sql_c_interval_day_to_hour,+ sql_c_interval_day_to_minute,+ sql_c_interval_day_to_second,+ sql_c_interval_hour_to_minute,+ sql_c_interval_hour_to_second,+ sql_c_interval_minute_to_second,+ sql_c_guid,+ odbcCTypeLen)+++logsrc :: Text+logsrc = fromString "Database.TransferDB.DumpDB"++-- | the maximum size of a chunk of variable length value+maxChunkSize :: (Num a) => a+maxChunkSize = 2 * 1024 * 1024++-- | the buffer size used to copy files+bufSize :: Int+bufSize = fromIntegral maxChunkSize++-- | file buffering mode+fileBuffering :: BufferMode+fileBuffering = BlockBuffering (Just $ 10 * 1024 * 1024)++-- | keep statistics by statement handle; for each handle record the+-- number of records dumped so far and the total size dumped so far+type StatisticsMap = Map.Map C.ByteString (Int, Int)++-- | dump database options+data DumpConfig = DumpConfig {+ dump_DSN :: String, -- ^ ODBC data source name + dump_UserName :: String, -- ^ user name + dump_Password :: String, -- ^ password + dump_Schema :: String, -- ^ schema to be dumped + dump_Description :: String, -- ^ dump description supplied by the user + dump_FilePath :: FilePath, -- ^ the dump file name+ dump_ParallelThreads :: Int, -- ^ the number of threads to be run in parallel+ dump_StatisticsVar :: TVar StatisticsMap, -- ^ the global statistics map+ dump_StartTime :: TimeSpec -- ^ the start time, used to compute the dump rate+ }+++instance HasDBInfo DumpConfig where+ extractDBInfo cfg = DBInfo (dump_DSN cfg) (dump_UserName cfg) (dump_Password cfg) (dump_Schema cfg)++-- | restore database options+data RestoreConfig = RestoreConfig {+ restore_DSN :: String, -- ^ ODBC data source name + restore_UserName :: String, -- ^ user name + restore_Password :: String, -- ^ password + restore_Schema :: String, -- ^ schema to be restored + restore_FilePath :: FilePath -- ^ the dump file name + }++instance HasDBInfo RestoreConfig where+ extractDBInfo cfg = DBInfo (restore_DSN cfg) (restore_UserName cfg) (restore_Password cfg) (restore_Schema cfg)++-- | dump a schema from an ODBC database to a binary file+dump :: (MonadIO m, MonadFail m, MonadBaseControl IO m) => ReaderT DumpConfig m ()+-- MonadBaseControl is required for logging+++dump = withStderrLogging $ do+ filename <- asks dump_FilePath+ config <- ask+ liftIO $ withBinaryFile filename WriteMode (\handle -> do+ result <- runMaybeT $ runReaderT (hDump handle) config+ maybe (fail "Database dump failed") return result)++-- | dumps the database schema to the file represented by the given handle+hDump :: (MonadIO m, MonadFail m) => Handle -> ReaderT DumpConfig (MaybeT m) ()+hDump handle = do+ liftIO $ B.hPut handle $ writeVersion V1+ header <- makeHeader+ liftIO $ B.hPut handle $ writeHeader header+ dumpTables handle+++-- | create the header of the dump file+makeHeader :: (MonadIO m) => ReaderT DumpConfig m HeaderV1+makeHeader = do+ timestamp <- liftIO getCurrentTime+ description <- asks dump_Description+ return $ HeaderV1 maxChunkSize timestamp (C.pack description)++-- | global environment for dumping tables in parallel+data ThreadedDump = ThreadedDump {+ threads_TablesChan :: TQueue String, -- ^ the channel to publish the name of tables+ threads_AllTablesPublishedVar :: TVar Bool, -- ^ when this is true, no more tables will be published on the tables channel+ threads_WorkerThreadsVar :: TVar Int, -- ^ the number of worker threads that are running; worker threads are the threads that actually dump data+ threads_HandleVar :: TMVar Handle, -- ^ provides synchronized access to the dump file handle+ threads_Config :: DumpConfig, -- ^ the dump configuration+ threads_HEnv :: SQLHENV, -- ^ shared environment allocated in the main thread; unixODBC has problem if the handle is allocated on another thread+ threads_AllocHandleChan :: TQueue (SQLSMALLINT, SQLHANDLE, TMVar SQLHANDLE) -- ^ channel to call allocHandle on the main thread+ }++instance HasDBInfo ThreadedDump where+ extractDBInfo = extractDBInfo . threads_Config++-- | dumps all tables in a schema, on a given connection+dumpTables :: (MonadIO m, MonadFail m) => Handle -> ReaderT DumpConfig (MaybeT m) ()+dumpTables handle = do+ threads <- asks dump_ParallelThreads+ case threads of+ 0 -> do liftIO $ log $ fromString "Dumping tables without using threads"+ _ <- withConnection' (\ _ hdbc -> withReaderT SingleThreaded $ forAllTables hdbc (0, 0) (dumpTable handle hdbc))+ return ()+ _ -> do liftIO $ log $ fromString $ "Dumping tables using " ++ (show threads) ++ " threads"+ tablesChan <- liftIO $ atomically newTQueue+ allTablesPublished <- liftIO $ atomically $ newTVar False+ workerThreads <- liftIO $ atomically $ newTVar 0+ dumpFileHandle <- liftIO $ atomically $ newTMVar handle+ henv <- setupEnv+ allocHandleChan <- liftIO $ atomically $ newTQueue+ finally (liftIO $ log (fromString $ "free environment handle " ++ (show henv)) >> freeHandle sql_handle_env henv) $ do+ withReaderT (\ cfg -> ThreadedDump tablesChan allTablesPublished workerThreads dumpFileHandle cfg henv allocHandleChan) $ do+ startWorkerThreads+ publishTables+ waitForWorkToEnd+ +-- | publish the tables to the chanel read by the worker threads+publishTables :: (MonadIO m, MonadFail m) => ReaderT ThreadedDump m ()+publishTables = do+ env <- ask+ henv <- asks threads_HEnv+ publishEndVar <- asks threads_AllTablesPublishedVar+ hdbc <- connect' henv+ let freehdbc = liftIO $ freeHandle sql_handle_dbc hdbc+ liftIO $ do+ result <- runMaybeT $ runReaderT (finally freehdbc $ forAllTables hdbc 0 publishTable) env+ maybe (log $ fromString $ "publishTables failed") (\ n -> log $ fromString $ "all tables have been published: " ++ (show n)) result+ atomically $ writeTVar publishEndVar True+ return ()++-- | monadic action to publish the table name on the tables channel+publishTable :: (MonadIO m, MonadFail m) => Int -> String -> ReaderT ThreadedDump m Int+publishTable crt tableName = do+ liftIO $ log $ fromString $ "publish table: " ++ tableName+ tablesChan <- asks threads_TablesChan+ liftIO $ atomically $ writeTQueue tablesChan tableName+ return (crt + 1)+ ++-- | start the worker threads; each worker thread will dump data in a temporary file,+-- then will append the contents of the temporary file to the contents of the dump file; the+-- append is done synchronized with the other worker threads, so only one worker thread will+-- append to the main dump file+startWorkerThreads :: (MonadIO m, MonadFail m) => ReaderT ThreadedDump m ()+startWorkerThreads = do+ count <- asks (dump_ParallelThreads . threads_Config)+ replicateM_ count startWorkerThread +++-- | start one worker thread+startWorkerThread :: (MonadIO m, MonadFail m) => ReaderT ThreadedDump m ()+startWorkerThread = do+ env <- ask+ henv <- asks threads_HEnv+ hdbc <- connect' henv+ let freehdbc = disconnect hdbc >> (liftIO $ freeHandle sql_handle_dbc hdbc) + threadsCountVar <- asks threads_WorkerThreadsVar+ t <- liftIO $ atomically $ do crt <- readTVar threadsCountVar+ writeTVar threadsCountVar (crt + 1)+ return (crt + 1)+ liftIO $ log $ fromString $ "Starting thread " ++ (show t)+ _ <- liftIO $ forkIO $ withTempFile "." "transfer-db.dmp"+ (\ path handle -> do+ log $ fromString $ "started worker thread: " ++ path+ hSetBuffering handle fileBuffering+ hSetBinaryMode handle True+ result <- runMaybeT $ runReaderT (finally freehdbc $ dumpThread handle henv hdbc) env+ log $ fromString $ "worker thread ended " ++ (maybe "with failure: " (\ _ -> "with success: ") result) ++ path+ atomically $ modifyTVar threadsCountVar (subtract 1) + )+ return ()++-- | the worker thread; gets a table name from the tables channel and dumps the data in the+-- temporary file; in the end, appends the temporary file contents to the end of the dump file contents+dumpThread :: (MonadIO m, MonadFail m) => Handle -> SQLHENV -> SQLHDBC -> ReaderT ThreadedDump m ()+dumpThread htmpfile _ hdbc = do+ tablesChan <- asks threads_TablesChan+ allTablesPublishedVar <- asks threads_AllTablesPublishedVar+ handleVar <- asks threads_HandleVar+ env <- ask+ + liftIO $ log $ fromString "entered dumpThread"+ let dumpNextTables :: STM (IO ())+ dumpNextTables = orElse dumpNextTables' waitTableOrEnd+ dumpNextTables' :: STM (IO ())+ dumpNextTables' = do+ tableName <- readTQueue tablesChan+ return $ do+ log $ fromString $ "start dumping table: " ++ tableName+ result <- runStringErrorT $ runReaderT (dumpTable htmpfile hdbc (0, 0) tableName) (MultiThreaded env)+ either (\s -> log $ fromString $ "dumping table " ++ tableName ++ " failed: " ++ s) (\ _ -> return ()) result+ join $ atomically $ dumpNextTables+ waitTableOrEnd :: STM (IO ())+ waitTableOrEnd = do+ allTablesPublished <- readTVar allTablesPublishedVar+ if allTablesPublished then return $ (log $ fromString $ "allTablesPublished = " ++ (show allTablesPublished)) >> finalizeDump else retry+ finalizeDump :: IO ()+ finalizeDump = do+ log $ fromString $ "finalizing dump thread"+ hdumpfile <- atomically $ takeTMVar handleVar+ copyTmpToDumpFile htmpfile hdumpfile+ atomically $ putTMVar handleVar hdumpfile+ ++ liftIO $ join $ atomically dumpNextTables++-- | finalizes the dump by copying the temporary file back into+-- the dump file+copyTmpToDumpFile :: Handle -> Handle -> IO ()+copyTmpToDumpFile htmp hdmp = do+ hFlush htmp+ hSeek htmp AbsoluteSeek 0+ hSeek hdmp SeekFromEnd 0+ allocaBytes bufSize+ (\ buf -> let copyFile = do+ sz <- hGetBuf htmp buf bufSize+ if sz > 0+ then do hPutBuf hdmp buf sz+ if sz >= bufSize then copyFile else return ()+ else return ()+ in copyFile )++-- | wait for worker threads to complete the work+waitForWorkToEnd :: (MonadIO m) => ReaderT ThreadedDump m ()+waitForWorkToEnd = do+ threadsCountVar <- asks threads_WorkerThreadsVar+ allocHandleChan <- asks threads_AllocHandleChan++ let waitIO = join $ atomically $ orElse (readTVar threadsCountVar >>= check . (<= 0) >> (return $ return ())) (allocHandleT allocHandleChan >>= \ io -> return (io >> waitIO))+ liftIO $ waitIO+ liftIO $ log $ fromString $ "all worker threads have finished"++-- | creates an IO action inside a STM monad to allocate a new handler in the current thread+allocHandleT :: (MonadIO m, MonadFail m) => TQueue (SQLSMALLINT, SQLHANDLE, TMVar SQLHANDLE) -> STM (m ())+allocHandleT chan = do+ (hType, hParent, retVar) <- readTQueue chan+ return $ allocHandle hType hParent >>= liftIO . atomically . (putTMVar retVar) ++-- | make a handle alloc request to the main thread and wait for result+allocHandleReq :: (MonadIO m, MonadFail m) => SQLSMALLINT -> SQLHANDLE -> ReaderT ThreadedDump m SQLHANDLE+allocHandleReq htype hparent = do+ allocHandleChan <- asks threads_AllocHandleChan+ resultVar <- liftIO $ atomically $ newEmptyTMVar+ liftIO $ atomically $ writeTQueue allocHandleChan (htype, hparent, resultVar)+ liftIO $ atomically $ takeTMVar resultVar++-- | environment for either single threaded or multi threaded dump+data SingleOrMulti = SingleThreaded DumpConfig | MultiThreaded ThreadedDump+ +-- | extract dumpConfig from a 'SingleOrMulti' structure+dumpConfig :: SingleOrMulti -> DumpConfig+dumpConfig (SingleThreaded x) = x+dumpConfig (MultiThreaded x) = threads_Config x++instance HasDBInfo SingleOrMulti where+ extractDBInfo = extractDBInfo . dumpConfig++-- | either directly alloc handle or call 'allocHandleReq' to alloc a handle deppending+-- on it is run on a threaded or non threaded environment+allocHandleSM :: (MonadIO m, MonadFail m) => SQLSMALLINT -> SQLHANDLE -> ReaderT SingleOrMulti m SQLHANDLE+allocHandleSM htype hparent = do+ env <- ask+ case env of+ SingleThreaded _ -> allocHandle htype hparent+ MultiThreaded threadeEnv -> withReaderT (const threadeEnv) $ allocHandleReq htype hparent++-- | workarround for unixODBC bug that requires that all handles should be allocated+-- on the main thread+collectColumnsInfoSM :: (MonadIO m, MonadFail m) => SQLHDBC -- ^ connection handle+ -> String -- ^ schema name+ -> String -- ^ table name+ -> ReaderT SingleOrMulti m [ColumnInfo]+collectColumnsInfoSM hdbc schemaName tableName = do+ env <- ask+ case env of+ SingleThreaded _ -> withReaderT (const odbcImplementation) $ collectColumnsInfo hdbc schemaName tableName+ MultiThreaded x -> withReaderT (const x) $ do+ hstmt <- allocHandleReq sql_handle_stmt hdbc+ columns hstmt Nothing (Just schemaName) (Just tableName) Nothing+ withReaderT (const odbcImplementation) $ collectColumnsInfo' hstmt++-- | dumps a single table+dumpTable :: (MonadIO m, MonadFail m) => Handle -> SQLHDBC -> (Int, Int) -> String -> ReaderT SingleOrMulti m (Int, Int)+dumpTable handle hdbc _ tableName = do+ schema <- extractSchema hdbc tableName+ liftIO $ debugS logsrc $ fromString $ "Schema " ++ (C.unpack $ schema_DBSchemaName schema) ++ "." ++ (C.unpack $ schema_TableName schema)+ liftIO $ sequence_ $ map debugFieldInfo $ schema_Fields schema + liftIO $ B.hPut handle $ writeSchema schema+ (!recs, !bytes) <- dumpTableData handle hdbc schema+ liftIO $ B.hPut handle writeEOT+ return (recs, bytes)+ +-- | logs the content of a 'FieldInfo' structure+debugFieldInfo :: FieldInfoV1 -> IO ()+debugFieldInfo f = do+ debugS logsrc $ fromString $ "\tfi_ColumnName: " ++ (C.unpack $ fi_ColumnName f)+ debugS logsrc $ fromString $ "\tfi_DataType: " ++ (show $ fi_DataType f)+ debugS logsrc $ fromString $ "\tfi_ColumnSize: " ++ (maybe "(null)" show $ fi_ColumnSize f)+ debugS logsrc $ fromString $ "\tfi_BufferLength: " ++ (maybe "(null)" show $ fi_BufferLength f)+ debugS logsrc $ fromString $ "\tfi_DecimalDigits: " ++ (maybe "(null)" show $ fi_DecimalDigits f)+ debugS logsrc $ fromString $ "\tfi_NumPrecRadix: " ++ (maybe "(null)" show $ fi_NumPrecRadix f)+ debugS logsrc $ fromString $ "\tfi_Nullabe: " ++ (show $ fi_Nullable f)+ +-- | extract schema infromation from the database, using an existing db connection+extractSchema :: (MonadIO m, MonadFail m) => SQLHDBC -> String -> ReaderT SingleOrMulti m SchemaV1+extractSchema hdbc tableName = do+ schemaName <- asks (dump_Schema.dumpConfig)+ fields <- extractSchemaFields hdbc tableName+ return $ SchemaV1 (C.pack schemaName) (C.pack tableName) fields++-- | extract the fields information from the database+extractSchemaFields :: (MonadIO m, MonadFail m) => SQLHDBC -> String -> ReaderT SingleOrMulti m [FieldInfoV1]+extractSchemaFields hdbc tableName = do+ schemaName <- asks (dump_Schema.dumpConfig)+ cols <- collectColumnsInfoSM hdbc schemaName tableName+ return $ map makeFieldInfo cols++-- | transforms a 'ColumnInfo' structure into a 'FieldInfoV1' structure+makeFieldInfo :: ColumnInfo -> FieldInfoV1+makeFieldInfo ci = FieldInfoV1 { fi_ColumnName = C.pack $ ci_ColumnName ci,+ fi_DataType = ci_DataType ci,+ fi_ColumnSize = ci_ColumnSize ci,+ fi_BufferLength = ci_BufferLength ci,+ fi_DecimalDigits = ci_DecimalDigits ci,+ fi_NumPrecRadix = ci_NumPrecRadix ci,+ fi_Nullable = ci_Nullable ci,+ fi_OrdinalPosition = ci_OrdinalPosition ci}++-- | dump the table records to the file, one by one; returns the (number of records,+-- size in bytes) of dumped data+dumpTableData :: (MonadIO m, MonadFail m) => Handle -> SQLHDBC -> SchemaV1 -> ReaderT SingleOrMulti m (Int, Int)+dumpTableData handle hdbc schema = do+ let tableName' = schema_QualifiedTableName schema+ tableName = C.unpack tableName'+ liftIO $ log $ fromString $ "dumping table: " ++ tableName+ select <- makeSelectSql schema+ hstmt <- allocHandleSM sql_handle_stmt hdbc+ env <- ask+ finally' ((log $ fromString $ "freeHandle for " ++ tableName) >> freeHandle sql_handle_stmt hstmt) $ do+ execDirect hstmt select (faillog $ "param data requested for select '" ++ select ++ "'")+ result <- liftIO $ allocaBytes (fromIntegral maxChunkSize)+ (\ p_transferBuf -> alloca+ (\ p_transferLenOrInd -> do+ let dumpAction = dumpRecord handle hstmt schema p_transferBuf maxChunkSize p_transferLenOrInd+ endAction hstmt x = logstats tableName' x >> return x+ failAction hstmt x s = logstats tableName' x >> fail s+ runStringErrorT $ runReaderT (forAllRecordsWithEndAndFail hstmt dumpAction (endAction hstmt) (failAction hstmt) (0,0)) env))+ (cnt, size) <- either (\ s -> faillog $ "transfer table " ++ (C.unpack $ schema_TableName schema) ++ " failed: " ++ s) return result+ liftIO $ log $ fromString $ "Finished dupming table " ++ (C.unpack $ schema_DBSchemaName schema)+ ++ "." ++ (C.unpack $ schema_TableName schema) ++ "; dumped " ++ (show cnt)+ ++ " records of " ++ (show size) ++ " bytes"+ size `seq` cnt `seq` return (cnt, size)++-- | dumps the data of one record into the file; it returns the (number of records, total bytes dumped)+dumpRecord :: (MonadIO m, MonadFail m) => Handle -> SQLHSTMT -> SchemaV1 -> SQLPOINTER -> SQLLEN -> Ptr SQLLEN -> (Int, Int) -> ReaderT SingleOrMulti m (Int, Int)+dumpRecord handle hstmt schema p_buf bufLen p_lenOrInd (cnt, sz) = do+ liftIO $ B.hPut handle writeRI+ -- TODO: if an error occurs writing any of the fields and one or more fields are not+ -- properly written in the file, the dump file will be corrupted+ let fields = zip (map fromIntegral [1 .. length fields']) fields'+ fields' = schema_Fields schema+ sz' <- foldM (dumpField handle hstmt p_buf bufLen p_lenOrInd) sz fields+ let cnt' = cnt + 1+ if cnt' `mod` 100000 == 0+ then do let tableName = schema_QualifiedTableName schema+ logStatistics tableName cnt' sz'+ else return ()+ cnt' `seq` sz' `seq` return (cnt', sz')++-- | monadic action that logs the data dumped until now; it uses a map that records the count of records+-- and the size dumped for each statement handle+logStatistics :: (MonadIO m, MonadFail m) => C.ByteString -> Int -> Int -> ReaderT SingleOrMulti m ()+logStatistics logkey cnt sz = do+ sizesVar <- asks readSizesVar+ (totalCnt, totalSz) <- liftIO $ atomically $ do+ sizesMap <- readTVar sizesVar+ let updatedSizesMap = Map.insert logkey (cnt, sz) sizesMap+ writeTVar sizesVar updatedSizesMap+ return $ Map.foldr' (\ (c1, s1) (c2, s2) -> (c1 + c2, s1 + s2)) (0, 0) updatedSizesMap+ startTime <- asks readStartTime+ crtTime <- liftIO $ getTime Monotonic+ let duration = diffTimeSpec crtTime startTime+ szRate = (fromIntegral totalSz) * 1000000000 `div` (toNanoSecs duration)+ liftIO $ log $ fromString $ ">>>>>> Running for " ++ (show $ sec duration) ++ " seconds"+ liftIO $ log $ fromString $ ">>>>>> (" ++ (C.unpack logkey) ++ ") " ++ (show totalCnt) ++ " records / " ++ (show $ totalSz) ++ " bytes" ++ ", " ++ (show szRate) ++ " bytes/sec"++-- | uncurried form of ('lotStatistics' hstmt)+logstats :: (MonadIO m, MonadFail m) => C.ByteString -> (Int, Int) -> ReaderT SingleOrMulti m ()+logstats hstmt = uncurry $ logStatistics hstmt++-- | get the 'TVar' 'StatiscsMap' from the environment+readSizesVar :: SingleOrMulti -> TVar StatisticsMap+readSizesVar (SingleThreaded x) = dump_StatisticsVar x+readSizesVar (MultiThreaded x) = (dump_StatisticsVar.threads_Config) x++-- | get the start time from the environment+readStartTime :: SingleOrMulti -> TimeSpec+readStartTime (SingleThreaded x) = dump_StartTime x+readStartTime (MultiThreaded x) = (dump_StartTime.threads_Config) x++-- | dump data of a single field; adds the size of dumped data to the total size+-- received as parameter+dumpField :: (MonadIO m, MonadFail m) => Handle -> SQLHSTMT -> SQLPOINTER -> SQLLEN -> Ptr SQLLEN -> Int -> (SQLSMALLINT, FieldInfoV1) -> m Int+dumpField handle hstmt p_buf buflen p_lenOrInd sz (crt, fld) =+ runReaderT dumpField' $ DumpFieldSpec handle hstmt p_buf buflen p_lenOrInd sz crt fld++-- | dump field data parameters+data DumpFieldSpec = DumpFieldSpec {+ dmpfld_Handle :: Handle, -- ^ dump file handle+ dmpfld_HStmt :: SQLHSTMT, -- ^ table select statement+ dmpfld_Buf :: SQLPOINTER, -- ^ data transfer buffer+ dmpfld_BufLen :: SQLLEN, -- ^ the size of data transfer buffer+ dmpfld_LenOrInd :: Ptr SQLLEN, -- ^ buffer to get null indicator or the actual size of transferred data+ dmpfld_Size :: Int, -- ^ the total data transferred prior to this field+ dmpfld_Crt :: SQLSMALLINT, -- ^ the number in the statement of the field to be dumped+ dmpfld_Field :: FieldInfoV1 -- ^ the description of the field+ }++-- | ReaderT monadic action that dumps a field into the dump file; returns the total dump size,+-- adding to the prior size the size in bytes of dumped data+dumpField' :: (MonadIO m, MonadFail m) => ReaderT DumpFieldSpec m Int+dumpField' = do+ dataType <- asks $ fi_DataType . dmpfld_Field+ case dataType of+ _ | dataType == sql_char -> dumpVarLenField sql_c_char+ | dataType == sql_varchar -> dumpVarLenField sql_c_char+ | dataType == sql_longvarchar -> dumpVarLenField sql_c_char+ | dataType == sql_wchar -> dumpVarLenField sql_c_wchar+ | dataType == sql_wvarchar -> dumpVarLenField sql_c_wchar+ | dataType == sql_wlongvarchar -> dumpVarLenField sql_c_wchar+ + | dataType == sql_decimal -> dumpVarLenField sql_c_char+ | dataType == sql_numeric -> dumpVarLenField sql_c_char++ | dataType == sql_bit -> dumpFixedField sql_c_bit+ | dataType == sql_tinyint -> dumpFixedField sql_c_tinyint+ | dataType == sql_smallint -> dumpFixedField sql_c_short+ | dataType == sql_integer -> dumpFixedField sql_c_long+ | dataType == sql_bigint -> dumpFixedField sql_c_sbigint++ | dataType == sql_real -> dumpFixedField sql_c_float+ | dataType == sql_float -> dumpFixedField sql_c_double+ | dataType == sql_double -> dumpFixedField sql_c_double++ | dataType == sql_binary -> dumpVarLenField sql_c_binary+ | dataType == sql_varbinary -> dumpVarLenField sql_c_binary+ | dataType == sql_longvarbinary -> dumpVarLenField sql_c_binary+ + | dataType == sql_type_date -> dumpFixedField sql_c_type_date+ | dataType == sql_type_time -> dumpFixedField sql_c_type_time+ | dataType == sql_type_timestamp -> dumpFixedField sql_c_type_timestamp++ | dataType == sql_interval_year -> dumpFixedField sql_c_interval_year+ | dataType == sql_interval_month -> dumpFixedField sql_c_interval_month+ | dataType == sql_interval_day -> dumpFixedField sql_c_interval_day+ | dataType == sql_interval_hour -> dumpFixedField sql_c_interval_hour+ | dataType == sql_interval_minute -> dumpFixedField sql_c_interval_minute+ | dataType == sql_interval_second -> dumpFixedField sql_c_interval_second+ | dataType == sql_interval_year_to_month -> dumpFixedField sql_c_interval_year_to_month+ | dataType == sql_interval_day_to_hour -> dumpFixedField sql_c_interval_day_to_hour+ | dataType == sql_interval_day_to_minute -> dumpFixedField sql_c_interval_day_to_minute+ | dataType == sql_interval_day_to_second -> dumpFixedField sql_c_interval_day_to_second+ | dataType == sql_interval_hour_to_minute -> dumpFixedField sql_c_interval_hour_to_minute+ | dataType == sql_interval_hour_to_second -> dumpFixedField sql_c_interval_hour_to_second+ | dataType == sql_interval_minute_to_second -> dumpFixedField sql_c_interval_minute_to_second+ | dataType == sql_guid -> dumpFixedField sql_c_guid+ | otherwise -> dumpUnknownFieldType++-- | dumps a variable field length; the variable field length will be dumped in chunks+-- each chunk having two fields: a length of the chunk and the actual data of the chunk. The+-- length of the chunk will be encoded on one, two or four bytes, depending on the maximum+-- chunk length and the maximum length of the field taken from the table's schema.+dumpVarLenField :: (MonadIO m, MonadFail m) => SQLSMALLINT -> ReaderT DumpFieldSpec m Int+dumpVarLenField bufferType = do+ bufferSize <- asks dmpfld_BufLen+ hstmt <- asks dmpfld_HStmt+ colnum <- asks dmpfld_Crt+ p_buffer <- asks dmpfld_Buf+ p_LenOrInd <- asks dmpfld_LenOrInd+ size <- asks dmpfld_Size+ lenlen <- octetLengthOfChunkSize+ + (_, size') <- forAllData hstmt colnum bufferType p_buffer bufferSize p_LenOrInd (dumpChunk lenlen) (0, size)+ size' `seq` return size'++-- | the octet length of the chunk size; it is calculated based on the transfer buffer size+-- and on the maxumum size of the field.+octetLengthOfChunkSize :: (Num a, Monad m) => ReaderT DumpFieldSpec m a+octetLengthOfChunkSize = do+ bufferSize <- asks dmpfld_BufLen+ fieldSize <- asks $ fi_BufferLength . dmpfld_Field+ let+ chunkSize = (fromIntegral bufferSize) - 1 -- for \0 character+ lenlen = maybe lenlen' (\sz -> if sz < chunkSize+ then if sz <= 256+ then 1+ else if sz <= 65536+ then 2+ else 4+ else lenlen') fieldSize+ lenlen' = if chunkSize <= 256+ then 1+ else if chunkSize <= 65536+ then 2+ else 4+ return lenlen++-- | dumps a chunk of data; increments the current size with the number of octets+-- dumped and returns this value+dumpChunk :: (MonadIO m, MonadFail m) => Int -> (Int, Int) -> ReaderT DumpFieldSpec m (Int, Int)+dumpChunk lenlen (chunkNo, size) = do+ nullable <- asks $ fi_Nullable.dmpfld_Field+ size' <- if chunkNo > 0 || nullable == sql_no_nulls + then dumpChunk' lenlen size+ else do p_lenOrInd <- asks dmpfld_LenOrInd+ lenOrInd <- liftIO $ peek p_lenOrInd+ if lenOrInd == sql_null_data+ then dumpNullIndicator Null size+ else do size'' <- dumpNullIndicator NotNull size+ dumpChunk' lenlen size''+ size' `seq` return (chunkNo+1, size')++dumpNullIndicator :: (MonadIO m) => NullIndicator -> Int -> ReaderT DumpFieldSpec m Int+dumpNullIndicator indicator size = do+ let bs = writeNullIndicator indicator+ handle <- asks dmpfld_Handle+ liftIO $ B.hPut handle bs+ let size' = size + (B.length bs)+ size' `seq` return size'++-- | dumps a chunk of data for a non null field; increments the current size with the number of octets+-- dumped and returns this value+dumpChunk' :: (MonadIO m, MonadFail m) => Int -> Int -> ReaderT DumpFieldSpec m Int+dumpChunk' lenlen size = do+ columnName <- asks $ fi_ColumnName.dmpfld_Field+ p_buf <- asks dmpfld_Buf+ p_lenOrInd <- asks dmpfld_LenOrInd+ handle <- asks dmpfld_Handle+ lenOrInd <- liftIO $ peek p_lenOrInd+ buffersize <- asks dmpfld_BufLen+ bs <- if lenOrInd > -1+ then let datalen = fromIntegral lenlen+ buffersize' = (fromIntegral buffersize) - 1 -- reserve space for null terminator+ chunklen = if datalen > buffersize' then buffersize' else datalen+ in return $ writeChunk (fromIntegral lenlen) chunklen (castPtr p_buf)+ else fail $ "dumpChunk' received unexpected null field (" ++ (show lenOrInd) ++ "); column " ++ (C.unpack columnName) + liftIO $ B.hPut handle bs+ let size' = size + (B.length bs)+ size' `seq` return size'+ +-- | dumps a fixed length field; the buffer length of the field will be taken from the description+-- of the field in the table's schema; the dumped data will only contain the data of the field;+-- the first parameter represents the ODBC C data type of the data to be read from the database+dumpFixedField :: (MonadIO m, MonadFail m) => SQLSMALLINT -> ReaderT DumpFieldSpec m Int+dumpFixedField bufferType = do+ bufferSize <- asks dmpfld_BufLen+ hstmt <- asks dmpfld_HStmt+ colnum <- asks dmpfld_Crt+ p_buffer <- asks dmpfld_Buf+ p_LenOrInd <- asks dmpfld_LenOrInd+ size <- asks dmpfld_Size++ liftIO $ poke p_LenOrInd 0+ getData hstmt colnum bufferType p_buffer bufferSize p_LenOrInd++ lenOrInd <- liftIO $ peek p_LenOrInd+ nullable <- asks $ fi_Nullable.dmpfld_Field+ + columnName <- asks $ fi_ColumnName.dmpfld_Field++ if nullable == sql_no_nulls+ then if lenOrInd == sql_null_data+ then fail $ "null value read from not nullable field " ++ (C.unpack columnName)+ else dumpFixedField' bufferType size lenOrInd+ else if lenOrInd == sql_null_data+ then dumpNullIndicator Null size+ else do size' <- dumpNullIndicator NotNull size+ dumpFixedField' bufferType size' lenOrInd++-- | dumps the data of a not null field+dumpFixedField' :: (MonadIO m, MonadFail m) => SQLSMALLINT -> Int -> SQLLEN -> ReaderT DumpFieldSpec m Int+dumpFixedField' bufferType size lenOrInd = do+ let wellKnownSize = odbcCTypeLen bufferType+ columnName <- asks $ fi_ColumnName.dmpfld_Field+ handle <- asks $ dmpfld_Handle+ p_buffer <- asks $ dmpfld_Buf+ bs <- if maybe False (lenOrInd /=) wellKnownSize+ then fail $ "the actual length of the field (" ++ (show lenOrInd) ++ ") is different from the schema size of the field (" ++ (show wellKnownSize) ++ "): " ++ (C.unpack columnName)+ else return $ writePlainBuf (castPtr p_buffer) (fromIntegral lenOrInd)+ liftIO $ B.hPut handle bs+ let size' = size + (B.length bs)+ size' `seq` return size'+++-- | dumps a field of an unknown type; the field will be converted to SQL_C_CHAR and dumped as+-- a string of characters+dumpUnknownFieldType :: (MonadIO m, MonadFail m) => ReaderT DumpFieldSpec m Int+dumpUnknownFieldType = fail "dumpUnknownFieldType not implemented"++-- | create the select for dumping table data+makeSelectSql :: (MonadIO m, MonadFail m) => SchemaV1 -> m String+makeSelectSql schema =+ let tableName = C.unpack $ schema_TableName schema+ schemaName = C.unpack $ schema_DBSchemaName schema+ qualifiedTableName = case schemaName of+ [] -> tableName+ s -> s ++ "." ++ tableName+ fieldsList = intercalate ", " $ map (C.unpack.fi_ColumnName) fields+ fields = schema_Fields schema+ in+ return $ "select " ++ fieldsList ++ " from " ++ qualifiedTableName+ ++-- | read the binary dump of a database and restores it in a destination ODBC data source+restore :: (MonadIO m, MonadFail m, MonadBaseControl IO m) => ReaderT RestoreConfig m () -- MonadBaseControl is required for logging+restore = withStderrLogging $ return ()++
+ lib/Database/TransferDB/DumpDB/Format.hs view
@@ -0,0 +1,226 @@+{-- |+Module : DumpDB+Description : Database agnostic dump file format+Copyright : (c) Mihai Giurgeanu, 2017+License : GPL-3+Maintainer : mihai.giurgeanu@gmail.com+Stability : experimental+Portability : Portable+--}++{-# LANGUAGE TemplateHaskell #-}+module Database.TransferDB.DumpDB.Format where++import Prelude hiding (fail)++import SQL.CLI (SQLSMALLINT, SQLINTEGER)+import Database.TransferDB.Commons (faillog)++import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Fail (MonadFail, fail)++import Data.Store (Store(size, peek, poke), Size(ConstSize, VarSize), Peek, Poke, encode)+import Data.Store.Core (pokeFromPtr, pokeFromForeignPtr, peekToPlainForeignPtr, unsafeEncodeWith)++import Data.String (fromString)+import Data.Word (Word8, Word16, Word32)++import Data.Time.Clock (UTCTime)++import Foreign.ForeignPtr (ForeignPtr, newForeignPtr_)+import Foreign.Ptr (Ptr)++import TH.Derive (derive, Deriving)++import qualified Data.ByteString.Char8 as C+import qualified Data.ByteString as B++-- | dump file format version+data Version = V1 deriving (Show, Ord, Eq)++-- | the version size is dependent on ++instance Store Version where+ size = ConstSize 1++ poke V1 = poke (1::Word8)++ peek = do+ v <- peek :: Peek Word8+ case v of+ _ | v == 1 -> return V1+ | otherwise -> fail $ "Unkown version tag found " ++ (show v)++-- | dump file header for version 1+data HeaderV1 = HeaderV1 {+ hv1_MaxChunkSize :: SQLINTEGER, -- ^ the maximum size of a data chunk; a field may have multiple data chunks+ hv1_Timestamp :: UTCTime, -- ^ the timestamp when the dump was made+ hv1_Description :: C.ByteString -- ^ the dump description provided by the user+ } deriving (Show, Eq)++$($(derive [d|+ instance Deriving (Store HeaderV1)+ |]))++-- | the schema reffers to the information about each field in one table+data SchemaV1 = SchemaV1 {+ schema_DBSchemaName :: C.ByteString, -- ^ database schema name+ schema_TableName :: C.ByteString, -- ^ the name of the table+ schema_Fields :: [FieldInfoV1] -- ^ information about each field in the table+ } deriving (Show, Eq)++-- | make a <schema>.<table_name> qualified table name+schema_QualifiedTableName :: SchemaV1 -> C.ByteString+schema_QualifiedTableName schema = (schema_DBSchemaName schema) `C.append` (C.pack ".") `C.append` (schema_TableName schema)+++-- | information about the name of the field, the length, precision, type, etc+data FieldInfoV1 = FieldInfoV1 {+ fi_ColumnName :: C.ByteString,+ fi_DataType :: SQLSMALLINT,+ fi_ColumnSize :: Maybe SQLINTEGER,+ fi_BufferLength :: Maybe SQLINTEGER,+ fi_DecimalDigits :: Maybe SQLSMALLINT,+ fi_NumPrecRadix :: Maybe SQLSMALLINT,+ fi_Nullable :: SQLSMALLINT,+ fi_OrdinalPosition :: SQLINTEGER+ } deriving (Show, Eq)++instance Ord FieldInfoV1 where+ compare f1 f2 = compare (fi_OrdinalPosition f1) (fi_OrdinalPosition f2)++$($(derive [d|+ instance Deriving (Store FieldInfoV1)+ |]))++$($(derive [d|+ instance Deriving (Store SchemaV1)+ |]))++-- | there are 2 types of record indicators:+-- * RI means that the following data is a new record in the current table;+-- * EOT (end of table) means that there is no more data for the current table+--+-- The current table is defined by the previous SCHEMA block in the dump file+data RecordIndicator = RI | EOT deriving (Show, Eq)++instance Store RecordIndicator where+ size = ConstSize 1++ poke RI = poke (255::Word8)+ poke EOT = poke (0 ::Word8)++ peek = do+ v <- peek :: Peek Word8+ case v of+ _ | v == 255 -> return RI+ | v == 0 -> return EOT+ | otherwise -> fail $ "Unknown value for record indicator found: " ++ (show v)++-- | indicates if a nullable field is Null or not Null; it is the first byte in encoded field value,+-- only for fields that are nullable; non-nullable fields have no null indicator+data NullIndicator = Null | NotNull deriving (Show, Eq)++instance Store NullIndicator where+ size = ConstSize 1++ poke Null = poke (0 ::Word8)+ poke NotNull = poke (255::Word8)++ peek = do+ v <- peek :: Peek Word8+ case v of+ _ | v == 255 -> return NotNull+ | v == 0 -> return Null+ | otherwise -> fail $ "Unknown value for null indicator: " ++ (show v)++-- | declare type Size as instance of Show class to be used in the tests and log messages+instance Show (Size a) where+ show (ConstSize sz) = "(ConstSize " ++ (show sz) ++ ")"+ show (VarSize _) = "VarSize"++-- | declare type Size as instance of Eq class to be used in test cases+instance Eq (Size a) where+ (==) (ConstSize sz1) (ConstSize sz2) = sz1 == sz2+ (==) _ _ = False+++-- | helper function to compute the store size in bytes of a value+sizeOf :: Size a -> a -> Int+sizeOf sza x = case sza of+ VarSize f -> f x+ ConstSize s -> s+++-- | the version is written as an Word8 length followed by the version+writeVersion :: Version -> B.ByteString+writeVersion v = let v' = encode v+ sz = fromIntegral $ B.length v'+ in+ B.cons sz v'++-- | the header is written as an Word16 length followed by the header itself+writeHeader :: HeaderV1 -> B.ByteString+writeHeader h = let h' = encode h+ sz = fromIntegral $ B.length h'+ in+ B.append (encode (sz :: Word16)) h'++-- | the schema is written as an Word32 length followed by the encoded schema+writeSchema :: SchemaV1 -> B.ByteString+writeSchema s = let s' = encode s+ sz = fromIntegral $ B.length s'+ in+ B.append (encode (sz :: Word32)) s'++-- | write an RI record indicator+writeRI :: B.ByteString+writeRI = encode RI++-- | write an EOT record indicator+writeEOT :: B.ByteString+writeEOT = encode EOT++-- | write a null indicator+writeNullIndicator :: NullIndicator -> B.ByteString+writeNullIndicator i = encode i++-- | a chunk of data consists of the length of the binary data and a pointer to+-- the memory of the data+data Chunk a = Chunk a (ForeignPtr Word8)++instance (Integral a, Store a) => Store (Chunk a) where+ size = VarSize $ \ (Chunk sz _) -> (sizeOf size sz) + (fromIntegral sz)++ poke (Chunk sz ptr) = do+ poke sz+ pokeFromForeignPtr ptr 0 (fromIntegral sz)++ peek = do+ sz <- peek+ ptr <- peekToPlainForeignPtr "Database.TransferDB.DumpDB.Format.Chunk" (fromIntegral sz)+ return $ Chunk sz ptr+++-- | write a chunk of binary data; the chunk has a length field followed by the binary data; the+-- length field may be on 1, 2 or 4 bytes; the first parameter is the length in bytes of the length+-- field, the second parameter is the length in bytes of the data+writeChunk :: Int -> Int -> Ptr Word8 -> B.ByteString+writeChunk lenlen sz ptr =+ let+ lenbs = case lenlen of+ 1 -> encode ((fromIntegral sz) :: Word8)+ 2 -> encode ((fromIntegral sz) :: Word16)+ 4 -> encode ((fromIntegral sz) :: Word32)+ _ -> error $ "encoding chunk failed because the length of chunk size field is " ++ (show lenlen) ++ " bytes; it should be either 1, 2 or 4 bytes"+ bufbs = writePlainBuf ptr sz+ in+ B.append lenbs bufbs++-- | encodes the memory buffer into a ByteString+writePlainBuf :: Ptr Word8 -> Int -> B.ByteString+writePlainBuf p l = unsafeEncodeWith pokeBytes l+ where+ pokeBytes :: Poke ()+ pokeBytes = pokeFromPtr p 0 l+
+ src/CorrectionPlan.hs view
@@ -0,0 +1,215 @@+{-|+Module : CorrectionPlan+Description : Given the plan and the execution log, it generates a new plan and an SQL script+ meant to correct eventual failures+License : GPL-3+Maintainer : mihai.giurgeanu@gmail.com+Stability : experimental+Portability : GHC+-}+{-# LANGUAGE FlexibleContexts, DeriveGeneric #-}+module CorrectionPlan where++import Prelude hiding (fail, log)++import Control.Logging (log, withStderrLogging)++import Control.Monad.Fail (MonadFail, fail)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Trans.Reader (ReaderT, asks, ask)+import Control.Monad.Trans.State (StateT, gets, modify', execStateT)++import qualified Data.ByteString.Lazy as B+import qualified Data.ByteString as B1 (hPutStr)+import Data.Csv (FromRecord)+import Data.Csv.Streaming (decode, Records(Cons, Nil), HasHeader(NoHeader))+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.List (foldl')+import Data.String(IsString(fromString))+import Data.Set (Set)+import qualified Data.Set as Set+import Data.Yaml.Aeson (encode)++import GHC.Generics (Generic)++import TransferPlan (TransferPlan(TransferPlan, plan_Source, plan_Destination), Batch(Batch), BatchItem(BatchItem))++import System.IO (Handle, hSetBinaryMode, hPutStrLn)++data CorrectionConfig = CorrectionConfig {+ correction_Plan :: TransferPlan, -- ^ the original plan+ correction_LogHandle :: Handle, -- ^ the input file handle to the analyzed execution log+ correction_ScriptHandle :: Handle, -- ^ the output file handle to the generated correction sql script+ correction_PlanHandle :: Handle -- ^ the output file handle to the generated correction plan+ }+ +-- | generates the contents of 2 files: a sql script and a new yaml transfer plan+-- to be run to correct the possible errors in running a previous plan, preventing duplicated+-- data or missing data.+--+-- The 2 files are meant to correct the following errors:+--+-- 1. some items in the original transfer plan were loaded 2 ore more times (so the data is duplicated in the destination table)+-- 2. some items were skipped in the previous run, so data is missing in the destination table.+--+-- In the first case, the table containing duplicated records is truncated (with a statement in+-- the sql script) and one or more batches are generated to load again the data in that table.+--+-- In the second case, one ore more batches are generated in the new plan to load the+-- items that were missed at the first run.+--+-- Please note, that the original plan should not contain duplicated data transfers, that is, items with+-- the same table source name and same where condition.+generateCorrections :: (MonadIO m, MonadFail m, MonadBaseControl IO m) => ReaderT CorrectionConfig m ()+generateCorrections = withStderrLogging $ do+ logh <- asks correction_LogHandle+ liftIO $ hSetBinaryMode logh True+ logstream <- liftIO $ B.hGetContents logh+ cfg <- ask+ let plan = correction_Plan cfg+ itemsMap' = itemsMap plan+ tablesMap'= tablesMap plan+ initialCorrectionState = CorrectionState cfg 0 Set.empty Set.empty Set.empty (Map.keysSet itemsMap') tablesMap'+ + s <- lift $ execStateT (processLogRecords $ decode NoHeader logstream) initialCorrectionState+ + source <- asks (plan_Source . correction_Plan)+ destination <- asks (plan_Destination . correction_Plan)+ batches <- fromItems (correctionState_GeneratedItems s) itemsMap'+ let newPlan = TransferPlan source destination batches+ planh <- asks correction_PlanHandle+ liftIO $ B1.hPutStr planh $ encode newPlan++-- | generate new plan batches list using original batch names and original items+fromItems :: (MonadIO m) => Set (String, String) -> Map (String, String) (String, BatchItem) -> m [Batch]+fromItems items m = Set.foldl' insertItem (return Map.empty) items >>= return . Map.elems+ where insertItem m' i = do+ m'' <- m'+ let originalinfo = Map.lookup i m+ case originalinfo of+ Nothing -> do liftIO $ log $ fromString $ "Error: Item " ++ (show i) ++ " not found in original plan. The item will not be included in the correction plan. This is a bug, please report it!"+ return m''+ Just (batchName, batchItem) -> let existingBatch = Map.lookup batchName m''+ in case existingBatch of+ Nothing -> return $ Map.insert batchName (Batch batchName [batchItem]) m''+ Just (Batch _ bis) -> return $ Map.insert batchName (Batch batchName (bis ++ [batchItem])) m''+ +-- | keeps the state of generating the correction scripts and batches during the processing of the log records+data CorrectionState = CorrectionState {+ correctionState_Config :: CorrectionConfig, -- ^ the current configuration+ correctionState_Rec :: Int, -- ^ the current record number+ correctionState_Truncated :: Set String, -- ^ the 'Set' of tables for which truncate script have been generated (due to duplication)+ correctionState_UniqueRecords :: Set (String, String), -- ^ the 'Set' of (source table, condition) found in processing the logs+ correctionState_GeneratedItems:: Set (String, String), -- ^ the 'Set' of (table name, where condition) representing items already added to the new correction plan+ correctionState_SkippedItems :: Set (String, String), -- ^ the 'Set' of (table name, where condition) representing items not found in the transfer log+ correctionState_TableItems :: Map String (Set (String, String)) -- ^ the 'Map' from table name to the 'Set' of (table name, where condition) indicating the items needed to reload the table+ }+ ++-- | create a map from the items in the batch to the (batch name, 'BatchItem') pair;+-- for each item, it will be considered+-- only the table name and the where condition+itemsMap :: TransferPlan -> Map (String, String) (String, BatchItem)+itemsMap (TransferPlan _ _ bs) = foldl' addBatchToItemsMap Map.empty bs++-- | adds the items inside a batch to an items map in the sense of 'itemsMap'+addBatchToItemsMap :: Map (String, String) (String, BatchItem) -> Batch -> Map (String, String) (String, BatchItem)+addBatchToItemsMap m (Batch name items) = foldl' (\ m' b@(BatchItem t _ w) -> Map.insert (makeItemKey t w) (name, b) m') m items++-- | create a map from the table names to the list of (table name, where condition) pairs from+-- all 'BatchItem' elements in the original plan used to load the given table+tablesMap :: TransferPlan -> Map String (Set (String, String))+tablesMap (TransferPlan _ _ bs) = foldl' addBatchToTablesMap Map.empty bs++-- | add the items inside the batch to the 'tablesMap' result+addBatchToTablesMap :: Map String (Set (String, String)) -> Batch -> Map String (Set (String, String)) +addBatchToTablesMap m (Batch _ items) = foldl' (\ m' (BatchItem t _ w) -> Map.insertWith Set.union t (Set.singleton $ makeItemKey t w) m') m items++-- | make an item key, that is a (table name, where condition) pair, from the table name and+-- the where condition as found in a 'BatchItem'+makeItemKey :: String -> Maybe String -> (String, String)+makeItemKey t w = (t, maybe "(-)" (\ w' -> "(" ++ w' ++ ")") w)++-- | it sequentially traverses the log records and generates the sql correction script and+-- the correction plan+processLogRecords :: (MonadIO m, MonadFail m) => Records LogRecord -> StateT CorrectionState m ()+processLogRecords (Cons r rs) = do+ n <- gets correctionState_Rec+ modify' (\ s -> s {correctionState_Rec = n + 1})+ processLogRecord r+ processLogRecords rs+ +processLogRecords (Nil err rest) = do+ n <- gets correctionState_Rec+ let restStr = map (toEnum.fromIntegral) $ B.unpack $ B.take 50 rest+ case err of+ Nothing -> do+ liftIO $ log $ fromString $ "All log records have been processed at record " ++ (show n) ++ ", at: '" ++ restStr ++ "'"+ generatedItems <- gets correctionState_GeneratedItems+ skippedItems <- gets correctionState_SkippedItems+ modify' (\ s -> s { correctionState_GeneratedItems = generatedItems `Set.union` skippedItems})+ Just msg -> do+ liftIO $ log $ fromString $ "Parsing log records failed with error: " ++ msg ++ " when reading record " ++ (show $ n + 1) ++ " at: '" ++ restStr ++ "'"+ liftIO $ log $ fromString "Warning: because of the error parsing the tranfer log, no skipped items are included in the generated corrections plan."+ ++-- | process a single log record; it checks if the record is duplicated and, if it is, it+-- generates the sql script and the plan to correct the duplicate processing of the record+processLogRecord :: (MonadIO m) => Either String LogRecord -> StateT CorrectionState m ()+processLogRecord (Left msg) = do+ n <- gets correctionState_Rec+ liftIO $ log $ fromString $ "Error processing log record " ++ (show n) ++ ": " ++ msg+processLogRecord (Right r@(LogRecord _ t _ _ _ w)) = do+ urs <- gets correctionState_UniqueRecords+ n <- gets correctionState_Rec+ if Set.member (t, w) urs+ then processDuplicatedRecord r+ else do modify' (\s -> s { correctionState_UniqueRecords = Set.insert (t, w) urs})+ srs <- gets correctionState_SkippedItems+ if Set.member (t, w) srs+ then modify' (\ s -> s {correctionState_SkippedItems = Set.delete (t, w) srs})+ else liftIO $ log $ fromString $ "WARN at record " ++ (show n) ++ ": item not duplicated but not found in the skipped items (" ++ (show r) ++ ")"+++-- | processes a duplicated record; generates the sql script to truncate the table+-- containing duplicated records and generates the items to reimport the table, based+-- on the original plan; updates the 'CorrectionState' structure+processDuplicatedRecord :: (MonadIO m) => LogRecord -> StateT CorrectionState m ()+processDuplicatedRecord r@(LogRecord _ t _ _ _ w) = do+ liftIO $ log $ fromString $ "Item <" ++ (show r) ++ "> is duplicated"+ truncated <- gets correctionState_Truncated+ if Set.member t truncated+ then liftIO $ log $ fromString $ "Table " ++ t ++ " already reloaded in the new plan by a previous duplicated item. Skip the table this time."+ else do sqlh <- gets (correction_ScriptHandle . correctionState_Config)+ liftIO $ hPutStrLn sqlh $ "TRUNCATE TABLE " ++ t ++ ";"+ generateItemsForDuplicatedRecord r+ modify' (\ s -> s { correctionState_Truncated = Set.insert t truncated})++-- | updates the 'CorrectionState' state by generating new plan items to reload+-- data for the table where duplicated records were found+generateItemsForDuplicatedRecord :: (MonadIO m) => LogRecord -> StateT CorrectionState m ()+generateItemsForDuplicatedRecord (LogRecord _ t _ _ _ _) = do+ tableItems <- gets correctionState_TableItems+ let items' = Map.lookup t tableItems+ case items' of+ Nothing -> liftIO $ log $ fromString $ "Error: duplicated items found in the log for table " ++ t ++ " but no items for this table could be found in the original plan. This seems to be a bug. Please report it!"+ Just items -> do generatedItems <- gets correctionState_GeneratedItems+ modify' (\s -> s {correctionState_GeneratedItems = Set.union generatedItems items})++-- | a transfer log record; a transfer log record is generated for each transferred item+-- in a batch, after the trasfer succeeded; analyzing the log we can spot the erros, that is+-- items that have not been transferred successfully or items that have been transferred twice+-- because the plan was ran multiple times+data LogRecord = LogRecord {+ log_Batch :: !String,+ log_SourceTable :: !String,+ log_DestTable :: !String,+ log_Count :: !Int,+ log_Size :: !Int,+ log_Where :: !String+ } deriving (Show, Generic)++instance FromRecord LogRecord
+ src/Generator.hs view
@@ -0,0 +1,219 @@+{-- | +Module : Generator +Description : Automatic creation of transfer plans from the source db +Copyright : (c) Mihai Giurgeanu, 2017 +License : GPL-3 +Maintainer : mihai.giurgeanu@gmail.com +Stability : experimental +Portability : Portable +--} + +module Generator where + +import Prelude hiding (fail, log) +import System.IO (hPutStrLn, hPutStr, stderr) + +import Control.Logging (log, withStderrLogging) + +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Fail (MonadFail, fail) +import Control.Monad.Trans.Reader (ReaderT, withReaderT, asks) +import Control.Monad.Trans.Maybe (MaybeT) + +import Data.String(IsString(fromString)) + +import SQL.CLI (SQLHDBC, sql_handle_stmt, sql_null_data, sql_char) +import SQL.CLI.Utils (tables, allocHandle, getData, forAllRecords, freeHandle, execDirect, forAllData) + +import Foreign.Marshal.Alloc (alloca, allocaBytes) +import Foreign.Storable (peek, poke) +import Foreign.C.String (peekCString) +import Foreign.Ptr (castPtr) + +import Text.Printf (printf) +import qualified Data.ByteString as B + +import Database.TransferDB.Commons (ProgramOptions, + DBInfo, + withConnection', + po_Source, + po_Dest, + dbi_Datasource, + dbi_User, + dbi_Password, + dbi_Schema) + +import TransferPlan (TransferPlan (TransferPlan), + DatabaseScope (Scope), + Batch(Batch), + BatchItem(BatchItem), + ColumnName, + batch_Items, batch_Where, plan_Batches) + +import Data.Yaml.Aeson (encode) +import Data.List (intercalate) + +maxKeyBuffer :: Int +maxKeyBuffer = 1024 + +-- | Writes to standard output a simple transfer plan, including all the tables +-- in the source database +makeSimplePlan :: ReaderT ProgramOptions (MaybeT IO) () +makeSimplePlan = + withStderrLogging $ withConnection' + (\ _ hdbc -> do + let addTable :: (MonadIO m) => Batch -> String -> m Batch + addTable batch tableName = do + liftIO $ log $ fromString $ "add table " ++ tableName + return batch { batch_Items = (batch_Items batch) ++ [BatchItem tableName Nothing Nothing] } + plan <- initPlan + liftIO $ log $ fromString "Reading tables from source ..." + batch <- withTables hdbc (Batch "All tables" []) addTable + let plan' = plan { plan_Batches = [batch] } + liftIO $ B.putStr $ encode plan') + + + +-- | Writes to standard output a transfer plan, counting the number of tables included in each batch, +-- and making batches to hold the given number of records +makePlanByTables :: Integer -> ReaderT ProgramOptions (MaybeT IO) () +makePlanByTables n = withStderrLogging $ do + liftIO $ log $ fromString $ "Making plan with " ++ (show n) ++ " tables per batch" + withConnection' + (\ _ hdbc -> do + let addTable :: (MonadIO m) => (Integer, Int, Batch, [Batch]) -> String -> m (Integer, Int, Batch, [Batch]) + addTable (crt, crtb, b, bs) tableName = do + liftIO $ log $ fromString $ "add table " ++ tableName + return $ if crt `mod` n == 0 && crt > 0 + then (crt + 1, crtb + 1, newBatch, bs ++ [b]) + else (crt + 1, crtb, addItem, bs) + where newBatch = Batch (printf "%04d" $ crtb + 1) [batchItem] + addItem = b { batch_Items = (batch_Items b) ++ [batchItem] } + batchItem = BatchItem tableName Nothing Nothing + liftIO $ log $ fromString "Reading tables from source ... " + (_, _, lastBatch, closeBatches) <- withTables hdbc (0, 1, Batch "0001" [], []) addTable + plan <- initPlan + let plan' = plan { plan_Batches = closeBatches ++ [lastBatch] } + liftIO $ B.putStr $ encode plan' ) + liftIO $ log $ fromString "Plan generated" + + +-- | Writes to standard output a transfer plan, counting the number of records included in each batch, +-- and making batches to hold the given number of records +makePlanByRows :: Integer -> ReaderT ProgramOptions (MaybeT IO) () +makePlanByRows n = withStderrLogging $ do + liftIO $ log $ fromString $ "Making a plan with " ++ (show n) ++ " rows transferred in each batch" + schemaName <- asks (dbi_Schema . po_Source) + withConnection' + (\ _ hdbc -> do + let addTable :: (MonadIO m, MonadFail m) => (Integer, Int, Batch, [Batch]) -> String -> m (Integer, Int, Batch, [Batch]) + addTable (crt, crtb, b, bs) tableName = do + liftIO $ log $ fromString $ "add table " ++ tableName + rowsstmt <- allocHandle sql_handle_stmt hdbc + ks <- keyColumns hdbc tableName + + let addRow :: (MonadIO m, MonadFail m) => (Integer, Int, Batch, [Batch], [String]) -> m (Integer, Int, Batch, [Batch], [String]) + addRow (crt', crtb', b', bs', vs') = do + if crt' `mod` 500000 == 0 then liftIO $ hPutStr stderr "." else return () + if crt' `mod` n == 0 && crt' > 0 + then do newVs <- readKeys + let newBatch :: Batch + newBatch = Batch (printf "%08d" (crtb' + 1)) [] + + closedBatch :: Batch + closedBatch = b' { batch_Items = (batch_Items b') ++ [closedItem] } + + closedItem :: BatchItem + closedItem = BatchItem tableName (Just ks) (Just closeWhere) + closeWhere = intercalate " and " ([ c ++ " >= '" ++ v ++ "'" | (c, v) <- zip ks vs'] ++ + [ c ++ " < '" ++ v ++ "'" | (c, v) <- zip ks newVs]) + + return (crt' + 1, crtb' + 1, newBatch, bs' ++ [closedBatch], newVs) + else return (crt' + 1, crtb', b', bs', vs') + + where readKeys :: (MonadIO m, MonadFail m) => m [String] + readKeys = do + liftIO $ allocaBytes maxKeyBuffer + (\ p_value -> + alloca + ( \ p_value_len -> + let readKey k = do + forAllData rowsstmt k sql_char (castPtr p_value) (fromIntegral maxKeyBuffer) p_value_len readChunk "" + readChunk :: String -> IO String + readChunk v = do + len <- peek p_value_len + if len > 0 + then peekCString p_value >>= return . (v ++) + else return v + in + sequence $ map (readKey.fromIntegral) [1.. (length ks)])) + + + selectKeys = "select " ++ colsList ++ " from " ++ (if length schemaName > 0 then schemaName ++ "." else "") ++ tableName ++ " order by " ++ colsList + colsList = intercalate ", " ks + + execDirect rowsstmt selectKeys (fail $ "unknown parameter data wanted for sql: " ++ selectKeys) + liftIO $ log $ fromString $ "reading records from table " ++ tableName + (crt', crtb', b', bs', vs') <- forAllRecords rowsstmt addRow (crt, crtb, b, bs, []) + liftIO $ freeHandle sql_handle_stmt rowsstmt + + return $ if crtb' /= crtb + then + -- if new batch has been created, then we add a new item for the current table + -- to get all records having the key greater with the last keys used for that table + let whereClause = intercalate " and " [c ++ " >= '" ++ v ++ "'" | (c, v) <- zip ks vs'] + item = BatchItem tableName (Just ks) (Just whereClause) + in + (crt', crtb', b' {batch_Items = (batch_Items b') ++ [item]} , bs') + else + -- if no new batch was generated (crtb == crtb') then add this table to the current batch + (crt', crtb', b' { batch_Items = (batch_Items b') ++ [BatchItem tableName Nothing Nothing] }, bs') + liftIO $ log $ fromString "reading tables from source ... " + (_, _, lastBatch, closedBatches) <- withTables hdbc (0, 1, Batch "00000001" [], []) addTable + plan <- initPlan + let plan' = plan { plan_Batches = closedBatches ++ [lastBatch] } + liftIO $ B.putStr $ encode plan') + liftIO $ log $ fromString "Plan generated" + + +-- | extract the names of the columns making a unique key for a given table +keyColumns :: (MonadIO m, MonadFail m) => SQLHDBC -> String -> m [ColumnName] +keyColumns _ _ = return ["ROWID"] -- TODO give a real implementation + +-- | run an action in the current environment on each table name from the current schema, +-- passing an accumulator value; returns the value of the accumulor +withTables :: (MonadFail m, MonadIO m) => SQLHDBC -> a -> (a -> String -> ReaderT ProgramOptions m a) -> ReaderT ProgramOptions m a +withTables hdbc arg f = do + tables_stmt <- allocHandle sql_handle_stmt hdbc + schema <- asks $ dbi_Schema.po_Source + tables tables_stmt Nothing (Just schema) Nothing (Just "TABLE") + let readTableName = liftIO $ allocaBytes 255 + (\ p_tableName -> + alloca + (\ p_tableName_ind -> do + poke p_tableName_ind 0 + _ <- getData tables_stmt 3 sql_char p_tableName 255 p_tableName_ind + tableName_ind <- liftIO $ peek p_tableName_ind + tableName <- if tableName_ind == sql_null_data + then return Nothing + else (liftIO . peekCString . castPtr) p_tableName >>= (return.Just) + return tableName)) + withTableName arg' = do + tableName <- readTableName + maybe (return arg') (f arg') tableName + result <- forAllRecords tables_stmt withTableName arg + liftIO $ freeHandle sql_handle_stmt tables_stmt + return result + +initPlan :: (Monad m) => ReaderT ProgramOptions m TransferPlan +initPlan = TransferPlan + <$> withReaderT po_Source initDatabaseScope + <*> withReaderT po_Dest initDatabaseScope + <*> return [] + +initDatabaseScope :: (Monad m) => ReaderT DBInfo m DatabaseScope +initDatabaseScope = Scope + <$> asks dbi_Datasource + <*> asks dbi_User + <*> asks dbi_Password + <*> asks dbi_Schema
+ src/Main.hs view
@@ -0,0 +1,213 @@+module Main where + +import Prelude hiding (fail) + +import SQL.CLI.ODBC (odbcImplementation) + +import TransferDB(transferDB, TransferOptions(TransferOptions)) +import Options (readPlan, Options(Options, option_Plan, option_Threads, option_Count, option_Drop)) +import qualified Generator as G +import Database.TransferDB.DumpDB (dump, restore, DumpConfig(DumpConfig), RestoreConfig(RestoreConfig)) + +import Database.TransferDB.Commons (ProgramOptions (ProgramOptions), + DBInfo (DBInfo, dbi_Datasource, dbi_User, dbi_Password, dbi_Schema)) + +import CorrectionPlan (generateCorrections, CorrectionConfig(CorrectionConfig)) + + +import Control.Concurrent.STM (newTVar, atomically) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Fail (MonadFail, fail) +import Control.Monad.Trans.Maybe (MaybeT, runMaybeT) +import Control.Monad.Trans.Reader (ReaderT, runReaderT) + +import Control.Logging (setLogLevel, LogLevel(LevelDebug, LevelInfo, LevelWarn, LevelError, LevelOther)) + +import qualified Data.Map.Strict as Map (empty) + + +import System.Clock (Clock(Monotonic), getTime) + +import System.Console.Program (single, interactive, showUsage) +import System.Console.Command (Action, Commands, Tree(Node), io, withNonOption, withOption, command) +import System.Console.Argument (Option, file, natural, string, option) + +import System.IO (hPutStrLn, stdin, stdout, stderr, withFile, IOMode(ReadMode, WriteMode), Handle) + + +main :: IO () +main = do + setLogLevel LevelInfo + (runMaybeT $ single commands) >>= maybe (fail "execution failed") (\ _ -> hPutStrLn stderr "execution finished") + +commands :: Commands (MaybeT IO) +commands = Node (command "transfer-db" "Copy data between 2 databases through ODBC, based on a transfer plan" . io $ do + liftIO $ hPutStrLn stderr "For usage type help" + interactive commands) + [ + Node (command "run" "Run a transfer plan" . (withNonOption file) $ runTransferPlan) [], + Node (command "makePlan" "Create a transfer plan based on info collected from the source db" makeSimplePlan) + [ + Node (command "splitByRows" "makes a batch at each <INT> records" . (withNonOption natural) $ makePlanByRows) [], + Node (command "splitByTables" "makes a batch at each <INT> tables" . (withNonOption natural) $ makePlanByTables) [] + ], + Node (command "correctivePlan" "Generate a new plan and a sql script to correct problems with a previous run" . (withNonOption file) $ generateCorrectionsAction) [], + Node (command "dump" "dumps a database schema to a binary file" dumpAction) [], + Node (command "restore" "restores a database from a a binary dump file created with dump command" restoreAction) [], + Node (command "help" "Show usage information" . io $ liftIO $ showUsage commands) [] + ] + +-- commands + +runTransferPlan :: FilePath -> Action (MaybeT IO) +runTransferPlan planFile = withOption parallelConnections + (\ p -> + withOption runCount + (\ c -> + withOption dropCount + (\ d -> io $ do + plan <- liftIO $ readPlan planFile + runReaderT transferDB (TransferOptions (Options { + option_Plan = plan, + option_Threads = fromIntegral p, + option_Count = fromIntegral c, + option_Drop = fromIntegral d}) odbcImplementation)))) + +makeSimplePlan :: Action (MaybeT IO) +makeSimplePlan = withProgramOptions G.makeSimplePlan + +makePlanByRows :: Integer -> Action (MaybeT IO) +makePlanByRows n = withProgramOptions $ G.makePlanByRows n + +makePlanByTables :: Integer -> Action (MaybeT IO) +makePlanByTables n = withProgramOptions $ G.makePlanByTables n + +dumpAction :: Action (MaybeT IO) +dumpAction = withOption parallelConnections + (\ p -> withSourceDb + (\ dbi -> withOption descriptionOption + (\ description -> withNonOption file + (\ dumpFile -> io $ do + statsVar <- liftIO $ atomically $ newTVar Map.empty -- create a TVar to collect statistics + startTime <- liftIO $ getTime Monotonic -- the start time, to compute the dump rate + runReaderT dump (DumpConfig + (dbi_Datasource dbi) + (dbi_User dbi) + (dbi_Password dbi) + (dbi_Schema dbi) + description + dumpFile + (fromIntegral p) + statsVar + startTime + ))))) + +restoreAction :: Action (MaybeT IO) +restoreAction = withDestinationDb + (\ dbi -> withNonOption file + (\ dumpFile -> io $ runReaderT restore (RestoreConfig + (dbi_Datasource dbi) + (dbi_User dbi) + (dbi_Password dbi) + (dbi_Schema dbi) + dumpFile))) + +generateCorrectionsAction :: (MonadIO m, MonadFail m) => FilePath -> Action m +generateCorrectionsAction file = withOption outputScript + (\ scriptFile -> withOption outputPlan + (\ planFile -> withOption inputLog + (\ log -> io $ do + plan <- liftIO $ readPlan file + result <- liftIO $ withInputFile log + (\ logHandle -> withOutputFile scriptFile + (\ scriptHandle -> withOutputFile planFile + (\ planHandle -> runMaybeT $ runReaderT generateCorrections $ CorrectionConfig plan logHandle scriptHandle planHandle))) + maybe ((liftIO $ hPutStrLn stderr "generateCorrectionPlan failed") >> fail "generateCorrectionPlan failed") return result + ))) + + + + +-- command builders + + +withSourceDb :: (MonadIO m) => (DBInfo -> Action m) -> Action m +withSourceDb f = withOption sourceDbOption + (\ db -> withOption sourceUserOption + (\ user -> withOption sourcePasswordOption + (\ password -> withOption (sourceSchemaOption user) + (\ schema -> f $ DBInfo db user password schema)))) + +withDestinationDb :: (MonadIO m) => (DBInfo -> Action m) -> Action m +withDestinationDb f = withOption destDbOption + (\ db -> withOption destUserOption + (\ user -> withOption destPasswordOption + (\ password -> withOption (destSchemaOption user) + (\ schema -> f $ DBInfo db user password schema)))) + +withProgramOptions :: (MonadIO m) => ReaderT ProgramOptions m () -> Action m +withProgramOptions f = withSourceDb (\ sdbi -> withDestinationDb ( \ ddbi -> io $ runReaderT f $ ProgramOptions sdbi ddbi)) + + +-- Options + +sourceDbOption :: Option String +sourceDbOption = option "d" ["sd", "source-db"] string "" "ODBC datasource name for source database" + +sourceUserOption :: Option String +sourceUserOption = option "u" ["su", "source-user"] string "" "User name to connect to source database" + +sourcePasswordOption :: Option String +sourcePasswordOption = option "p" ["sp", "password"] string "" "Password to connect to source database" + +sourceSchemaOption :: String -> Option String +sourceSchemaOption userName = option "s" ["ss", "schema"] string userName "The schema to connect to in the source database. Defaults to the user name" + +destDbOption :: Option String +destDbOption = option "D" ["dd", "dest-db"] string "<datasource>" "ODBC datasource name for dest database" + +destUserOption :: Option String +destUserOption = option "U" ["du", "dest-user"] string "<username>" "User name to connect to dest database" + +destPasswordOption :: Option String +destPasswordOption = option "P" ["dp", "dest-password"] string "<passoword>" "Password to connect to dest database" + +destSchemaOption :: String -> Option String +destSchemaOption userName = option "S" ["ds", "dest-schema"] string userName "The schema to connect to in the dest database. Defaults to the user name" + +descriptionOption :: Option String +descriptionOption = option "m" ["description"] string "" "The description of the dump file" + +parallelConnections :: Option Integer +parallelConnections = option "t" ["threads"] natural 0 "The numbers of parallel threads to run" + +runCount :: Option Integer +runCount = option "c" ["count", "only"] natural 0 "The number of batches to be run. If 0 or not specified, all batches will be run" + +dropCount :: Option Integer +dropCount = option "x" ["drop", "skip", "ignore"] natural 0 "The number of batches in the plan to skip before starting to run the plan" + +inputLog :: Option FilePath +inputLog = option "i" ["log"] file "-" "The file name of the transfer-db log of the run you need to correct. If not specified or the name is '-', then the log will be read from standard input" + +outputScript :: Option FilePath +outputScript = option "s" ["script"] file "-" "The name of the sql script file that will be generated. If not specified or '-' the script will be written on standard output" + +outputPlan :: Option FilePath +outputPlan = option "p" ["plan"] file "-" "The name of the plan to be generated. If not specified or '-', the plan will be written on standard output" + + +-- | runs an IO action by passing an input file handler to it. If the file name is +-- "-", the input file will be standard input +withInputFile :: FilePath -> (Handle -> IO a) -> IO a +withInputFile fileName f = if fileName == "-" + then f stdin + else withFile fileName ReadMode f + + +-- | runs an IO action by passing an output file handler to it. If the file name is +-- "-", the output file will be the standard output +withOutputFile :: FilePath -> (Handle -> IO a) -> IO a +withOutputFile fileName f = if fileName == "-" + then f stdout + else withFile fileName WriteMode f
+ src/Options.hs view
@@ -0,0 +1,35 @@+{-- | +Module : Options +Description : Plan execution options +Copyright : (c) Mihai Giurgeanu, 2017 +License : GPL-3 +Maintainer : mihai.giurgeanu@gmail.com +Stability : experimental +Portability : Portable +--} +module Options where + +import TransferPlan (TransferPlan) + +import Data.Yaml.Aeson (decodeFileEither, prettyPrintParseException, ParseException) + +import System.IO (hPutStrLn, stderr) +import System.Environment (getProgName) + +readPlan :: FilePath -> IO TransferPlan +readPlan planFile = + let failParse :: ParseException -> IO a + failParse exception = do + hPutStrLn stderr $ "Error parsing the transfer plan file (" ++ planFile ++ ")" + fail $ prettyPrintParseException exception + in do + thePlan <- decodeFileEither planFile + either failParse return thePlan + + +data Options = Options { + option_Plan :: TransferPlan, + option_Threads :: Int, + option_Count :: Int, + option_Drop :: Int + }
+ src/TransferDB.hs view
@@ -0,0 +1,530 @@+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+
+ src/TransferPlan.hs view
@@ -0,0 +1,124 @@+{-- | +Module : TransferPlan +Description : Logic for reading and writing transfer plan configuration options. +Copyright : (c) Mihai Giurgeanu, 2017 +License : GPL-3 +Maintainer : mihai.giurgeanu@gmail.com +Stability : experimental +Portability : Portable +--} + +-- Allow to use Text literals +{-# LANGUAGE OverloadedStrings #-} + +module TransferPlan where + +import Data.Yaml.Aeson (ToJSON(toJSON, toEncoding), + FromJSON(parseJSON), + Value(Object), + object, + (.=), (.:), (.:?)) + +import Data.Aeson (pairs) +import Data.Aeson.Types (typeMismatch) + +import Data.Semigroup ((<>)) + +-- | information about the source and destination data sources, about +-- the data that should be transferred and about how to split this data +-- in batches +data TransferPlan = TransferPlan { + plan_Source :: DatabaseScope, -- ^ defines the source database connection and schema + plan_Destination :: DatabaseScope, -- ^ defines the destination database connection and schema + plan_Batches :: [Batch] -- ^ defines what data should be transferred between the 2 scopes + } + +instance ToJSON TransferPlan where + toJSON x = object ["source" .= plan_Source x, "destination" .= plan_Destination x, "batches" .= plan_Batches x] + toEncoding x = pairs ("source" .= plan_Source x <> "destination" .= plan_Destination x <> "batches" .= plan_Batches x) + +instance FromJSON TransferPlan where + parseJSON (Object x) = TransferPlan + <$> x .: "source" + <*> x .: "destination" + <*> x .: "batches" + + parseJSON invalid = typeMismatch "TransferPlan" invalid + +-- | information for locating the tables in the source and destination databases +data DatabaseScope = Scope { + scope_Db :: String, -- ^ data source name + scope_UserName :: String, -- ^ user name to connect to the data source + scope_Password :: String, -- ^ password to connect to the data source + scope_Schema :: String -- ^ the schema for source or destination tables + } + +instance ToJSON DatabaseScope where + toJSON x = object ["db" .= scope_Db x, "user" .= scope_UserName x, "password" .= scope_Password x, "schema" .= scope_Schema x] + toEncoding x = pairs ("db" .= scope_Db x <> "user" .= scope_UserName x <> "password" .= scope_Password x <> "schema" .= scope_Schema x) + +instance FromJSON DatabaseScope where + parseJSON (Object x) = do + db <- x .: "db" + user' <- x .:? "user" + password' <- x .:? "password" + schema' <- x .:? "schema" + let user = maybe "" id user' + password = maybe "" id password' + schema = maybe user id schema' + return $ Scope db user password schema + + parseJSON invalid = typeMismatch "DatabaseScope" invalid + +data Batch = Batch { + batch_Name :: String, -- ^ a name identifying the batch to the user; it is used in log messages + batch_Items :: [BatchItem] -- ^ each batch item defines what data in a table will be transferred in that batch + } + +instance ToJSON Batch where + toJSON x = object ["name" .= batch_Name x, "items" .= batch_Items x] + toEncoding x = pairs ("name" .= batch_Name x <> "items" .= batch_Items x) + +instance FromJSON Batch where + parseJSON (Object x) = Batch + <$> x .: "name" + <*> x .: "items" + + parseJSON invalid = typeMismatch "Batch" invalid + +data BatchItem = BatchItem { + batch_Table :: String, -- ^ the table name in the source database + batch_OrderBy :: Maybe [ColumnName], -- ^ the columns the query should be ordered by; usually the keys by witch we split the batch + batch_Where :: Maybe WhereCondition -- ^ the where clause to select the records in this batch + } + +instance ToJSON BatchItem where + toJSON x = object ["table" .= batch_Table x, "orderBy" .= batch_OrderBy x, "where" .= batch_Where x] + toEncoding x = pairs ps + where ps = case batch_Where x of + Nothing -> psOrderAndTable + y -> psOrderAndTable <> "where" .= y + psOrderAndTable = case batch_OrderBy x of + Nothing -> psTable + y -> psTable <> "orderBy" .= y + psTable = "table" .= batch_Table x + +instance FromJSON BatchItem where + parseJSON (Object x) = do + table <- x .: "table" + orderBy <- x .:? "orderBy" + whereClause <- x .:? "where" + + let orderBy' = maybe Nothing listOrNothing orderBy + whereClause' = maybe Nothing listOrNothing whereClause + return $ BatchItem table orderBy' whereClause' + + parseJSON invalid = typeMismatch "BatchOption" invalid + +-- | returns Nothing for an empty parameter list +listOrNothing :: [a] -> Maybe [a] +listOrNothing [] = Nothing +listOrNothing x = Just x + +type ColumnName = String +type WhereCondition = String
+ tests/Database/TransferDB/DumpDB/FormatSpec.hs view
@@ -0,0 +1,206 @@+module Database.TransferDB.DumpDB.FormatSpec (spec) where++import Test.Hspec+import Test.QuickCheck++import Database.TransferDB.DumpDB.Format++import Data.Store (Store(size, peek, poke),+ Size(ConstSize, VarSize),+ PeekException,+ encode,+ decode)++import Data.Time.Clock (UTCTime(UTCTime), secondsToDiffTime)+import Data.Time.Calendar (Day(ModifiedJulianDay))+import Data.Word (Word8, Word16, Word32)+import Data.Int (Int16, Int64)+import Foreign.C.Types (CShort(CShort), CLong(CLong))++import Foreign.Marshal (allocaBytes)+import Foreign.Storable (Storable (pokeElemOff))++import qualified Data.ByteString.Char8 as C+import qualified Data.ByteString as B++import System.Endian (getSystemEndianness, Endianness(BigEndian))++import SQL.CLI (SQLINTEGER)++spec :: Spec+spec = do+ describe "Version" $ do+ it "has constant size of 1 byte" $ do+ (size::Size Version) `shouldBe` ConstSize 1+ context "when using value V1" $ do+ it "encodes to a bytestring of length 1 byte" $ do+ (B.length $ encode V1) `shouldBe` 1+ it "encodes to a bytestring that has a single octet with value 1" $ do+ encode V1 `shouldBe` B.pack [1]+ describe "decode" $ do+ it "decodes the encoded value to the same value" $ do+ (decode $ encode V1) `shouldBe` Right V1++ describe "HeaderV1" $ do+ it "has the size given by the size of its components" $ property $+ \ header -> let maxChunkSizeS = sizeOf (size :: Size SQLINTEGER) (hv1_MaxChunkSize header)+ timestampS = sizeOf (size :: Size UTCTime) (hv1_Timestamp header)+ descriptionS = sizeOf (size :: Size C.ByteString) (hv1_Description header)+ in case (size::Size HeaderV1) of+ ConstSize _ -> False+ VarSize f -> f header == maxChunkSizeS + timestampS + descriptionS+ it "encodes to a bytestring with the length equal to the size returned by sizeOf" $ property $+ \ header -> (B.length $ encode (header::HeaderV1)) == sizeOf (size :: Size HeaderV1) header+ describe "decode" $ do+ it "decodes an encoded value to the same value" $ property $+ \ header -> (decode $ encode (header::HeaderV1)) == Right header++ describe "SchemaV1" $ do+ it "encodes to a bytestring with the length equal to the size returned by sizeOf" $ property $+ \ sch -> (B.length $ encode (sch::SchemaV1)) == sizeOf (size :: Size SchemaV1) sch+ describe "decode" $ do+ it "decodes an encoded value to the same value" $ property $+ \ sch -> (decode $ encode (sch::SchemaV1)) == Right sch++ describe "FieldInfoV1" $ do+ it "encodes to a bytestring with the length equal to the size returned by sizeOf" $ property $+ \ fld -> (B.length $ encode (fld::FieldInfoV1)) == sizeOf (size :: Size FieldInfoV1) fld+ describe "decode" $ do+ it "decodes an encoded value to the same value" $ property $+ \ fld -> (decode $ encode (fld::FieldInfoV1)) == Right fld++ describe "RecordIndicator" $ do+ it "has constant size of 1 byte" $ do+ (size::Size RecordIndicator) `shouldBe` ConstSize 1+ context "when using value RI" $ do+ it "encodes to a bytstring of length 1" $ do+ (B.length $ encode RI) `shouldBe` 1+ it "encodes to a bytestring containing a single byte with value 255" $ do+ encode RI `shouldBe` B.pack [255]+ describe "decode" $ do+ it "decodes the encoded value back to the original value" $ do+ (decode $ encode RI) `shouldBe` Right RI + context "when using value EOT" $ do+ it "encodes to a bytstring of length 1" $ do+ (B.length $ encode EOT) `shouldBe` 1+ it "encodes to a bytestring containing a single byte with value 0" $ do+ encode EOT `shouldBe` B.pack [0]+ describe "decode" $ do+ it "decodes the encoded value back to the original value" $ do+ (decode $ encode EOT) `shouldBe` Right EOT++ describe "sizeOf" $ do+ it "should be 1 for any Word8" $ property $+ \ int -> sizeOf (size :: Size Word8) int == 1++ describe "writeVersion" $ do+ it "shouldBe the bytes '11' for V1" $ do+ writeVersion V1 `shouldBe` B.pack [1,1]++ describe "writeHeader" $ do+ it "should write the length of the header as first 2 bytes" $ property $+ \ h -> let h' = writeHeader h+ in+ ((decode (B.take 2 h')) :: Either PeekException Word16) == (Right $ fromIntegral $ sizeOf (size :: Size HeaderV1) h)+ it "should encode the header after the length" $ property $+ \ h -> let h' = writeHeader h+ in+ decode (B.drop 2 h') == Right h+ it "should create a ByteString with the length 2 + the length encoded as the first 2 bytes" $ property $+ \ h -> let h' = writeHeader h+ in+ (Right $ fromIntegral $ (B.length h') - 2) == ((decode (B.take 2 h')) :: Either PeekException Word16)+ describe "writeSchema" $ do+ it "should write the length of the encoded schema as the first 4 bytes" $ property $+ \ s -> let s' = writeSchema s+ in+ ((decode (B.take 4 s')) :: Either PeekException Word32) == (Right $ fromIntegral $ sizeOf (size :: Size SchemaV1) s)+ it "should encode the header after the length" $ property $+ \ s -> let s' = writeSchema s+ in+ decode (B.drop 4 s') == Right s+ it "should create a ByteString with the lengh 4 + the length encoded as the first 4 bytes" $ property $+ \ s -> let s' = writeSchema s+ in+ (Right $ fromIntegral $ (B.length s') - 4) == ((decode (B.take 4 s')) :: Either PeekException Word32)+ describe "writeRI" $ do+ it "should return a bytestring with a sigle byte with value 255" $ do+ writeRI `shouldBe` B.pack [255]+ describe "writeEOT" $ do+ it "should return a bytestring with a single byte with value 255" $ do+ writeEOT `shouldBe` B.pack [0]+ describe "writeNullIndicator" $ do+ it "should write a Null value into a bytestring with only one byte whose value is 0" $ do+ writeNullIndicator Null `shouldBe` B.pack [0]+ it "should write a NotNull indicator into a bytestring with only one byte whose value is 255" $ do+ writeNullIndicator NotNull `shouldBe` B.pack [255]+ describe "writeChunk" $ do+ context "when using 1 byte chunk size field" $ do+ let lenlen = 1+ it "should encode bytes [1, 2, 3, 4] to a bytestring containing bytes [4, 1, 2, 3, 4]" $ do+ allocaBytes 4+ (\ ptr -> do+ pokeElemOff ptr 0 (1::Word8)+ pokeElemOff ptr 1 (2::Word8)+ pokeElemOff ptr 2 (3::Word8)+ pokeElemOff ptr 3 (4::Word8)++ return $ writeChunk lenlen 4 ptr+ ) `shouldReturn` B.pack [4, 1, 2, 3, 4]+ context "when using 2 byte chunk size field" $ do+ let lenlen = 2+ expected = if getSystemEndianness == BigEndian then [0, 4, 1, 2, 3, 4] else [4, 0, 1, 2, 3, 4]+ it ("should encode bytes [1, 2, 3, 4] to a bytestring containing bytes " ++ (show expected)) $ do+ allocaBytes 4+ (\ ptr -> do+ pokeElemOff ptr 0 (1::Word8)+ pokeElemOff ptr 1 (2::Word8)+ pokeElemOff ptr 2 (3::Word8)+ pokeElemOff ptr 3 (4::Word8)++ return $ writeChunk lenlen 4 ptr+ ) `shouldReturn` B.pack expected+ context "when using 4 byte chunk size field" $ do+ let lenlen = 4+ expected = if getSystemEndianness == BigEndian then [0, 0, 0, 4, 1, 2, 3, 4] else [4, 0, 0, 0, 1, 2, 3, 4]+ it ("should encode bytes [1, 2, 3, 4] to a bytestring containing bytes " ++ (show expected)) $ do+ allocaBytes 4+ (\ ptr -> do+ pokeElemOff ptr 0 (1::Word8)+ pokeElemOff ptr 1 (2::Word8)+ pokeElemOff ptr 2 (3::Word8)+ pokeElemOff ptr 3 (4::Word8)++ return $ writeChunk lenlen 4 ptr+ ) `shouldReturn` B.pack expected+ + describe "writePlainBuf" $ do+ it "should encode a buffer containing bytes 1, 2, 3 and 4 into a bytestring containing bytes [1,2,3,4]" $ do+ allocaBytes 4+ (\ ptr -> do+ pokeElemOff ptr 0 (1::Word8)+ pokeElemOff ptr 1 (2::Word8)+ pokeElemOff ptr 2 (3::Word8)+ pokeElemOff ptr 3 (4::Word8)+ + return $ writePlainBuf ptr 4+ ) `shouldReturn` B.pack [1, 2, 3, 4]+ + +instance Arbitrary HeaderV1 where+ arbitrary = HeaderV1 <$> arbitrary <*> arbitrary <*> arbitrary++instance Arbitrary SchemaV1 where+ arbitrary = SchemaV1 <$> arbitrary <*> arbitrary <*> (listOf arbitrary)++instance Arbitrary FieldInfoV1 where+ arbitrary = FieldInfoV1 <$> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary++instance Arbitrary UTCTime where+ arbitrary = do+ day <- arbitrary+ seconds <- choose (0, 86400)+ return $ UTCTime (ModifiedJulianDay day) (secondsToDiffTime seconds)++instance Arbitrary C.ByteString where+ arbitrary = listOf arbitrary >>= return.C.pack
+ tests/Database/TransferDB/DumpDBSpec.hs view
@@ -0,0 +1,158 @@+module Database.TransferDB.DumpDBSpec (spec) where++import Database.TransferDB.DumpDB+import Database.TransferDB.DumpDB.Format++import Test.Hspec+import Test.QuickCheck++import Control.Monad.Trans.Reader (runReaderT)++spec :: Spec+spec = do+ describe "octetLengthOfChunkSize" $ do+ context "bufferLength is 256" $ do+ let bufferSize = 256+ it "returns 1 for unknown maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined Nothing undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 0 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 0) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 1 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 1) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 128 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 128) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 255 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 255) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 256 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 256) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 257 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 257) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 65535 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65535) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 65536 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65536) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 65537 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65537) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 2*1024*1024 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just $ 2*1024*1024) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ context "bufferLength is 65536" $ do+ let bufferSize = 65536+ it "returns 2 for unknown maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined Nothing undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 1 for 0 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 0) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 1 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 1) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 128 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 128) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 255 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 255) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 256 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 256) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 2 for 257 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 257) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65535 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65535) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65536 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65536) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65537 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65537) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 2*1024*1024 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just $ 2*1024*1024) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ context "bufferLength is 2*1024*1024" $ do+ let bufferSize = 2*1024*1024+ it "returns 4 for unknown maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined Nothing undefined undefined undefined undefined)) `shouldBe` Just (4 :: Int)+ it "returns 1 for 0 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 0) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 1 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 1) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 128 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 128) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 255 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 255) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 256 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 256) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 2 for 257 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 257) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65535 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65535) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65536 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65536) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 4 for 65537 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65537) undefined undefined undefined undefined)) `shouldBe` Just (4 :: Int)+ it "returns 4 for 2*1024*1024 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just $ 2*1024*1024) undefined undefined undefined undefined)) `shouldBe` Just (4 :: Int)+ + context "bufferLength is 257" $ do+ let bufferSize = 257+ it "returns 1 for unknown maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined Nothing undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 0 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 0) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 1 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 1) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 128 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 128) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 255 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 255) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 256 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 256) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 257 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 257) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 65535 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65535) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 65536 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65536) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 65537 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65537) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 2*1024*1024 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just $ 2*1024*1024) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ context "bufferLength is 65537" $ do+ let bufferSize = 65537+ it "returns 2 for unknown maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined Nothing undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 1 for 0 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 0) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 1 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 1) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 128 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 128) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 255 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 255) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 256 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 256) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 2 for 257 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 257) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65535 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65535) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65536 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65536) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65537 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65537) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 2*1024*1024 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just $ 2*1024*1024) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ context "bufferLength is 2*1024*1024 + 1" $ do+ let bufferSize = 2*1024*1024 + 1+ it "returns 4 for unknown maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined Nothing undefined undefined undefined undefined)) `shouldBe` Just (4 :: Int)+ it "returns 1 for 0 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 0) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 1 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 1) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 128 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 128) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 255 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 255) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 1 for 256 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 256) undefined undefined undefined undefined)) `shouldBe` Just (1 :: Int)+ it "returns 2 for 257 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 257) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65535 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65535) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 2 for 65536 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65536) undefined undefined undefined undefined)) `shouldBe` Just (2 :: Int)+ it "returns 4 for 65537 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just 65537) undefined undefined undefined undefined)) `shouldBe` Just (4 :: Int)+ it "returns 4 for 2*1024*1024 maximum character length" $ do+ runReaderT octetLengthOfChunkSize (DumpFieldSpec undefined undefined undefined bufferSize undefined undefined undefined (FieldInfoV1 undefined undefined undefined (Just $ 2*1024*1024) undefined undefined undefined undefined)) `shouldBe` Just (4 :: Int)
+ tests/hspec-tests.hs view
@@ -0,0 +1,2 @@+-- autodiscovery of HSpec tests+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ transfer-db.cabal view
@@ -0,0 +1,82 @@+name: transfer-db+version: 0.3.1.0+-- synopsis:+description: Simple SQL/CLI application that transfers data between 2 databases.+homepage: http://hub.darcs.net/mihaigiurgeanu/transfer-db+license: BSD3+license-file: LICENSE+author: Mihai Giurgeanu+maintainer: mihai.giurgeanu@gmail.com+copyright: 2017 Mihai Giurgeanu+category: Database+build-type: Simple+cabal-version: >=1.10+extra-source-files: README.md,+ ChangeLog++library+ hs-source-dirs: lib+ default-language: Haskell2010+ exposed-modules: Database.TransferDB.DumpDB,+ Database.TransferDB.DumpDB.Format,+ Database.TransferDB.Commons,+ Control.Monad.Trans.StringError+ build-depends: base >= 4.7 && < 5,+ sqlcli,+ sqlcli-odbc,+ transformers >= 0.5 && < 0.6,+ bytestring >= 0.10.8 && < 0.11,+ logging >= 3.0 && < 4,+ store >= 0.4.3 && < 0.5,+ store-core >= 0.4.1 && < 0.5,+ th-utilities >= 0.2.0.1 && < 0.3,+ time >= 1.8 && < 1.9,+ text >= 0.11.3.1,+ monad-control >=0.3.2.3,+ stm >= 2.4 && < 2.5,+ temporary >= 1.1 && < 1.3,+ containers >= 0.5 && < 0.6,+ clock >= 0.7 && < 0.8+ +executable transfer-db+ hs-source-dirs: src+ main-is: Main.hs+ other-modules: TransferPlan, Options, TransferDB, Generator, CorrectionPlan+ default-language: Haskell2010+ ghc-options: -threaded -rtsopts -O2+ build-depends: base >= 4.7 && < 5,+ transformers >= 0.5 && < 0.6,+ yaml >= 0.8.22 && < 0.9,+ aeson >= 1.0,+ console-program >= 0.4.2.0 && < 0.5,+ bytestring >= 0.10.8 && < 0.11,+ logging >= 3.0 && < 4,+ time >= 1.8 && < 1.9,+ stm >= 2.4 && < 2.5,+ containers >= 0.5 && < 0.6,+ clock >= 0.7 && < 0.8,+ monad-control >=0.3.2.3,+ cassava >= 0.5.0.0 && < 0.6,+ sqlcli,+ sqlcli-odbc,+ transfer-db++test-suite hspec-tests+ hs-source-dirs: tests+ type: exitcode-stdio-1.0+ main-is: hspec-tests.hs+ ghc-options: -threaded+ other-modules: Database.TransferDB.DumpDB.FormatSpec,+ Database.TransferDB.DumpDBSpec+ default-language: Haskell2010+ build-depends: base >= 4.7 && < 5,+ hspec >= 2.4.4 && < 2.5,+ QuickCheck >= 2.10.1 && < 2.11,+ store >= 0.4.3 && < 0.5,+ time >= 1.8 && < 1.9,+ bytestring >= 0.10.8 && < 0.11,+ transformers >= 0.5 && < 0.6,+ cpu >= 0.1.2 && < 0.2,+ transfer-db,+ sqlcli+