diff --git a/ChangeLog b/ChangeLog new file mode 100644 --- /dev/null +++ b/ChangeLog @@ -0,0 +1,81 @@ +2018-03-10 Mihai Giurgeanu + + * stack.yaml: update dependencies in to work with remote repositories + + * Release 0.3.1.0 + +2017-12-29 Mihai Giurgeanu + + * src/Main.hs (commands): add correctivePlan command + +2017-12-27 Mihai Giurgeanu + + * src/Options.hs (readPlan): returns the Plan only + +2017-12-19 Mihai Giurgeanu + + * 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 + + * src/TransferDB.hs (transferDB): process batches in parallel + + * src/Main.hs (runTransferPlan): add multithreading command line option + +2017-11-26 Mihai Giurgeanu + + * lib/Database/TransferDB/Commons.hs (withEnvConnection): add withEnvConnection and withEnvConnection' + +2017-11-24 Mihai Giurgeanu + + * src/Main.hs (commands): add parallel threads option to dump command + +2017-11-07 Mihai Giurgeanu + + * src/Main.hs (commands): add dump and restore commands to transfer-db + executable. + +2017-11-03 Mihai Giurgeanu + + * lib/Database/TransferDB/DumpDB.hs: add database agnostic dump library + +2017-11-02 Mihai Giurgeanu + + * src/TransferDB.hs: use simple logging framework instead 'hPutStrLn stderr' + +2017-10-27 Mihai Giurgeanu + + * (release 0.3.0.0): subcommands in program interface, generate plans + +2017-09-21 Mihai Giurgeanu + + * (change summary): add functionality to generate plans + +2017-09-19 Mihai Giurgeanu + + * src/Main.hs (main): change program interface to accept commands (run, help) + +2017-09-19 Mihai Giurgeanu + + * (release 0.2.0.0): transfer plan, work in restartable batches + +2017-09-18 Mihai Giurgeanu + + * 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 + + * src/Main.hs (transferDB): set autocommit off + (transferTable): add transaction demarcaion on each transferred table + +2017-09-15 Mihai Giurgeanu + + * 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 + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/lib/Control/Monad/Trans/StringError.hs b/lib/Control/Monad/Trans/StringError.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Monad/Trans/StringError.hs @@ -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 #-} diff --git a/lib/Database/TransferDB/Commons.hs b/lib/Database/TransferDB/Commons.hs new file mode 100644 --- /dev/null +++ b/lib/Database/TransferDB/Commons.hs @@ -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 diff --git a/lib/Database/TransferDB/DumpDB.hs b/lib/Database/TransferDB/DumpDB.hs new file mode 100644 --- /dev/null +++ b/lib/Database/TransferDB/DumpDB.hs @@ -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 () + + diff --git a/lib/Database/TransferDB/DumpDB/Format.hs b/lib/Database/TransferDB/DumpDB/Format.hs new file mode 100644 --- /dev/null +++ b/lib/Database/TransferDB/DumpDB/Format.hs @@ -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 . 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 + diff --git a/src/CorrectionPlan.hs b/src/CorrectionPlan.hs new file mode 100644 --- /dev/null +++ b/src/CorrectionPlan.hs @@ -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 diff --git a/src/Generator.hs b/src/Generator.hs new file mode 100644 --- /dev/null +++ b/src/Generator.hs @@ -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 diff --git a/src/Main.hs b/src/Main.hs new file mode 100644 --- /dev/null +++ b/src/Main.hs @@ -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 records" . (withNonOption natural) $ makePlanByRows) [], + Node (command "splitByTables" "makes a batch at each 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 "" "ODBC datasource name for dest database" + +destUserOption :: Option String +destUserOption = option "U" ["du", "dest-user"] string "" "User name to connect to dest database" + +destPasswordOption :: Option String +destPasswordOption = option "P" ["dp", "dest-password"] string "" "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 diff --git a/src/Options.hs b/src/Options.hs new file mode 100644 --- /dev/null +++ b/src/Options.hs @@ -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 + } diff --git a/src/TransferDB.hs b/src/TransferDB.hs new file mode 100644 --- /dev/null +++ b/src/TransferDB.hs @@ -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 + diff --git a/src/TransferPlan.hs b/src/TransferPlan.hs new file mode 100644 --- /dev/null +++ b/src/TransferPlan.hs @@ -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 diff --git a/tests/Database/TransferDB/DumpDB/FormatSpec.hs b/tests/Database/TransferDB/DumpDB/FormatSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Database/TransferDB/DumpDB/FormatSpec.hs @@ -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 diff --git a/tests/Database/TransferDB/DumpDBSpec.hs b/tests/Database/TransferDB/DumpDBSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Database/TransferDB/DumpDBSpec.hs @@ -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) diff --git a/tests/hspec-tests.hs b/tests/hspec-tests.hs new file mode 100644 --- /dev/null +++ b/tests/hspec-tests.hs @@ -0,0 +1,2 @@ +-- autodiscovery of HSpec tests +{-# OPTIONS_GHC -F -pgmF hspec-discover #-} diff --git a/transfer-db.cabal b/transfer-db.cabal new file mode 100644 --- /dev/null +++ b/transfer-db.cabal @@ -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 +