clickhouse-haskell (empty) → 0.1.1.0
raw patch · 34 files changed
+5417/−0 lines, 34 filesdep +HUnitdep +QuickCheckdep +aesonsetup-changed
Dependencies added: HUnit, QuickCheck, aeson, array, async, attoparsec, base, binary, bytestring, bytestring-to-vector, call-stack, clickhouse-haskell, containers, data-default-class, data-dword, exceptions, filepath, hashable, hashmap, haxl, hspec, http-client, io-streams, monad-loops, monad-parallel, mtl, network, network-ip, network-simple, parsec, resource-pool, split, streaming-commons, text, time, transformers, tz, unix-time, unordered-containers, uri-encode, uuid, vector, word8
Files
- ChangeLog.md +5/−0
- LICENSE +21/−0
- README.md +291/−0
- Setup.hs +2/−0
- clickhouse-haskell.cabal +162/−0
- src/Database/ClickHouseDriver.hs +14/−0
- src/Database/ClickHouseDriver/Block.hs +143/−0
- src/Database/ClickHouseDriver/CBits/bigint.c +41/−0
- src/Database/ClickHouseDriver/CBits/datetime.c +49/−0
- src/Database/ClickHouseDriver/CBits/varuint.c +65/−0
- src/Database/ClickHouseDriver/Client.hs +346/−0
- src/Database/ClickHouseDriver/ClientProtocol.hs +68/−0
- src/Database/ClickHouseDriver/Column.hs +998/−0
- src/Database/ClickHouseDriver/Connection.hs +554/−0
- src/Database/ClickHouseDriver/Defines.hs +130/−0
- src/Database/ClickHouseDriver/Error.hs +804/−0
- src/Database/ClickHouseDriver/HTTP.hs +35/−0
- src/Database/ClickHouseDriver/HTTP/Client.hs +279/−0
- src/Database/ClickHouseDriver/HTTP/Connection.hs +81/−0
- src/Database/ClickHouseDriver/HTTP/Helpers.hs +94/−0
- src/Database/ClickHouseDriver/HTTP/Types.hs +46/−0
- src/Database/ClickHouseDriver/IO/BufferedReader.hs +215/−0
- src/Database/ClickHouseDriver/IO/BufferedWriter.hs +139/−0
- src/Database/ClickHouseDriver/Include/bigint.h +13/−0
- src/Database/ClickHouseDriver/Include/datetime.h +10/−0
- src/Database/ClickHouseDriver/Include/varuint.h +12/−0
- src/Database/ClickHouseDriver/Pool.hs +68/−0
- src/Database/ClickHouseDriver/QueryProcessingStage.hs +11/−0
- src/Database/ClickHouseDriver/ServerProtocol.hs +81/−0
- src/Database/ClickHouseDriver/Types.hs +339/−0
- test/Spec.hs +11/−0
- test/Test/ColumnSpec.hs +137/−0
- test/Test/HTTPSpec.hs +70/−0
- test/Test/IO.hs +83/−0
+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Revision history for clickhouse-haskell++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2020 EMQX++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ README.md view
@@ -0,0 +1,291 @@+**Clickhouse-haskell**+======================+ClickHouse Haskell Driver with HTTP and native (TCP) interface support.+Support both insert and ordinary query.+This project has been heavily referenced from the python version. Link: https://github.com/mymarilyn/clickhouse-driver++**Features**+========++* External Data for query processing+* Types support+ * [U]Int8/16/32/64+ * String/FixedString(N)+ * Array(T)+ * Nullable(T)+ * Decimal+ * SimpleAggregateFunction(F, T)+ * Tuple(T1, T2, ...)+ * Date/DateTime('timezone')/DateTime64('timezone')+ * Enum8/16+ * Nested+ * LowCardinality(T)+ * UUID+* Query progress information.+* Block by block results streaming.+* Reading query profile info.+* Receiving server logs.+* Applying Haxl(Concurrency library developed by Facebook) for caching query requests and concurrent querying processes.++**Usage of the HTTP Client**+=====++In the HTTP client, data can be fetched from Clickhouse server in the format of pure text string, and structured JSON in which user can read the value according to a given key.++## **Example of data fetch using http client**+```Haskell+import Database.ClickHouseDriver+import qualified Data.Text.IO as TIO++main :: IO()+main = do+ env <- httpClient "default" "" --username and password+ showtables <- runQuery env (getText "SHOW TABLES")+ TIO.putStr showtables+```+Result+```+ Fetching 1 queries.+ array0+ array1+ array_t+ array_test+ array_test2+ big+ cardin+ crd+ crd2+ dt+ int_test+ ip+ t+ tande+ tande2+ tande3+ test_table+ test_table2+ test_table3+ test_table4+ test_table5+ test_table6+ tuple+ tuple2+```+```Haskell++ puretext <- runQuery env (getText "SELECT * FROM test_table")+ TIO.putStr puretext+```+stdout: +```+ Fetching 1 queries.+ 9987654321 Suzuki 12507 [667]+ 9987654321 Suzuki 12507 [667]+ 0000000001 JOHN 1557 [45,45,45]+ 1234567890 CONNOR 533 [1,2,3,4]+ 3543364534 MARRY 220 [0,1,2,3,121,2]+ 2258864346 JAME 4452 [42,-10988,66,676,0]+ 0987654321 Connan 9984 [24]+ 0987654321 Connan 9984 [24]+ 9987654321 Suzuki 12507 [667]+```+```Haskell + json <- runQuery env (getJSON "SELECT * FROM test_table")+ print json+```+stdout:+```Haskell+ Right [fromList [("numArray",Array [Number 45.0,Number 45.0,Number 45.0]),("item", String "JOHN"),("id",String "0000000001"),("number",Number 1557.0)],fromList [("numArray",Array [Number 1.0,Number 2.0,Number 3.0,Number 4.0]),("item",String "CONNOR"),("id",String "1234567890"),("number",Number 533.0)],fromList [("numArray",Array [Number 0.0,Number 1.0,Number 2.0,Number 3.0,Number 121.0,Number 2.0]),("item",String "MARRY"),("id",String "3543364534"),("number",Number 220.0)],fromList [("numArray",Array [Number 42.0,Number -10988.0,Number 66.0,Number 676.0,Number 0.0]),("item",String "JAME"),("id",String "2258864346"),("number",Number 4452.0)],fromList [("numArray",Array [Number 24.0]),("item",String "Connan"),("id",String "0987654321"),("number",Number 9984.0)],fromList [("numArray",Array [Number 24.0]),("item",String "Connan"),("id",String "0987654321"),("number",Number 9984.0)],fromList [("numArray",Array [Number 667.0]),("item",String "Suzuki"),("id",String "9987654321"),("number",Number 12507.0)],fromList [("numArray",Array [Number 667.0]),("item",String "Suzuki"),("id",String "9987654321"),("number",Number 12507.0)],fromList [("numArray",Array [Number 667.0]),("item",String "Suzuki"),("id",String "9987654321"),("number",Number 12507.0)]]+```++There is also a built-in Clickhouse type for user to send data in rows to Clickhouse server.++## **Algebraic data type for the Clickhouse types** ++```Haskell +data ClickhouseType+ = CKBool Bool+ | CKInt8 Int8+ | CKInt16 Int16+ | CKInt32 Int32+ | CKInt64 Int64+ | CKUInt8 Word8+ | CKUInt16 Word16+ | CKUInt32 Word32+ | CKUInt64 Word64+ | CKString ByteString+ | CKFixedLengthString Int ByteString+ | CKTuple (Vector ClickhouseType)+ | CKArray (Vector ClickhouseType)+ | CKDecimal32 Float+ | CKDecimal64 Float+ | CKDecimal128 Float+ | CKIPv4 IP4+ | CKIPv6 IP6+ | CKDate {+ year :: !Integer,+ month :: !Int,+ day :: !Int + }+ | CKNull+ deriving (Show, Eq)+```++## **Example of sending data from memory**+```Haskell+main = do+ env <- httpClient "default" "12345612341"+ create <- exec "CREATE TABLE test (x Int32) ENGINE = Memory" env+ print create+ isSuccess <- insertOneRow "test" [CKInt32 100] env+ print isSuccess+ result <- runQuery env (getText "select * from test")+ TIO.putStr result+```+stdout:+```+Right "Inserted successfully"+Right "Inserted successfully"+Fetching 1 queries.+100+```++## **Example of sending data from CSV**+```Haskell+main :: IO()+main = do+ env <- httpClient "default" "12345612341"+ isSuccess <- insertFromFile "test_table" CSV "./test/example.csv" env+ putStr (case isSuccess of+ Right y -> y+ Left x -> CL8.unpack x)+ query <- runQuery env (getText "SELECT * FROM test_table")+ TIO.putStr query+```+where in example.csv+```CSV+0000000011,Bob,123,'[1,2,3]'+0000000012,Bob,124,'[4,5,6]'+0000000013,Bob,125,'[7,8,9,10]'+0000000014,Bob,126,'[11,12,13]'+```++stdout:++```+0000000010 Alice 123 [1,2,3]+0000000010 Alice 123 [1,2,3]+0000000010 Alice 123 [1,2,3]+0000000010 Alice 123 [1,2,3]+0000000011 Bob 123 [1,2,3]+0000000012 Bob 124 [4,5,6]+0000000013 Bob 125 [7,8,9,10]+0000000014 Bob 126 [11,12,13]+```++**Usage of the Native(TCP) interface**+==========================++## **Ping**+```Haskell+ conn <- defaultClient+ Database.ClickHouseDriver.ping conn+```+stdout:+```+Just "PONG!"+```++## **Example of making query with the native interface**+```Haskell+main :: IO ()+main = do+ env <- defaultClient --localhost 9000+ res <- query "SHOW TABLES" env+ print res+```+stdout:+``` Haskell+[[CKString "test"],[CKString "test_table"],[CKString "test_table2"]]+```++## **Example of making insert query with the native interface**+```+conn <- defaultClient+insertMany conn "INSERT INTO crd VALUES"+ [+ [CKString "123", CKString "hi"],+ [CKString "456", CKString "lo"]+ ]+```+In the terminal interface of clickhouse it will show:+```+id──┬─card─┐+│ 123 │ hi │+│ 456 │ lo │+```+## **Use of Haxl for concurrency**+We can perform multiple fetches concurrently like this:+```Haskell+queryTests :: GenHaxl u w (V.Vector (V.Vector ClickhouseType))+queryTests = do+ one <- fetch "SELECT * FROM UUID_test"+ two <- fetch "SELECT * FROM array_t"+ three <- fetch "SHOW DATABASES"+ four <- fetch "SHOW TABLES"+ return $ V.concat [one, two ,three, four]++```+```+Fetching 4 queries.+[[CKString "417ddc5d-e556-4d27-95dd-a34d84e46a50"],...+```++## **Stream profile and process infomation**++The native interface supports reading infomations coming from server. Originally they come with the queried data wrapped in the algebraic data types:++```Haskell+data CKResult = CKResult+ { query_result :: Vector (Vector ClickhouseType),+ query_info :: {-# UNPACK #-} ! QueryInfo+ }++data QueryInfo = QueryInfo + { profile_info :: {-# UNPACK #-} !BlockStreamProfileInfo,+ progress :: {-# UNPACK #-} !Progress,+ elapsed :: {-# UNPACK #-} !Word+ }++ data BlockStreamProfileInfo = ProfileInfo+ { number_rows :: {-# UNPACK #-} !Word,+ blocks :: {-# UNPACK #-} !Word,+ number_bytes :: {-# UNPACK #-} !Word,+ applied_limit :: {-# UNPACK #-} !Bool,+ rows_before_limit :: {-# UNPACK #-} !Word,+ calculated_rows_before_limit :: {-# UNPACK #-} !Bool+ }++data Progress = Prog+ { rows :: {-# UNPACK #-} !Word,+ bytes :: {-# UNPACK #-} !Word,+ total_rows :: {-# UNPACK #-} !Word,+ written_rows :: {-# UNPACK #-} !Word,+ written_bytes :: {-# UNPACK #-} !Word+ }+```+One can use executeWithInfo to get results that come with those information.+For example:+```Haskell+main = do+ conn <- defaultClient+ res <- executeWithInfo "show databases" conn+ print $ query_result res+ print $ query_info res+```+The code above prints:+```Haskell+[[CKString "_temporary_and_external_tables"],[CKString "default"],[CKString "system"]]++QueryInfo {profile_info = ProfileInfo {number_rows = 3, blocks = 1, number_bytes = 4224, applied_limit = True, rows_before_limit = 0, calculated_rows_before_limit = True}, progress = Prog {rows = 3, bytes = 331, total_rows = 0, written_rows = 0, written_bytes = 0}, elapsed = 0}+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ clickhouse-haskell.cabal view
@@ -0,0 +1,162 @@+cabal-version: 1.12+name: clickhouse-haskell+version: 0.1.1.0+synopsis: A Haskell library as database client for Clickhouse+homepage: https://github.com/MaboroshiChan/clickhouse-haskell/blob/master/README.md+bug-reports: https://github.com/2049foundation/clickhouse-haskell/issues+author: Shi You+maintainer: youshi@emqx.io+copyright: Copyright (c) 2020-present, EMQ X, Inc.+category: Database+license: MIT+license-file: LICENSE+build-type: Simple+extra-source-files:+ README.md+ ChangeLog.md+description: + Clickhouse-Haskell is a library for making connection with the server of+ column-oriented DBMS, Clickhouse. This library supports SELECT, INSERT and other query commands.++ Clickhouse-Haskell is also implemented with Haxl, an concurrent data accessing API developed by Facebook,+ for efficient interaction with the server.++ For more detail, see: https://github.com/2049foundation/clickhouse-haskell#readme++source-repository head+ type: git+ location: https://github.com/MaboroshiChan/clickhouse-haskell++library+ exposed-modules:+ Database.ClickHouseDriver+ Database.ClickHouseDriver.HTTP+ Database.ClickHouseDriver.HTTP.Helpers+ Database.ClickHouseDriver.HTTP.Client+ Database.ClickHouseDriver.HTTP.Types+ Database.ClickHouseDriver.HTTP.Connection+ Database.ClickHouseDriver.Error+ Database.ClickHouseDriver.QueryProcessingStage+ Database.ClickHouseDriver.Defines+ Database.ClickHouseDriver.ServerProtocol+ Database.ClickHouseDriver.ClientProtocol+ Database.ClickHouseDriver.Block+ Database.ClickHouseDriver.Client+ Database.ClickHouseDriver.Connection+ Database.ClickHouseDriver.Types+ Database.ClickHouseDriver.Column+ Database.ClickHouseDriver.IO.BufferedWriter+ Database.ClickHouseDriver.IO.BufferedReader+ Database.ClickHouseDriver.Pool+ other-modules:+ Paths_clickhouse_haskell+ hs-source-dirs:+ src+ build-depends:+ base >=4.12 && <5,+ containers >= 0.5.7 && < 0.6,+ array >= 0.5.1 && < 0.6,+ time >= 1.9 && < 1.12,+ transformers >= 0.5.2 && < 0.6,+ aeson >= 1.4.1 && < 1.6,+ attoparsec >= 0.13.2 && < 0.14,+ bytestring >= 0.10.8 && < 0.11,+ binary >= 0.8.3 && < 0.9,+ hashable >= 1.2.7 && < 1.4,+ text >= 1.2.4 && < 1.3,+ unordered-containers >= 0.2.13 && < 0.3,+ vector >= 0.12.1 && < 0.13,+ async >= 2.2.2 && < 2.3,+ bytestring-to-vector >= 0.3.0 && < 0.4,+ call-stack >= 0.2.0 && < 0.3,+ data-default-class >= 0.1.2 && < 0.2,+ data-dword >= 0.3.2 && < 0.4,+ exceptions >= 0.10.4 && < 0.11,+ mtl >= 2.2.2 && < 2.3,+ filepath >= 1.4.1 && < 1.5,+ hashmap >= 1.3.3 && < 1.4,+ haxl >= 2.3.0 && < 2.4,+ http-client >= 0.6.4 && < 0.7,+ network >= 3.1.0 && < 3.2,+ parsec >= 3.1.14 && < 3.2,+ streaming-commons >= 0.2.2 && < 0.3,+ io-streams >= 1.5.2 && < 1.6,+ monad-loops >= 0.4.3 && < 0.5,+ monad-parallel >= 0.7.2 && < 0.8,+ network-ip >= 0.3.0 && < 0.4,+ split >= 0.2.3 && < 0.3,+ network-simple >= 0.4.5 && < 0.5,+ resource-pool >= 0.2.3 && < 0.3,+ tz >= 0.1.3 && < 0.2,+ unix-time >= 0.4.7 && < 0.5,+ uri-encode >= 1.5.0 && < 1.6,+ uuid >= 1.3.13 && < 1.4,+ word8 >= 0.1.3 && < 0.2+ default-language: Haskell2010+ c-sources: ./src/Database/ClickHouseDriver/CBits/varuint.c+ , ./src/Database/ClickHouseDriver/CBits/datetime.c+ , ./src/Database/ClickHouseDriver/CBits/bigint.c+ include-dirs: ./src/Database/ClickHouseDriver/Include + includes: varuint.h+ , datetime.h+ , bigint.h+ install-includes: varuint.h+ , datetime.h+ , bigint.h++test-suite clickhouse-haskell-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Test.IO+ Test.ColumnSpec+ Test.HTTPSpec+ Paths_clickhouse_haskell+ hs-source-dirs:+ test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ clickhouse-haskell, + base >= 4.12 && <5,+ QuickCheck >= 2.13 && <3,+ hspec >= 2.0 && <3,+ HUnit >= 1.6 && <1.7,+ containers >= 0.5.7 && < 0.6,+ array >= 0.5.1 && < 0.6,+ time >= 1.9 && < 1.12,+ transformers >= 0.5.2 && < 0.6,+ aeson >= 1.4.1 && < 1.6,+ attoparsec >= 0.13.2 && < 0.14,+ bytestring >= 0.10.8 && < 0.11,+ binary >= 0.8.3 && < 0.9,+ hashable >= 1.2.7 && < 1.4,+ text >= 1.2.4 && < 1.3,+ unordered-containers >= 0.2.13 && < 0.3,+ vector >= 0.12.1 && < 0.13,+ async >= 2.2.2 && < 2.3,+ bytestring-to-vector >= 0.3.0 && < 0.4,+ call-stack >= 0.2.0 && < 0.3,+ data-default-class >= 0.1.2 && < 0.2,+ data-dword >= 0.3.2 && < 0.4,+ exceptions >= 0.10.4 && < 0.11,+ mtl >= 2.2.2 && < 2.3,+ filepath >= 1.4.1 && < 1.5,+ hashmap >= 1.3.3 && < 1.4,+ haxl >= 2.3.0 && < 2.4,+ http-client >= 0.6.4 && < 0.7,+ network >= 3.1.0 && < 3.2,+ parsec >= 3.1.14 && < 3.2,+ streaming-commons >= 0.2.2 && < 0.3,+ io-streams >= 1.5.2 && < 1.6,+ monad-loops >= 0.4.3 && < 0.5,+ monad-parallel >= 0.7.2 && < 0.8,+ network-ip >= 0.3.0 && < 0.4,+ split >= 0.2.3 && < 0.3,+ network-simple >= 0.4.5 && < 0.5,+ resource-pool >= 0.2.3 && < 0.3,+ tz >= 0.1.3 && < 0.2,+ unix-time >= 0.4.7 && < 0.5,+ uri-encode >= 1.5.0 && < 1.6,+ uuid >= 1.3.13 && < 1.4,+ word8 >= 0.1.3 && < 0.2+ default-language: Haskell2010
+ src/Database/ClickHouseDriver.hs view
@@ -0,0 +1,14 @@++module Database.ClickHouseDriver( + module Database.ClickHouseDriver.Client,+ module Database.ClickHouseDriver.Defines,+ module Database.ClickHouseDriver.Connection,+ module Database.ClickHouseDriver.Pool,+ module Database.ClickHouseDriver.Column+) where++import Database.ClickHouseDriver.Client+import Database.ClickHouseDriver.Defines+import Database.ClickHouseDriver.Connection+import Database.ClickHouseDriver.Pool+import Database.ClickHouseDriver.Column
+ src/Database/ClickHouseDriver/Block.hs view
@@ -0,0 +1,143 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++----------------------------------------------------------------------+++{-# LANGUAGE FlexibleContexts #-}++-- | This module provides functions for handling data streaming communications.+-- For internal use only.+-- ++module Database.ClickHouseDriver.Block+ ( BlockInfo (..),+ writeInfo,+ readInfo,+ readBlockInputStream,+ Block (..),+ defaultBlockInfo,+ writeBlockOutputStream,+ defaultBlock+ )+where++import Database.ClickHouseDriver.Column+ ( ClickhouseType, readColumn, writeColumn )+import Database.ClickHouseDriver.Defines as Defines+ ( _DBMS_MIN_REVISION_WITH_BLOCK_INFO )+import Database.ClickHouseDriver.Types+ ( writeBlockInfo,+ Block(..),+ BlockInfo(..),+ Context(Context),+ ServerInfo(revision) )+import Database.ClickHouseDriver.IO.BufferedReader+ ( readBinaryInt32,+ readBinaryStr,+ readBinaryUInt8,+ readVarInt,+ Reader )+import Database.ClickHouseDriver.IO.BufferedWriter+ ( writeBinaryInt32,+ writeBinaryStr,+ writeBinaryUInt8,+ writeVarUInt,+ Writer )+import Data.ByteString ( ByteString )+import Data.ByteString.Builder ( Builder )+import Data.Vector (Vector)+import Data.Vector ((!))+import qualified Data.Vector as V+--Debug+--import Debug.Trace++defaultBlockInfo :: BlockInfo+defaultBlockInfo =+ Info+ { is_overflows = False,+ bucket_num = -1+ }++defaultBlock :: Block+defaultBlock =+ ColumnOrientedBlock {+ columns_with_type=V.empty,+ cdata = V.empty,+ info = defaultBlockInfo+ }++-- | write block informamtion to string builder+writeInfo :: BlockInfo->Writer Builder+writeInfo (Info is_overflows bucket_num) = do+ writeVarUInt 1+ writeBinaryUInt8 (if is_overflows then 1 else 0)+ writeVarUInt 2+ writeBinaryInt32 bucket_num+ writeVarUInt 0++-- | read information from block information+readInfo :: BlockInfo -> Reader BlockInfo+readInfo info@Info {is_overflows = io, bucket_num = bn} = do+ field_num <- readVarInt+ case field_num of+ 1 -> do+ io' <- readBinaryUInt8+ readInfo Info {is_overflows = if io' == 0 then False else True, bucket_num = bn}+ 2 -> do+ bn' <- readBinaryInt32+ readInfo Info {is_overflows = io, bucket_num = bn'}+ _ -> return info++-- | Read a stream of data into a block. Data are read into column type+readBlockInputStream :: ServerInfo->Reader Block+readBlockInputStream server_info = do+ let defaultInfo =+ Info+ { is_overflows = False,+ bucket_num = -1+ } -- TODO should have considered the revision+ info <- readInfo defaultInfo+ n_columns <- readVarInt+ n_rows <- readVarInt+ let loop :: Int -> Reader (Vector ClickhouseType, ByteString, ByteString)+ loop n = do+ column_name <- readBinaryStr+ column_type <- readBinaryStr+ column <- readColumn server_info (fromIntegral n_rows) column_type+ return (column, column_name, column_type)+ v <- V.generateM (fromIntegral n_columns) loop+ let datas = (\(x, _, _) -> x) <$> v+ names = (\(_, x, _) -> x) <$> v+ types = (\(_, _, x) -> x) <$> v+ return+ ColumnOrientedBlock+ { cdata = datas,+ info = info,+ columns_with_type = V.zip names types+ }++-- | write data from column type into string builder.+writeBlockOutputStream :: Context->Block->Writer Builder+writeBlockOutputStream ctx@(Context _ server_info _)+ (ColumnOrientedBlock columns_with_type cdata info) = do+ let revis = fromIntegral $+ revision $+ (case server_info of+ Nothing -> error ""+ Just info -> info)+ if revis >= Defines._DBMS_MIN_REVISION_WITH_BLOCK_INFO+ then writeBlockInfo info+ else return ()+ let n_rows = fromIntegral $ V.length cdata+ n_columns = fromIntegral $ V.length (cdata ! 0)+ writeVarUInt n_rows+ writeVarUInt n_columns+ V.imapM_ (\i (col, t)->do+ writeBinaryStr col+ writeBinaryStr t+ writeColumn ctx col t (cdata ! i)+ ) columns_with_type
+ src/Database/ClickHouseDriver/CBits/bigint.c view
@@ -0,0 +1,41 @@+#include<stdio.h>+#include<stdint.h>+#include<math.h>+#include "bigint.h"++double word128_division(__int64_t hi, __int64_t lo, int scale){+ __int128_t i_128;+ if (hi > UINT64_MAX){+ i_128 = UINT64_MAX - hi;+ i_128 = i_128 << 64;+ i_128 = -i_128;+ i_128 = i_128 - (UINT64_MAX - lo) - 1;+ }+ else{+ i_128 = hi;+ i_128 = i_128 << 64;+ i_128 += lo;+ }+ double power = pow(10, scale);+ return (double) i_128 / power;+}++__int64_t low_bits_128(double x, int scale){+ __int128_t i128 = x * pow(10, scale);+ return i128 & UINT64_MAX;+}++__int64_t hi_bits_128(double x, int scale){+ __int128_t i128 = x * pow(10, scale);+ return (i128 >> 64) & UINT64_MAX;+}++__int64_t low_bits_negative_128(double x, int scale){+ __int128_t i128 = x * pow(10, scale);+ return UINT64_MAX - (i128 & UINT64_MAX) + 1;+}++__int64_t hi_bits_negative_128(double x, int scale){+ __int128_t i128 = x * pow(10, scale);+ return UINT64_MAX - ((i128 >> 64) & UINT64_MAX);+}
+ src/Database/ClickHouseDriver/CBits/datetime.c view
@@ -0,0 +1,49 @@+#define __USE_XOPEN+#define _GNU_SOURCE+#include<time.h>+#include<stdio.h>+#include<stdlib.h>+#include<string.h>+#include<math.h>+#include "datetime.h"++char * convert_time(time_t time, char * timezone, size_t length){+ struct tm mytm = {0};+ mytm.tm_isdst = 1;+ timezone[length] = '\0';+ char * buf = malloc(sizeof(char) * 30);+ putenv(timezone);+ tzset();+ localtime_r(&time, &mytm);+ strftime(buf, 50, "%a, %d %b %Y %H:%M:%S %z(%Z)", &mytm);+ return buf;+}++char * convert_time64(time_t time, char * timezone, size_t length, size_t scale){+ struct tm my_tm = {0};+ my_tm.tm_isdst = 1;+ char * buf = malloc(sizeof(char) * 40);+ putenv(timezone);+ tzset();+ int exp = pow(10, scale);+ time_t scaled = time / exp;+ localtime_r(&scaled, &my_tm);+ return buf;+}++time_t parse_time(char * time_string, char * timezone, size_t length1, size_t length2){+ struct tm my_tm = {0};+ my_tm.tm_isdst = 0;+ timezone[length1] = '\0';+ time_string[length2] = '\0';+ time_t result;+ putenv(timezone);+ tzset();+ strptime(time_string, "%d %b %Y %H:%M:%S", &my_tm);+}++time_t convert_time_from_int32(time_t time){+ struct tm* gmt = gmtime(&time);+ time_t converted = mktime(gmt);+ return converted;+}
+ src/Database/ClickHouseDriver/CBits/varuint.c view
@@ -0,0 +1,65 @@+#include <stdlib.h>+#include <stdio.h>+#include <string.h>+#include "varuint.h"++/**+ *Encode integer using LEB128+ */++const char *write_varint(u_int16_t number)+{+ char *ostr = malloc(sizeof(char) *32);+ char *ptr = ostr;+ memset(ostr, '\0', sizeof(char) *32);+ for (size_t i = 0; i < 9; ++i)+ {+ u_int8_t byte = number &0x7F;+ if (number > 0x7F)+ byte |= 0x80;+ *ptr = byte;+ ++ptr;++ number >>= 7;+ if (!number)+ return ostr;+ }+ return ostr;+}++u_int16_t read_varint(u_int16_t cont,char *istr, size_t size)+{+ const char *end = istr + size;+ int byte;+ for (size_t i = 0; i < 9; ++i)+ {+ if(istr == end){+ break;+ }+ byte = *istr;+ ++istr;+ cont |= (byte & 0x7F) << (7 *i);+ if (!(byte & 0x80))+ break;+ }+ return cont;+}++size_t count_read(char *istr, size_t size)+{+ const char *end = istr + size;+ size_t n = 0;+ int byte;+ for (size_t i = 0; i < 9; ++i)+ {+ if (istr == end){+ return 0;+ }+ byte = *istr;+ ++istr;+ ++n;+ if (!(byte & 0x80))+ break;+ }+ return n;+}
+ src/Database/ClickHouseDriver/Client.hs view
@@ -0,0 +1,346 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.+----------------------------------------------------------------------------++{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE NamedFieldPuns #-}++-- | This module provides implementations of user's APIs+--++module Database.ClickHouseDriver.Client+ ( -- * Data Fetch and Insert+ query,+ queryWithInfo,+ deploySettings,+ insertMany,+ insertOneRow,+ ping,+ withQuery,+ Database.ClickHouseDriver.Client.fetch,+ fetchWithInfo,+ execute,+ -- * Communication+ client,+ defaultClient,+ closeClient,+ -- * Connection pool+ defaultClientPool,+ createClient,+ createClientPool+ )+where++import Database.ClickHouseDriver.Column ( ClickhouseType )+import Database.ClickHouseDriver.Connection+ ( ping',+ tcpConnect,+ sendQuery,+ sendData,+ processInsertQuery,+ receiveResult )+import Database.ClickHouseDriver.Pool ( createConnectionPool )+import Database.ClickHouseDriver.Defines ( _BUFFER_SIZE )+import qualified Database.ClickHouseDriver.Defines as Defines+import Database.ClickHouseDriver.Types+ ( ConnParams(..),+ CKResult(CKResult, query_result),+ TCPConnection(TCPConnection, tcpSocket),+ getServerInfo,+ defaultQueryInfo )+import Database.ClickHouseDriver.IO.BufferedReader ( createBuffer )+import Control.Concurrent.Async ( mapConcurrently )+import Control.Exception ( SomeException, try )+import Control.Monad.State ( StateT(runStateT) )+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as C8+import Data.Hashable ( Hashable(hashWithSalt) )+import Data.Typeable ( Typeable )+import Data.Vector ( Vector )+import Haxl.Core+ ( putFailure,+ putSuccess,+ dataFetch,+ initEnv,+ runHaxl,+ stateEmpty,+ stateGet,+ stateSet,+ BlockedFetch(..),+ DataSource(fetch),+ DataSourceName(..),+ PerformFetch(SyncFetch),+ Env(states),+ GenHaxl,+ ShowP(..),+ StateKey(State) ) +import qualified Network.Simple.TCP as TCP+import Text.Printf ( printf )+import Data.Pool (Pool(..), withResource, destroyAllResources)+import Data.Time.Clock ( NominalDiffTime )+import Data.Default.Class (def)++{-# INLINE _DEFAULT_PING_WAIT_TIME #-}+_DEFAULT_PING_WAIT_TIME :: Integer+_DEFAULT_PING_WAIT_TIME = 10000++{-# INLINE _DEFAULT_USERNAME #-}+_DEFAULT_USERNAME :: [Char]+_DEFAULT_USERNAME = "default"++{-# INLINE _DEFAULT_HOST_NAME #-}+_DEFAULT_HOST_NAME :: [Char]+_DEFAULT_HOST_NAME = "localhost"++{-# INLINE _DEFAULT_PASSWORD #-}+_DEFAULT_PASSWORD :: [Char]+_DEFAULT_PASSWORD = ""++{-# INLINE _DEFAULT_PORT_NAME #-}+_DEFAULT_PORT_NAME :: [Char]+_DEFAULT_PORT_NAME = "9000"++{-# INLINE _DEFAULT_DATABASE#-}+_DEFAULT_DATABASE :: [Char]+_DEFAULT_DATABASE = "default"++{-# INLINE _DEFAULT_COMPRESSION_SETTING #-}+_DEFAULT_COMPRESSION_SETTING :: Bool+_DEFAULT_COMPRESSION_SETTING = False++-- | GADT +data Query a where+ FetchData :: String+ -- ^ SQL statement such as "SELECT * FROM table"+ ->Query (Either String CKResult)+ -- ^ result data in Haskell type and additional information++deriving instance Show (Query a)++deriving instance Typeable Query++deriving instance Eq (Query a)++instance ShowP Query where showp = show++instance Hashable (Query a) where+ hashWithSalt salt (FetchData cmd) = hashWithSalt salt cmd++instance DataSourceName Query where+ dataSourceName _ = "ClickhouseServer"++instance DataSource u Query where+ fetch (resource) _flags env = SyncFetch $ \blockedFetches -> do+ printf "Fetching %d queries.\n" (length blockedFetches)+ mapConcurrently (fetchData resource) blockedFetches+ return ()++instance StateKey Query where+ data State Query = CKResource TCPConnection+ | CKPool (Pool TCPConnection)++class Resource a where+ client :: Either String a->IO(Env () w)+ -- ^ Either wrong message of resource with type a++-- | fetch data+fetchData :: State Query->BlockedFetch Query->IO ()+fetchData (CKResource tcpconn) fetch = do+ let (queryStr, var) = case fetch of+ BlockedFetch (FetchData q) var' -> (C8.pack q, var')+ e <- Control.Exception.try $ do+ sendQuery tcpconn queryStr Nothing+ sendData tcpconn "" Nothing+ let serverInfo = case getServerInfo tcpconn of+ Just info -> info+ Nothing -> error "Empty server information"+ let sock = tcpSocket tcpconn+ buf <- createBuffer _BUFFER_SIZE sock+ (res, _) <- runStateT (receiveResult serverInfo defaultQueryInfo) buf+ return res+ either+ (putFailure var)+ (putSuccess var)+ (e :: Either SomeException (Either String CKResult))+fetchData (CKPool pool) fetch = do+ let (queryStr, var) = case fetch of+ BlockedFetch (FetchData q) var' -> (C8.pack q, var')+ e <- Control.Exception.try $ do+ withResource pool $ \conn->do+ sendQuery conn queryStr Nothing+ sendData conn "" Nothing+ let serverInfo = case getServerInfo conn of+ Just info -> info+ Nothing -> error "Empty server information"+ let sock = tcpSocket conn+ buf <- createBuffer _BUFFER_SIZE sock+ (res, _) <- runStateT (receiveResult serverInfo defaultQueryInfo) buf+ return res+ either+ (putFailure var)+ (putSuccess var)+ (e :: Either SomeException (Either String CKResult))++deploySettings :: TCPConnection -> IO (Env () w)+deploySettings tcp =+ initEnv (stateSet (CKResource tcp) stateEmpty) ()++defaultClient :: IO (Env () w)+defaultClient = createClient def + +createClient :: ConnParams->IO(Env () w)+createClient ConnParams{+ username' + ,host' + ,port' + ,password' + ,compression'+ ,database' + } = do+ tcp <- tcpConnect + host' + port'+ username' + password' + database'+ compression'+ case tcp of+ Left e -> client ((Left e) :: Either String (Pool TCPConnection))+ Right conn -> client $ Right conn+ +defaultClientPool :: Int+ -- ^ number of stripes+ ->NominalDiffTime+ -- ^ idle time for each stripe+ ->Int+ -- ^ maximum resources for reach stripe+ ->IO (Env () w)+ -- ^ Haxl env wrapped in IO monad.+defaultClientPool = createClientPool def++createClientPool :: ConnParams+ -- ^ parameters for connection settings+ ->Int+ -- ^ number of stripes+ ->NominalDiffTime+ -- ^ idle time for each stripe+ ->Int+ -- ^ maximum resources for reach stripe+ ->IO(Env () w)+createClientPool params numberStripes idleTime maxResources = do + pool <- createConnectionPool params numberStripes idleTime maxResources+ client $ Right pool++instance Resource TCPConnection where+ client (Left e) = error e+ client (Right src) = initEnv (stateSet (CKResource src) stateEmpty) ()++instance Resource (Pool TCPConnection) where+ client (Left e) = error e+ client (Right src) = initEnv (stateSet (CKPool src) stateEmpty) ()++-- | fetch data alone with query information+fetchWithInfo :: String->GenHaxl u w (Either String CKResult)+fetchWithInfo = dataFetch . FetchData++-- | Fetch data+fetch :: String+ -- ^ SQL SELECT command+ ->GenHaxl u w (Either String (Vector (Vector ClickhouseType)))+ -- ^ result wrapped in Haxl monad for other tasks run with concurrency.+fetch str = do+ result_with_info <- fetchWithInfo str+ case result_with_info of+ Right CKResult{query_result=r}->return $ Right r+ Left err -> return $ Left err++-- | query result contains query information.+queryWithInfo :: String->Env () w->IO (Either String CKResult)+queryWithInfo query source = runHaxl source (executeQuery query)+ where+ executeQuery :: String->GenHaxl u w (Either String CKResult)+ executeQuery = dataFetch . FetchData++-- | query command+query :: Env () w+ -- ^ Haxl environment for connection+ ->String+ -- ^ Query command for "SELECT" and "SHOW" only+ ->IO (Either String (Vector (Vector ClickhouseType)))+query source cmd = do+ query_with_info <- queryWithInfo cmd source+ case query_with_info of+ Right CKResult{query_result=r}->return $ Right r+ Left err->return $ Left err++-- | For general use e.g. creating table,+-- multiple queries, multiple insertions. +execute :: Env u w -> GenHaxl u w a -> IO a+execute = runHaxl++withQuery :: Env () w+ -- ^ enviroment i.e. the database resource+ ->String+ -- ^ sql statement+ ->(Either String (Vector (Vector ClickhouseType))->IO a)+ -- ^ callback function that returns type a+ ->IO a+ -- ^ type a wrapped in IO monad.+withQuery source cmd f = query source cmd >>= f++insertMany :: Env () w->String->[[ClickhouseType]]->IO(BS.ByteString)+insertMany source cmd items = do+ let st :: Maybe (State Query) = stateGet $ states source+ case st of+ Nothing -> error "No Connection."+ Just (CKResource tcp) -> processInsertQuery tcp (C8.pack cmd) Nothing items+ Just (CKPool pool) -> + withResource pool $ \tcp->do+ processInsertQuery tcp (C8.pack cmd) Nothing items++insertOneRow :: Env () w+ ->String+ -- ^ SQL command+ ->[ClickhouseType]+ -- ^ a row of local clickhouse data type to be serialized and inserted. + ->IO(BS.ByteString)+ -- ^ The resulting bytestring indicates success or failure.+insertOneRow source cmd items = insertMany source cmd [items]++-- | ping pong +ping :: Env () w->IO()+ping source = do+ let get :: Maybe (State Query) = stateGet $ states source+ case get of+ Nothing -> print "empty source"+ Just (CKResource tcp)+ -> ping' Defines._DEFAULT_PING_WAIT_TIME tcp >>= print+ Just (CKPool pool)+ -> withResource pool $ \src->+ ping' Defines._DEFAULT_PING_WAIT_TIME src >>= print ++-- | close connection+closeClient :: Env () w -> IO()+closeClient source = do+ let get :: Maybe (State Query) = stateGet $ states source+ case get of+ Nothing -> return ()+ Just (CKResource TCPConnection{tcpSocket=sock})+ -> TCP.closeSock sock+ Just (CKPool pool)+ -> destroyAllResources pool
+ src/Database/ClickHouseDriver/ClientProtocol.hs view
@@ -0,0 +1,68 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.+-------------------------------------------------------------------------+-- This module provides constant for client protocols. +{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++module Database.ClickHouseDriver.ClientProtocol where++import Data.ByteString (ByteString)+import Data.Vector (Vector, fromList, (!?))+++-- Name, version, revision, default DB+_HELLO :: Word+_HELLO = 0 :: Word++-- Query id, settings,+_QUERY :: Word+_QUERY = 1 :: Word++-- A block of data+_DATA :: Word+_DATA = 2 :: Word++-- Cancel the query execution+_CANCEL :: Word+_CANCEL = 3 :: Word++-- Check that the connection to the server is alive+_PING :: Word+_PING = 4 :: Word++_TABLES_STATUS_REQUEST :: Word+_TABLES_STATUS_REQUEST = 5 :: Word++_COMPRESSION_ENABLE :: Word+_COMPRESSION_ENABLE = 1 :: Word++_COMPRESSION_DISABLE :: Word+_COMPRESSION_DISABLE = 0 :: Word++_COMPRESSION_METHOD_LZ4 :: Word+_COMPRESSION_METHOD_LZ4 = 1 :: Word++_COMPRESSION_METHOD_LZ4HC :: Word+_COMPRESSION_METHOD_LZ4HC = 2 :: Word++_COMPRESSION_METHOD_ZSTD :: Word+_COMPRESSION_METHOD_ZSTD = 3 :: Word++_COMPRESSION_METHOD_BYTE_LZ4 :: Integer+_COMPRESSION_METHOD_BYTE_LZ4 = 0x82++_COMPRESSION_METHOD_BYTE_ZSTD :: Integer+_COMPRESSION_METHOD_BYTE_ZSTD = 0x90++typeStr :: Vector ByteString+typeStr = fromList ["Hello", "Query", "Data", "Cancel", "Ping", "TablesStatusRequest"]++toString :: Int -> ByteString+toString n = + case typeStr !? n of+ Nothing -> "Unknown Packet"+ Just t -> t
+ src/Database/ClickHouseDriver/Column.hs view
@@ -0,0 +1,998 @@+-------------------------------------------------------------------------+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | This module contains the implementations of+-- serialization and deserialization of Clickhouse data types.+module Database.ClickHouseDriver.Column+ ( -- * Serialize and deserialize+ ClickhouseType (..),+ readColumn,+ writeColumn,++ -- * Operations on ClickhouseType+ transpose,+ Database.ClickHouseDriver.Column.putStrLn,+ )+where++import Database.ClickHouseDriver.Types (ClickhouseType (..), Context (..), ServerInfo (..))+import Database.ClickHouseDriver.IO.BufferedReader+ ( Reader,+ readBinaryInt16,+ readBinaryInt32,+ readBinaryInt64,+ readBinaryInt8,+ readBinaryStr,+ readBinaryStrWithLength,+ readBinaryUInt128,+ readBinaryUInt16,+ readBinaryUInt32,+ readBinaryUInt64,+ readBinaryUInt8,+ )+import Database.ClickHouseDriver.IO.BufferedWriter+ ( Writer,+ writeBinaryFixedLengthStr,+ writeBinaryInt16,+ writeBinaryInt32,+ writeBinaryInt64,+ writeBinaryInt8,+ writeBinaryStr,+ writeBinaryUInt128,+ writeBinaryUInt16,+ writeBinaryUInt32,+ writeBinaryUInt64,+ writeBinaryUInt8,+ writeVarUInt,+ )+import Control.Monad.State.Lazy (MonadIO (..))+import Data.Binary (Word64, Word8)+import Data.Bits (shift, (.&.), (.|.))+import Data.ByteString (ByteString, isPrefixOf)+import qualified Data.ByteString as BS+ ( drop,+ filter,+ intercalate,+ length,+ splitWith,+ take,+ unpack,+ )+import Data.ByteString.Builder (Builder)+import Data.ByteString.Char8 (readInt)+import qualified Data.ByteString.Char8 as C8+import Data.ByteString.Unsafe+ ( unsafePackCString,+ unsafeUseAsCStringLen,+ )+import qualified Data.HashMap.Strict as Map+import Data.Hashable (Hashable (hash))+import Data.Int (Int32, Int64)+import qualified Data.List as List+import Data.Maybe (fromJust, fromMaybe)+import Data.Time+ ( TimeZone (..),+ addDays,+ diffDays,+ fromGregorian,+ getCurrentTimeZone,+ toGregorian,+ )+import Data.UUID as UUID+ ( fromString,+ fromWords,+ toString,+ toWords,+ )+import Data.Vector (Vector, (!))+import qualified Data.Vector as V+ ( cons,+ drop,+ foldl',+ fromList,+ generate,+ length,+ map,+ mapM,+ mapM_,+ replicateM,+ scanl',+ sum,+ take,+ toList,+ zipWith,+ zipWithM_,+ )+import Foreign.C (CString)+import Network.IP.Addr+ ( IP4 (..),+ IP6 (..),+ ip4FromOctets,+ ip4ToOctets,+ ip6FromWords,+ ip6ToWords,+ )+#define EQUAL 61+#define COMMA 44+#define SPACE 32+#define QUOTE 39+--Debug+--import Debug.Trace++-- Notice: Codes in this file might be difficult to read.+---------------------------------------------------------------------------------------+---Readers+readColumn ::+ -- | Server information is needed in case of some parameters are missing+ ServerInfo ->+ -- | number of rows+ Int ->+ -- | data type+ ByteString ->+ Reader (Vector ClickhouseType)+readColumn server_info n_rows spec+ | "String" `isPrefixOf` spec = V.replicateM n_rows (CKString <$> readBinaryStr)+ | "Array" `isPrefixOf` spec = readArray server_info n_rows spec+ | "FixedString" `isPrefixOf` spec = readFixed n_rows spec+ | "DateTime" `isPrefixOf` spec = readDateTime server_info n_rows spec+ | "Date" `isPrefixOf` spec = readDate n_rows+ | "Tuple" `isPrefixOf` spec = readTuple server_info n_rows spec+ | "Nullable" `isPrefixOf` spec = readNullable server_info n_rows spec+ | "LowCardinality" `isPrefixOf` spec = readLowCardinality server_info n_rows spec+ | "Decimal" `isPrefixOf` spec = readDecimal n_rows spec+ | "Enum" `isPrefixOf` spec = readEnum n_rows spec+ | "Int" `isPrefixOf` spec = readIntColumn n_rows spec+ | "UInt" `isPrefixOf` spec = readIntColumn n_rows spec+ | "IPv4" `isPrefixOf` spec = readIPv4 n_rows+ | "IPv6" `isPrefixOf` spec = readIPv6 n_rows+ | "SimpleAggregateFunction" `isPrefixOf` spec = readSimpleAggregateFunction server_info n_rows spec+ | "UUID" `isPrefixOf` spec = readUUID n_rows+ | otherwise = error ("Unknown Type: " Prelude.++ C8.unpack spec)++writeColumn ::+ -- | context contains client information and server information+ Context ->+ -- | column name+ ByteString ->+ -- | column type (String, Int, etc)+ ByteString ->+ -- | items to be serialized.+ Vector ClickhouseType ->+ -- | result wrapped in a customized Writer Monad used for concatenating string builders.+ Writer Builder+writeColumn ctx col_name cktype items+ | "String" `isPrefixOf` cktype = writeStringColumn col_name items+ | "FixedString(" `isPrefixOf` cktype = writeFixedLengthString col_name cktype items+ | "Int" `isPrefixOf` cktype = writeIntColumn col_name cktype items+ | "UInt" `isPrefixOf` cktype = writeUIntColumn col_name cktype items+ | "Nullable(" `isPrefixOf` cktype = writeNullable ctx col_name cktype items+ | "Tuple" `isPrefixOf` cktype = writeTuple ctx col_name cktype items+ | "Enum" `isPrefixOf` cktype = writeEnum col_name cktype items+ | "Array" `isPrefixOf` cktype = writeArray ctx col_name cktype items+ | "UUID" `isPrefixOf` cktype = writeUUID col_name items+ | "IPv4" `isPrefixOf` cktype = writeIPv4 col_name items+ | "IPv6" `isPrefixOf` cktype = writeIPv6 col_name items+ | "Date" `isPrefixOf` cktype = writeDate col_name items+ | "LowCardinality" `isPrefixOf` cktype = writeLowCardinality ctx col_name cktype items+ | "DateTime" `isPrefixOf` cktype = writeDateTime col_name cktype items+ | "Decimal" `isPrefixOf` cktype = writeDecimal col_name cktype items+ | otherwise = error ("Unknown Type in the column: " Prelude.++ C8.unpack col_name)+---------------------------------------------------------------------------------------------+readFixed :: Int -> ByteString -> Reader (Vector ClickhouseType)+readFixed n_rows spec = do+ let l = BS.length spec+ let str_number = BS.take (l - 13) (BS.drop 12 spec)+ let number = case readInt str_number of+ Nothing -> 0 -- This can't happen+ Just (x, _) -> x+ V.replicateM n_rows (readFixedLengthString number)++readFixedLengthString :: Int -> Reader ClickhouseType+readFixedLengthString str_len = CKString <$> readBinaryStrWithLength str_len++writeStringColumn :: ByteString -> Vector ClickhouseType -> Writer Builder+writeStringColumn col_name =+ V.mapM_+ ( \case+ CKString s -> writeBinaryStr s+ CKNull -> writeVarUInt 0+ _ -> error (typeMismatchError col_name)+ )++writeFixedLengthString :: ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeFixedLengthString col_name spec items = do+ let l = BS.length spec+ let Just (len, _) = readInt $ BS.take (l - 13) (BS.drop 12 spec)+ V.mapM_+ ( \case+ CKString s -> writeBinaryFixedLengthStr (fromIntegral len) s+ CKNull -> const () <$> V.replicateM (fromIntegral len) (writeVarUInt 0)+ x -> error (typeMismatchError col_name ++ " got: " ++ show x)+ )+ items++---------------------------------------------------------------------------------------------++-- | read data in format of bytestring into format of haskell type.+readIntColumn :: Int -> ByteString -> Reader (Vector ClickhouseType)+readIntColumn n_rows "Int8" = V.replicateM n_rows (CKInt8 <$> readBinaryInt8)+readIntColumn n_rows "Int16" = V.replicateM n_rows (CKInt16 <$> readBinaryInt16)+readIntColumn n_rows "Int32" = V.replicateM n_rows (CKInt32 <$> readBinaryInt32)+readIntColumn n_rows "Int64" = V.replicateM n_rows (CKInt64 <$> readBinaryInt64)+readIntColumn n_rows "UInt8" = V.replicateM n_rows (CKUInt8 <$> readBinaryUInt8)+readIntColumn n_rows "UInt16" = V.replicateM n_rows (CKUInt16 <$> readBinaryUInt16)+readIntColumn n_rows "UInt32" = V.replicateM n_rows (CKUInt32 <$> readBinaryUInt32)+readIntColumn n_rows "UInt64" = V.replicateM n_rows (CKUInt64 <$> readBinaryUInt64)+readIntColumn _ x = error ("expect an integer but got: " ++ show x)++writeIntColumn :: ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeIntColumn col_name spec items = do+ let Just (indicator, _) = readInt $ BS.drop 3 spec -- indicator indicates which integer type, is it Int8 or Int64 etc.+ writeIntColumn' indicator col_name items+ where+ writeIntColumn' :: Int -> ByteString -> Vector ClickhouseType -> Writer Builder+ writeIntColumn' indicator col_name =+ case indicator of+ 8 ->+ V.mapM_ -- mapM_ acts like for-loop in this context, since it repeats monadic actions.+ ( \case+ CKInt8 x -> writeBinaryInt8 x+ CKNull -> writeBinaryInt8 0+ _ -> error (typeMismatchError col_name)+ )+ 16 ->+ V.mapM_+ ( \case+ CKInt16 x -> writeBinaryInt16 x+ CKNull -> writeBinaryInt16 0+ _ -> error (typeMismatchError col_name)+ )+ 32 ->+ V.mapM_+ ( \case+ CKInt32 x -> writeBinaryInt32 x+ CKNull -> writeBinaryInt32 0+ _ -> error (typeMismatchError col_name)+ )+ 64 ->+ V.mapM_+ ( \case+ CKInt64 x -> writeBinaryInt64 x+ CKNull -> writeBinaryInt64 0+ _ -> error (typeMismatchError col_name)+ )++writeUIntColumn :: ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeUIntColumn col_name spec items = do+ let Just (indicator, _) = readInt $ BS.drop 4 spec+ writeUIntColumn' indicator col_name items+ where+ writeUIntColumn' :: Int -> ByteString -> Vector ClickhouseType -> Writer Builder+ writeUIntColumn' indicator col_name =+ case indicator of+ 8 ->+ V.mapM_+ ( \case+ CKUInt8 x -> writeBinaryUInt8 x+ CKNull -> writeBinaryUInt8 0+ _ -> error (typeMismatchError col_name)+ )+ 16 ->+ V.mapM_+ ( \case+ CKUInt16 x -> writeBinaryInt16 $ fromIntegral x+ CKNull -> writeBinaryInt16 0+ _ -> error (typeMismatchError col_name)+ )+ 32 ->+ V.mapM_+ ( \case+ CKUInt32 x -> writeBinaryInt32 $ fromIntegral x+ CKNull -> writeBinaryInt32 0+ _ -> error (typeMismatchError col_name)+ )+ 64 ->+ V.mapM_+ ( \case+ CKUInt64 x -> writeBinaryInt64 $ fromIntegral x+ CKNull -> writeBinaryInt64 0+ _ -> error (typeMismatchError col_name)+ )++---------------------------------------------------------------------------------------------++{-+ There are two types of Datetime+ DateTime(TZ) or DateTime64(precision,TZ)+ server information is required if TZ parameter is missing.+-}+readDateTime :: ServerInfo -> Int -> ByteString -> Reader (Vector ClickhouseType)+readDateTime server_info n_rows spec = do+ let (scale, spc) = readTimeSpec spec+ case spc of+ Nothing -> readDateTimeWithSpec server_info n_rows scale ""+ Just tz_name -> readDateTimeWithSpec server_info n_rows scale tz_name+ +readTimeSpec :: ByteString -> (Maybe Int, Maybe ByteString)+readTimeSpec spec'+ | "DateTime64" `isPrefixOf` spec' = do+ let l = BS.length spec'+ let inner_specs = BS.take (l - 12) (BS.drop 11 spec')+ let split = getSpecs inner_specs+ case split of+ [] -> (Nothing, Nothing)+ [x] -> (Just $ fst $ fromJust $ readInt x, Nothing)+ [x, y] -> (Just $ fst $ fromJust $ readInt x, Just y)+ | otherwise = do+ let l = BS.length spec'+ let inner_specs = BS.take (l - 12) (BS.drop 10 spec')+ (Nothing, Just inner_specs)+ +readDateTimeWithSpec :: ServerInfo -> Int -> Maybe Int -> ByteString -> Reader (Vector ClickhouseType)+readDateTimeWithSpec ServerInfo {timezone = maybe_zone} n_rows Nothing tz_name = do+ data32 <- readIntColumn n_rows "Int32"+ let tz_to_send =+ if tz_name /= ""+ then "TZ=" <> tz_name+ else fromMaybe "" maybe_zone+ let toDateTimeStringM =+ V.mapM+ ( \(CKInt32 x) -> do+ c_str <-+ unsafeUseAsCStringLen+ tz_to_send+ (uncurry (c_convert_time (fromIntegral x)))+ unsafePackCString c_str+ )+ data32+ toDateTimeString <- liftIO $ toDateTimeStringM+ return $ V.map CKString toDateTimeString++readDateTimeWithSpec ServerInfo {timezone = maybe_zone} n_rows (Just scl) tz_name = do+ data64 <- readIntColumn n_rows "Int64"+ let scale = 10 ^ fromIntegral scl+ let tz_to_send =+ if tz_name /= ""+ then "TZ=" <> tz_name+ else fromMaybe "" maybe_zone+ let toDateTimeStringM =+ V.mapM+ ( \(CKInt64 x) -> do+ c_str <-+ unsafeUseAsCStringLen+ tz_to_send+ (uncurry (c_convert_time64 (fromIntegral x / scale)))+ unsafePackCString c_str+ )+ data64+ toDateTimeString <- liftIO $ toDateTimeStringM+ return $ V.map CKString toDateTimeString++writeDateTime :: ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeDateTime col_name spec items = do+ let (scale', spc) = readTimeSpec spec+ case scale' of+ Nothing -> do+ case spc of+ Nothing -> do+ TimeZone {timeZoneName = tz'} <- liftIO $ getCurrentTimeZone+ writeDateTimeWithSpec $ C8.pack tz'+ Just spec -> writeDateTimeWithSpec spec+ Just _ -> do+ --TODO: Can someone implement this? I am so pissed off!+ undefined+ where+ writeDateTimeWithSpec :: ByteString -> Writer Builder+ writeDateTimeWithSpec tz_name = do+ V.mapM_+ ( \case+ (CKInt32 i32) -> do+ converted <- liftIO $ convert_time_from_int32 i32+ writeBinaryInt32 converted+ (CKString time_str) -> do+ converted <-+ liftIO $+ unsafeUseAsCStringLen+ tz_name+ ( \(tz, l) ->+ unsafeUseAsCStringLen+ time_str+ (\(time_str, l2) -> c_write_time time_str tz l l2)+ )+ writeBinaryInt32 converted+ _ -> error (typeMismatchError col_name)+ )+ items++foreign import ccall unsafe "datetime.h convert_time" c_convert_time :: Int64 -> CString -> Int -> IO CString++foreign import ccall unsafe "datetime.h convert_time" c_convert_time64 :: Float -> CString -> Int -> IO CString++foreign import ccall unsafe "datetime.h parse_time" c_write_time :: CString -> CString -> Int -> Int -> IO Int32++foreign import ccall unsafe "datetime.h convert_time_from_int32" convert_time_from_int32 :: Int32 -> IO Int32++------------------------------------------------------------------------------------------------+readLowCardinality :: ServerInfo -> Int -> ByteString -> Reader (Vector ClickhouseType)+readLowCardinality _ 0 _ = return (V.fromList [])+readLowCardinality server_info n spec = do+ readBinaryUInt64 --state prefix+ let l = BS.length spec+ let inner = BS.take (l - 16) (BS.drop 15 spec)+ serialization_type <- readBinaryUInt64+ -- Lowest bytes contains info about key type.+ let key_type = serialization_type .&. 0xf+ index_size <- readBinaryUInt64+ -- Strip the 'Nullable' tag to avoid null map reading.+ index <- readColumn server_info (fromIntegral index_size) (stripNullable inner)+ readBinaryUInt64 -- #keys+ keys <- case key_type of+ 0 -> V.map fromIntegral <$> V.replicateM n readBinaryUInt8+ 1 -> V.map fromIntegral <$> V.replicateM n readBinaryUInt16+ 2 -> V.map fromIntegral <$> V.replicateM n readBinaryUInt32+ 3 -> V.map fromIntegral <$> V.replicateM n readBinaryUInt64+ if "Nullable" `isPrefixOf` inner+ then do+ let nullable = fmap (\k -> if k == 0 then CKNull else index ! k) keys+ return nullable+ else return $ fmap (index !) keys+ where+ stripNullable :: ByteString -> ByteString+ stripNullable spec+ | "Nullable" `isPrefixOf` spec = BS.take (l - 10) (BS.drop 9 spec)+ | otherwise = spec+ l = BS.length spec++writeLowCardinality :: Context -> ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeLowCardinality ctx col_name spec items = do+ let inner = BS.take (BS.length spec - 16) (BS.drop 15 spec)+ (keys, index) <-+ if "Nullable" `isPrefixOf` inner+ then do+ --let null_inner_spec = BS.take (BS.length inner - 10) (BS.drop 9 spec)+ let hashedItem = hashItems True items+ let key_by_index_element = V.foldl' insertKeys Map.empty hashedItem+ let keys = V.map (\k -> key_by_index_element Map.! k + 1) hashedItem+ -- First element is NULL if column is nullable+ let index = V.fromList $ 0 : (Map.keys $ key_by_index_element)+ return (keys, index)+ else do+ let hashedItem = hashItems False items+ let key_by_index_element = V.foldl' insertKeys Map.empty hashedItem+ let keys = V.map (key_by_index_element Map.!) hashedItem+ let index = V.fromList $ Map.keys $ key_by_index_element+ return (keys, index)+ if V.length index == 0+ then return ()+ else do+ let int_type = floor $ logBase 2 (fromIntegral $ V.length index) / 8 :: Int64+ let has_additional_keys_bit = 1 `shift` 9+ let need_update_dictionary = 1 `shift` 10+ let serialization_type =+ has_additional_keys_bit+ .|. need_update_dictionary+ .|. int_type+ let nullsInner =+ if "Nullable" `isPrefixOf` inner+ then BS.take (BS.length inner - 10) (BS.drop 9 spec)+ else inner+ writeBinaryUInt64 1 --state prefix+ writeBinaryInt64 serialization_type+ writeBinaryInt64 $ fromIntegral $ V.length index+ writeColumn ctx col_name nullsInner items+ writeBinaryInt64 $ fromIntegral $ V.length items+ case int_type of+ 0 -> V.mapM_ (writeBinaryUInt8 . fromIntegral) keys+ 1 -> V.mapM_ (writeBinaryUInt16 . fromIntegral) keys+ 2 -> V.mapM_ (writeBinaryUInt32 . fromIntegral) keys+ 3 -> V.mapM_ (writeBinaryUInt64 . fromIntegral) keys+ where+ insertKeys :: (Hashable a, Eq a) => Map.HashMap a Int -> a -> Map.HashMap a Int+ insertKeys m a = if Map.member a m then m else Map.insert a (Map.size m) m++ hashItems :: Bool -> Vector ClickhouseType -> Vector Int+ hashItems isNullable items =+ V.map+ ( \case+ CKInt16 x -> hash x+ CKInt8 x -> hash x+ CKInt32 x -> hash x+ CKInt64 x -> hash x+ CKUInt8 x -> hash x+ CKUInt16 x -> hash x+ CKUInt32 x -> hash x+ CKUInt64 x -> hash x+ CKString str -> hash str+ CKNull ->+ if isNullable+ then hash (0 :: Int)+ else error $ typeMismatchError col_name+ _ -> error $ typeMismatchError col_name+ )+ items++---------------------------------------------------------------------------------------------------------------------------------+{-+ Informal description (in terms of regular expression form) for this config:+ (\Null | \SOH)^{n_rows}+ \Null means null and \SOH which equals 1 means not null. The `|` in the middle means or.+-}+readNullable :: ServerInfo -> Int -> ByteString -> Reader (Vector ClickhouseType)+readNullable server_info n_rows spec = do+ let l = BS.length spec+ let cktype = BS.take (l - 10) (BS.drop 9 spec) -- Read Clickhouse type inside the bracket after the 'Nullable' spec.+ config <- readNullableConfig n_rows+ items <- readColumn server_info n_rows cktype+ let result = V.generate n_rows (\i -> if config ! i == 1 then CKNull else items ! i)+ return result+ where+ readNullableConfig :: Int -> Reader (Vector Word8)+ readNullableConfig n_rows = do+ config <- readBinaryStrWithLength n_rows+ (return . V.fromList . BS.unpack) config++writeNullable :: Context -> ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeNullable ctx col_name spec items = do+ let l = BS.length spec+ let inner = BS.take (l - 10) (BS.drop 9 spec)+ writeNullsMap items+ writeColumn ctx col_name inner items+ where+ writeNullsMap :: Vector ClickhouseType -> Writer Builder+ writeNullsMap =+ V.mapM_+ ( \case+ CKNull -> writeBinaryInt8 1+ _ -> writeBinaryInt8 0+ )++---------------------------------------------------------------------------------------------------------------------------------+{-+ Format:+ "+ One element of array of arrays can be represented as tree:+ (0 depth) [[3, 4], [5, 6]]+ | |+ (1 depth) [3, 4] [5, 6]+ | | | |+ (leaf) 3 4 5 6+ Offsets (sizes) written in breadth-first search order. In example above+ following sequence of offset will be written: 4 -> 2 -> 4+ 1) size of whole array: 4+ 2) size of array 1 in depth=1: 2+ 3) size of array 2 plus size of all array before in depth=1: 2 + 2 = 4+ After sizes info comes flatten data: 3 -> 4 -> 5 -> 6+ "+ Quoted from https://github.com/mymarilyn/clickhouse-driver/blob/master/clickhouse_driver/columns/arraycolumn.py++Here we don't implement in the form of BFS; instead, we use bottom up method:+First off, we compute the array of integer in which elements represent the size of subarrays (we call them spec arrays)+, where the function `readArraySpec`, `cut`, and `intervalize` do their jobs.+Second, we place the array of atomic elements (or flatten data) at the last position.+Then we cut the array of atomic elements according to the last spec array and form nested a nested array.+Finally we pop out the last spec array.+Repeat this process until all spec arrays are gone.++For example:+The target array is [[3, 4], [5, 6]]+The array on the right hand side of `|` is array of atomic elements.+The algorithm would be like this:+[2] [2,2] | [3,4,5,6]+-> [2] | [[3,4],[5,6]]+-> [[3,4],[5,6]]++For another example:+ target [[["Alex","Bob"], ["John"]],[["Jane"],["Steven","Mike","Sarah"]],[["Hello","world"]]]+ [3] [2,2,1] [2,1,1,3,2] | ["Alex","Bob","John","Jane","Steven","Mike","Sarah","Hello","world"]+-> [3] [2,2,1] | [["Alex","Bob"],["John"],["Jane"],["Steven","Mike","Sarah"],["Hello","world"]]+-> [3] | [[["Alex","Bob"],["John"]],[["Jane"],["Steven","Mike","Sarah"]],[["Hello","world"]]]+-> [[["Alex","Bob"],["John"]],[["Jane"],["Steven","Mike","Sarah"]],[["Hello","world"]]]++-}+readArray :: ServerInfo -> Int -> ByteString -> Reader (Vector ClickhouseType)+readArray server_info n_rows spec = do+ (lastSpec, x : xs) <- genSpecs spec [V.fromList [fromIntegral n_rows]]+ -- lastSpec which is not `Array`+ -- x:xs is the+ let numElem = fromIntegral $ V.sum x -- number of elements in the nested array.+ elems <- readColumn server_info numElem lastSpec+ let result' = foldl combine elems (x : xs)+ let CKArray arr = result' ! 0+ return arr+ where+ combine :: Vector ClickhouseType -> Vector Word64 -> Vector ClickhouseType+ combine elems config =+ let intervals = intervalize (fromIntegral <$> config)+ cut (a, b) = CKArray $ V.take b (V.drop a elems) -- cut element array+ embed = (\(l, r) -> cut (l, r - l + 1)) <$> intervals+ in embed+ intervalize :: Vector Int -> Vector (Int, Int)+ intervalize vec = V.drop 1 $ V.scanl' (\(_, b) v -> (b + 1, v + b)) (-1, -1) vec -- drop the first tuple (-1,-1)++readArraySpec :: Vector Word64 -> Reader (Vector Word64)+readArraySpec sizeArr = do+ let arrSum = (fromIntegral . V.sum) sizeArr+ offsets <- V.replicateM arrSum readBinaryUInt64+ let offsets' = V.cons 0 (V.take (arrSum - 1) offsets)+ let sizes = V.zipWith (-) offsets offsets'+ return sizes++genSpecs :: ByteString -> [Vector Word64] -> Reader (ByteString, [Vector Word64])+genSpecs spec rest@(x : _) = do+ let l = BS.length spec+ let cktype = BS.take (l - 7) (BS.drop 6 spec)+ if "Array" `isPrefixOf` spec+ then do+ next <- readArraySpec x+ genSpecs cktype (next : rest)+ else return (spec, rest)++writeArray :: Context -> ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeArray ctx col_name spec items = do+ let lens =+ V.scanl'+ ( \total ->+ ( \case+ (CKArray xs) -> total + V.length xs+ x ->+ error $+ "unexpected type in the column: "+ ++ show col_name+ ++ " with data"+ ++ show x+ )+ )+ 0+ items+ V.mapM_ (writeBinaryInt64 . fromIntegral) (V.drop 1 lens)+ let innerSpec = BS.take (BS.length spec - 7) (BS.drop 6 spec)+ let innerVector = V.map (\case CKArray xs -> xs) items+ let flattenVector =+ innerVector >>= \v -> do v+ writeColumn ctx col_name innerSpec flattenVector++--------------------------------------------------------------------------------------+readTuple :: ServerInfo -> Int -> ByteString -> Reader (Vector ClickhouseType)+readTuple server_info n_rows spec = do+ let l = BS.length spec+ let innerSpecString = BS.take (l - 7) (BS.drop 6 spec) -- Tuple(...) the dots represent innerSpecString+ let arr = V.fromList (getSpecs innerSpecString)+ datas <- V.mapM (readColumn server_info n_rows) arr+ let transposed = transpose datas+ return $ CKTuple <$> transposed++writeTuple :: Context -> ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeTuple ctx col_name spec items = do+ let inner = BS.take (BS.length spec - 7) (BS.drop 6 spec)+ let spec_arr = V.fromList $ getSpecs inner+ let transposed =+ transpose+ ( V.map+ ( \case+ CKTuple tupleVec -> tupleVec+ other ->+ error+ ( "expected type: " ++ show other+ ++ "in the column:"+ ++ show col_name+ )+ )+ items+ )+ if V.length spec_arr /= V.length transposed+ then+ error $+ "length of the given array does not match, column name = "+ ++ show col_name+ else do+ V.zipWithM_ (writeColumn ctx col_name) spec_arr transposed++--------------------------------------------------------------------------------------+readEnum :: Int -> ByteString -> Reader (Vector ClickhouseType)+readEnum n_rows spec = do+ let l = BS.length spec+ innerSpec =+ if "Enum8" `isPrefixOf` spec+ then BS.take (l - 7) (BS.drop 6 spec)+ else BS.take (l - 8) (BS.drop 7 spec) -- otherwise it is `Enum16`+ pres_pecs = getSpecs innerSpec+ specs =+ (\(name, Just (n, _)) -> (n, name))+ <$> ((\[x, y] -> (x, readInt y)) . BS.splitWith (== EQUAL) <$> pres_pecs) --61 means '='+ specsMap = Map.fromList specs+ if "Enum8" `isPrefixOf` spec+ then do+ values <- V.replicateM n_rows readBinaryInt8+ return $ CKString . (specsMap Map.!) . fromIntegral <$> values+ else do+ values <- V.replicateM n_rows readBinaryInt16+ return $ CKString . (specsMap Map.!) . fromIntegral <$> values++writeEnum :: ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeEnum col_name spec items = do+ let l = BS.length spec+ innerSpec =+ if "Enum8" `isPrefixOf` spec+ then BS.take (l - 7) (BS.drop 6 spec)+ else BS.take (l - 8) (BS.drop 7 spec)+ pres_pecs = getSpecs innerSpec+ specs =+ (\(name, Just (n, _)) -> (name, n))+ <$> ((\[x, y] -> (x, readInt y)) . BS.splitWith (== EQUAL) . BS.filter (/= QUOTE) <$> pres_pecs) --61 is '='+ specsMap = Map.fromList specs+ V.mapM_+ ( \case+ CKString str ->+ ( if "Enum8" `isPrefixOf` spec+ then writeBinaryInt8 $ fromIntegral $ specsMap Map.! str+ else writeBinaryInt16 $ fromIntegral $ specsMap Map.! str+ )+ CKNull ->+ if "Enum8" `isPrefixOf` spec+ then writeBinaryInt8 0+ else writeBinaryInt16 0+ _ -> error $ typeMismatchError col_name+ )+ items++----------------------------------------------------------------------+readDate :: Int -> Reader (Vector ClickhouseType)+readDate n_rows = do+ let epoch_start = fromGregorian 1970 1 1+ days <- V.replicateM n_rows readBinaryUInt16+ let dates = V.map (\x -> addDays (fromIntegral x) epoch_start) days+ toTriple = V.map toGregorian dates+ toCK = V.map (\(y, m, d) -> CKDate y m d) toTriple+ return toCK++writeDate :: ByteString -> Vector ClickhouseType -> Writer Builder+writeDate col_name items = do+ let epoch_start = fromGregorian 1970 1 1+ let serialize =+ V.map+ ( \case+ CKDate y m d -> diffDays (fromGregorian y m d) epoch_start+ _ ->+ error $+ "unexpected type in the column: " ++ show col_name+ ++ " whose type should be Date"+ )+ items+ V.mapM_ (writeBinaryInt16 . fromIntegral) serialize++--------------------------------------------------------------------------------------+readDecimal :: Int -> ByteString -> Reader (Vector ClickhouseType)+readDecimal n_rows spec = do+ let l = BS.length spec+ let inner_spec = getSpecs $ BS.take (l - 9) (BS.drop 8 spec)+ let (specific, Just (scale, _)) = case inner_spec of+ [] -> error "No spec"+ [scale'] ->+ if "Decimal32" `isPrefixOf` spec+ then (readDecimal32, readInt scale')+ else+ if "Decimal64" `isPrefixOf` spec+ then (readDecimal64, readInt scale')+ else (readDecimal128, readInt scale')+ [precision', scale'] -> do+ let Just (precision, _) = readInt precision'+ if precision <= 9 || "Decimal32" `isPrefixOf` spec+ then (readDecimal32, readInt scale')+ else+ if precision <= 18 || "Decimal64" `isPrefixOf` spec+ then (readDecimal64, readInt scale')+ else (readDecimal128, readInt scale')+ raw <- specific n_rows+ let final = fmap (trans scale) raw+ return final+ where+ readDecimal32 :: Int -> Reader (Vector ClickhouseType)+ readDecimal32 n_rows = readIntColumn n_rows "Int32"+ + readDecimal64 :: Int -> Reader (Vector ClickhouseType)+ readDecimal64 n_rows = readIntColumn n_rows "Int64"++ readDecimal128 :: Int -> Reader (Vector ClickhouseType)+ readDecimal128 n_rows =+ V.replicateM n_rows $ do+ lo <- readBinaryUInt64+ hi <- readBinaryUInt64+ return $ CKUInt128 lo hi++ trans :: Int -> ClickhouseType -> ClickhouseType+ trans scale (CKInt32 x) = CKDecimal32 (fromIntegral x / fromIntegral (10 ^ scale))+ trans scale (CKInt64 x) = CKDecimal64 (fromIntegral x / fromIntegral (10 ^ scale))+ trans scale (CKUInt128 lo hi) = CKDecimal128 (word128_division hi lo scale)++writeDecimal :: ByteString -> ByteString -> Vector ClickhouseType -> Writer Builder+writeDecimal col_name spec items = do+ let l = BS.length spec+ let inner_specs = getSpecs $ BS.take (l - 9) (BS.drop 8 spec)+ let (specific, Just (pre_scale, _)) = case inner_specs of+ [] -> error "No spec"+ [scale'] ->+ if "Decimal32" `isPrefixOf` spec+ then (writeDecimal32, readInt scale')+ else+ if "Decimal64" `isPrefixOf` spec+ then (writeDecimal64, readInt scale')+ else (writeDecimal128, readInt scale')+ [precision', scale'] -> do+ let Just (precision, _) = readInt precision'+ if precision <= 9 || "Decimal32" `isPrefixOf` spec+ then (writeDecimal32, readInt scale')+ else+ if precision <= 18 || "Decimal64" `isPrefixOf` spec+ then (writeDecimal64, readInt scale')+ else (writeDecimal128, readInt scale')+ let scale = 10 ^ pre_scale+ specific scale items+ where+ writeDecimal32 :: Int -> Vector ClickhouseType -> Writer Builder+ writeDecimal32 scale vec =+ V.mapM_+ ( \case+ CKDecimal32 f32 -> writeBinaryInt32 $ fromIntegral $ floor $ (f32 * fromIntegral scale)+ _ -> error $ typeMismatchError col_name+ )+ vec+ writeDecimal64 :: Int -> Vector ClickhouseType -> Writer Builder+ writeDecimal64 scale vec =+ V.mapM_+ ( \case+ CKDecimal64 f64 -> writeBinaryInt32 $ fromIntegral $ floor $ (f64 * fromIntegral scale)+ _ -> error $ typeMismatchError col_name+ )+ vec+ writeDecimal128 :: Int -> Vector ClickhouseType -> Writer Builder+ writeDecimal128 scale items = do+ V.mapM_+ ( \case+ CKDecimal128 x -> do+ if x >= 0+ then do+ writeBinaryUInt64 $ low_bits_128 x scale+ writeBinaryUInt64 $ hi_bits_128 x scale+ else do+ writeBinaryUInt64 $ low_bits_negative_128 (- x) scale+ writeBinaryUInt64 $ hi_bits_negative_128 (- x) scale+ )+ items++foreign import ccall unsafe "bigint.h word128_division" word128_division :: Word64 -> Word64 -> Int -> Double++foreign import ccall unsafe "bigint.h low_bits_128" low_bits_128 :: Double -> Int -> Word64++foreign import ccall unsafe "bigint.h hi_bits_128" hi_bits_128 :: Double -> Int -> Word64++foreign import ccall unsafe "bigint.h low_bits_negative_128" low_bits_negative_128 :: Double -> Int -> Word64++foreign import ccall unsafe "bigint.h hi_bits_negative_128" hi_bits_negative_128 :: Double -> Int -> Word64+----------------------------------------------------------------------------------------------+readIPv4 :: Int -> Reader (Vector ClickhouseType)+readIPv4 n_rows = V.replicateM n_rows (CKIPv4 . ip4ToOctets . IP4 <$> readBinaryUInt32)++readIPv6 :: Int -> Reader (Vector ClickhouseType)+readIPv6 n_rows = V.replicateM n_rows (CKIPv6 . ip6ToWords . IP6 <$> readBinaryUInt128)++writeIPv4 :: ByteString -> Vector ClickhouseType -> Writer Builder+writeIPv4 col_name =+ V.mapM_+ ( \case+ CKIPv4 (w1, w2, w3, w4) ->+ writeBinaryUInt32 $+ unIP4 $+ ip4FromOctets w1 w2 w3 w4+ CKNull -> writeBinaryInt32 0+ _ -> error $ typeMismatchError col_name+ )++writeIPv6 :: ByteString -> Vector ClickhouseType -> Writer Builder+writeIPv6 col_name =+ V.mapM_+ ( \case+ CKIPv6 (w1, w2, w3, w4, w5, w6, w7, w8) ->+ writeBinaryUInt128 $+ unIP6 $+ ip6FromWords w1 w2 w3 w4 w5 w6 w7 w8+ CKNull -> writeBinaryUInt64 0+ _ -> error $ typeMismatchError col_name+ )++----------------------------------------------------------------------------------------------+readSimpleAggregateFunction :: ServerInfo -> Int -> ByteString -> Reader (Vector ClickhouseType)+readSimpleAggregateFunction server_info n_rows spec = do+ let l = BS.length spec+ let [func, cktype] = getSpecs $ BS.take (l - 25) (BS.drop 24 spec)+ readColumn server_info n_rows cktype+----------------------------------------------------------------------------------------------+readUUID :: Int -> Reader (Vector ClickhouseType)+readUUID n_rows = do+ V.replicateM n_rows $ do+ w2 <- readBinaryUInt32+ w1 <- readBinaryUInt32+ w3 <- readBinaryUInt32+ w4 <- readBinaryUInt32+ return $+ CKString $+ C8.pack $+ UUID.toString $ UUID.fromWords w1 w2 w3 w4++writeUUID :: ByteString -> Vector ClickhouseType -> Writer Builder+writeUUID col_name =+ V.mapM_+ ( \case+ CKString uuid_str -> do+ case UUID.fromString $ C8.unpack uuid_str of+ Nothing ->+ error $+ "UUID parsing error in the column"+ ++ show col_name+ ++ " wrong data: "+ ++ show uuid_str+ Just uuid -> do+ let (w2, w1, w3, w4) = UUID.toWords uuid+ writeBinaryUInt32 w1+ writeBinaryUInt32 w2+ writeBinaryUInt32 w3+ writeBinaryUInt32 w4+ CKNull -> do+ writeBinaryUInt64 0+ writeBinaryUInt64 0+ )++----------------------------------------------------------------------------------------------+---Helpers+#define COMMA 44+#define SPACE 32++-- | Get rid of commas and spaces+getSpecs :: ByteString -> [ByteString]+getSpecs str = BS.splitWith (== COMMA) (BS.filter (/= SPACE) str)++transpose :: Vector (Vector ClickhouseType) -> Vector (Vector ClickhouseType)+transpose = rotate + where+ rotate matrix =+ let transposedList = List.transpose (V.toList <$> V.toList matrix)+ toVector = V.fromList <$> V.fromList transposedList+ in toVector++typeMismatchError :: ByteString -> String+typeMismatchError col_name = "Type mismatch in the column " ++ show col_name++-- | print in format+putStrLn :: Vector (Vector ClickhouseType) -> IO ()+putStrLn v = C8.putStrLn $ BS.intercalate "\n" $ V.toList $ V.map to_str v+ where+ to_str :: Vector ClickhouseType -> ByteString+ to_str row = BS.intercalate "," $ V.toList $ V.map help row++ help :: ClickhouseType -> ByteString+ help (CKString s) = s+ help (CKDecimal64 n) = C8.pack $ show n+ help (CKDecimal32 n) = C8.pack $ show n+ help (CKDecimal n) = C8.pack $ show n+ help (CKInt8 n) = C8.pack $ show n+ help (CKInt16 n) = C8.pack $ show n+ help (CKInt32 n) = C8.pack $ show n+ help (CKInt64 n) = C8.pack $ show n+ help (CKUInt8 n) = C8.pack $ show n+ help (CKUInt16 n) = C8.pack $ show n+ help (CKUInt32 n) = C8.pack $ show n+ help (CKUInt64 n) = C8.pack $ show n+ help (CKTuple xs) = "(" <> to_str xs <> ")"+ help (CKArray xs) = "[" <> to_str xs <> "]"+ help CKNull = "null"+ help (CKIPv4 ip4) = C8.pack $ show ip4+ help (CKIPv6 ip6) = C8.pack $ show ip6+ help (CKDate y m d) =+ C8.pack (show y)+ <> "-"+ <> C8.pack (show m)+ <> "-"+ <> C8.pack (show d)
+ src/Database/ClickHouseDriver/Connection.hs view
@@ -0,0 +1,554 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE BlockArguments #-}++-- | This module contains the implementations of communication with Clickhouse server.+-- Most of functions are for internal use. +-- User should just use Database.ClickHouseDriver.+--++module Database.ClickHouseDriver.Connection+ ( tcpConnect,+ sendQuery,+ receiveData,+ sendData,+ receiveResult,+ closeConnection,+ processInsertQuery,+ ping',+ versionTuple,+ sendCancel,+ )+where++import qualified Database.ClickHouseDriver.Block as Block+import qualified Database.ClickHouseDriver.ClientProtocol as Client+import Database.ClickHouseDriver.Column (ClickhouseType, transpose)+import Database.ClickHouseDriver.Defines+ ( _BUFFER_SIZE,+ _CLIENT_NAME,+ _CLIENT_REVISION,+ _CLIENT_VERSION_MAJOR,+ _CLIENT_VERSION_MINOR,+ _DBMS_MIN_REVISION_WITH_CLIENT_INFO,+ _DBMS_MIN_REVISION_WITH_QUOTA_KEY_IN_CLIENT_INFO,+ _DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME,+ _DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE,+ _DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES,+ _DBMS_MIN_REVISION_WITH_VERSION_PATCH,+ _DBMS_NAME,+ _DEFAULT_INSERT_BLOCK_SIZE,+ _STRINGS_ENCODING,+ )+import qualified Database.ClickHouseDriver.Error as Error+import qualified Database.ClickHouseDriver.QueryProcessingStage as Stage+import qualified Database.ClickHouseDriver.ServerProtocol as Server+import Database.ClickHouseDriver.Types+ ( Block (ColumnOrientedBlock, cdata),+ CKResult (CKResult),+ ClientInfo (ClientInfo),+ ClientSetting+ ( ClientSetting,+ insert_block_size,+ strings_as_bytes,+ strings_encoding+ ),+ Context (Context, client_info, client_setting, server_info),+ Interface (HTTP),+ Packet+ ( Block,+ EndOfStream,+ ErrorMessage,+ Hello,+ MultiString,+ Progress,+ StreamProfileInfo,+ queryData+ ),+ QueryInfo,+ ServerInfo (..),+ TCPConnection (..),+ getDefaultClientInfo,+ getServerInfo,+ readBlockStreamProfileInfo,+ readProgress,+ storeProfile,+ storeProgress,+ )+import Database.ClickHouseDriver.IO.BufferedReader+ ( Buffer (socket),+ Reader,+ createBuffer,+ readBinaryStr,+ readVarInt,+ refill,+ )+import Database.ClickHouseDriver.IO.BufferedWriter+ ( MonoidMap,+ Writer,+ writeBinaryStr,+ writeVarUInt,+ )+import Control.Monad.Loops (iterateWhile)+import Control.Monad.State.Lazy (get, runStateT)+import Control.Monad.Writer (WriterT (runWriterT), execWriterT)+import Control.Monad (when)+import Data.Maybe (fromMaybe)+import Data.ByteString (ByteString)+import Data.Foldable (forM_)+import Data.ByteString.Builder+ ( Builder,+ toLazyByteString,+ )+import Data.ByteString.Char8 (unpack)+import qualified Data.List as List (transpose)+import Data.List.Split (chunksOf)+import Data.Vector ((!))+import qualified Data.Vector as V+ ( concat,+ fromList,+ length,+ map,+ )+import qualified Network.Simple.TCP as TCP+ ( closeSock,+ connectSock,+ sendLazy,+ )+import Network.Socket (Socket)+import System.Timeout (timeout)++--Debug+--import Debug.Trace ( trace )++-- | This module mainly focuses how to make connection+-- | to clickhouse database and protocols to send and receive data+versionTuple :: ServerInfo -> (Word, Word, Word)+versionTuple (ServerInfo _ major minor patch _ _ _) = (major, minor, patch)++-- | internal implementation for ping test. +ping' :: Int+ -- ^ Time limit+ ->TCPConnection+ -- ^ host name, port number, and socket are needed+ ->IO (Maybe String)+ -- ^ response `ping`, or nothing indicating server is not properly connected.+ping' timeLimit TCPConnection {tcpHost = host, tcpPort = port, tcpSocket = sock} =+ timeout timeLimit $ do+ r <- execWriterT $ writeVarUInt Client._PING+ TCP.sendLazy sock (toLazyByteString r)+ buf <- createBuffer 1024 sock+ (packet_type, _) <- runStateT (iterateWhile (== Server._PROGRESS) readVarInt) buf+ if packet_type /= Server._PONG+ then do+ let p_type = Server.toString $ fromIntegral packet_type+ let report =+ "Unexpected packet from server " <> show host <> ":"+ <> show port+ <> ", expected "+ <> "Pong!"+ <> ", got "+ <> show p_type+ return $+ show+ Error.ServerException+ { code = Error._UNEXPECTED_PACKET_FROM_SERVER,+ message = report,+ nested = Nothing+ }+ else return "PONG!"++-- | send hello has to make for every new connection environment.+sendHello :: (ByteString, ByteString, ByteString)+ -- ^ (database name, username, password)+ ->Socket+ -- ^ socket connected to Clickhouse server + ->IO ()+sendHello (database, username, password) sock = do+ (_, w) <- runWriterT writeHello+ TCP.sendLazy sock (toLazyByteString w)+ where+ writeHello :: Writer Builder+ writeHello = do+ writeVarUInt Client._HELLO+ writeBinaryStr ("ClickHouse " <> _CLIENT_NAME)+ writeVarUInt _CLIENT_VERSION_MAJOR+ writeVarUInt _CLIENT_VERSION_MINOR+ writeVarUInt _CLIENT_REVISION+ writeBinaryStr database+ writeBinaryStr username+ writeBinaryStr password++-- | receive server information if connection is successful, otherwise it would receive error message.+receiveHello :: Buffer+ -- ^ Read `Hello` response from server+ ->IO (Either ByteString ServerInfo)+ -- ^ Either error message or server information will be received.+receiveHello buf = do+ (res, _) <- runStateT receiveHello' buf+ return res+ where+ receiveHello' :: Reader (Either ByteString ServerInfo)+ receiveHello' = do+ packet_type <- readVarInt+ if packet_type == Server._HELLO+ then do+ server_name <- readBinaryStr+ server_version_major <- readVarInt+ server_version_minor <- readVarInt+ server_revision <- readVarInt+ server_timezone <-+ if server_revision >= _DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE+ then do Just <$> readBinaryStr+ else return Nothing+ server_display_name <-+ if server_revision >= _DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME+ then do+ readBinaryStr+ else return ""+ server_version_dispatch <-+ if server_revision >= _DBMS_MIN_REVISION_WITH_VERSION_PATCH+ then do+ readVarInt+ else return server_revision+ return $+ Right+ ServerInfo+ { name = server_name,+ version_major = server_version_major,+ version_minor = server_version_minor,+ version_patch = server_version_dispatch,+ revision = server_revision,+ timezone = server_timezone,+ display_name = server_display_name+ }+ else+ if packet_type == Server._EXCEPTION+ then do+ e <- readBinaryStr+ e2 <- readBinaryStr+ e3 <- readBinaryStr+ return $ Left ("exception" <> e <> " " <> e2 <> " " <> e3)+ else return $ Left "Error"++-- | connect to database through TCP port, used in Client module.+tcpConnect ::+ -- | host name to connect+ ByteString ->+ -- | port name to connect. Default would be 8123+ ByteString ->+ -- | username. Default would be "default"+ ByteString ->+ -- | password. Default would be empty string+ ByteString ->+ -- | database. Default would be "default"+ ByteString ->+ -- | choose if send and receive data in compressed form. Default would be False.+ Bool ->+ IO (Either String TCPConnection)+tcpConnect host port user password database compression = do+ (sock, sockaddr) <- TCP.connectSock (unpack host) (unpack port)+ sendHello (database, user, password) sock+ let isCompressed = if compression then Client._COMPRESSION_ENABLE else Client._COMPRESSION_DISABLE++ buf <- createBuffer _BUFFER_SIZE sock+ hello <- receiveHello buf+ case hello of+ Right x ->+ return $+ Right+ TCPConnection+ { tcpHost = host,+ tcpPort = port,+ tcpUsername = user,+ tcpPassword = password,+ tcpSocket = sock,+ tcpSockAdrr = sockaddr,+ context =+ Context+ { server_info = Just x,+ client_info = Nothing,+ client_setting =+ Just $+ ClientSetting+ { insert_block_size = _DEFAULT_INSERT_BLOCK_SIZE,+ strings_as_bytes = False,+ strings_encoding = _STRINGS_ENCODING+ }+ },+ tcpCompression = isCompressed+ }+ Left "Exception" -> return $ Left "exception"+ Left x -> do+ print ("error is " <> x)+ TCP.closeSock sock+ return $ Left "Connection error"++sendQuery :: TCPConnection+ -- ^ To get socket and server info+ ->ByteString+ -- ^ SQL statement+ ->Maybe ByteString+ -- ^ query_id if any+ ->IO ()+sendQuery TCPConnection {context = Context {server_info = Nothing}} _ _ = error "Empty server info"+sendQuery+ TCPConnection+ { tcpCompression = comp,+ tcpSocket = sock,+ context =+ Context+ { server_info = Just info+ }+ }+ query+ query_id = do+ (_, r) <- runWriterT $ do+ writeVarUInt Client._QUERY+ writeBinaryStr $ fromMaybe "" query_id+ let revision' = revision info+ when (revision' >= _DBMS_MIN_REVISION_WITH_CLIENT_INFO)+ $ do + let client_info+ = getDefaultClientInfo (_DBMS_NAME <> " " <> _CLIENT_NAME)+ writeInfo client_info info+ writeVarUInt 0 -- TODO add write settings+ writeVarUInt Stage._COMPLETE+ writeVarUInt comp+ writeBinaryStr query+ TCP.sendLazy sock (toLazyByteString r)++sendData :: TCPConnection+ -- ^ To get socket and context+ ->ByteString+ -- ^ table name + ->Maybe Block+ -- ^ a block data if any+ ->IO ()+sendData TCPConnection {tcpSocket = sock, context = ctx} table_name maybe_block = do+ -- TODO: ADD REVISION+ let info = Block.defaultBlockInfo+ r <- execWriterT $ do+ writeVarUInt Client._DATA+ writeBinaryStr table_name+ case maybe_block of+ Nothing -> do+ Block.writeInfo info+ writeVarUInt 0 -- #col+ writeVarUInt 0 -- #row+ Just block -> do+ Block.writeBlockOutputStream ctx block+ TCP.sendLazy sock $ toLazyByteString r++-- | Cancel last query sent to server+sendCancel :: TCPConnection -> IO ()+sendCancel TCPConnection {tcpSocket = sock} = do+ c <- execWriterT $ writeVarUInt Client._CANCEL+ TCP.sendLazy sock (toLazyByteString c)++processInsertQuery ::+ -- | source+ TCPConnection ->+ -- | query without data+ ByteString ->+ -- | query id+ Maybe ByteString ->+ -- | data in Haskell type.+ [[ClickhouseType]] ->+ -- | return 1 if insertion successfully completed+ IO ByteString+processInsertQuery+ tcp@TCPConnection {tcpSocket = sock, context = Context {client_setting = client_setting}}+ query_without_data+ query_id+ items = do+ sendQuery tcp query_without_data query_id+ sendData tcp "" Nothing+ buf <- createBuffer 2048 sock+ let info = case getServerInfo tcp of+ Nothing -> error "empty server info"+ Just s -> s+ (sample_block, _) <-+ runStateT+ ( iterateWhile+ ( \case+ Block {..} -> False+ MultiString _ -> True+ Hello -> True+ x -> error ("unexpected packet type: " ++ show x)+ )+ $ receivePacket info+ )+ buf+ case client_setting of+ Nothing -> error "empty client settings"+ Just ClientSetting {insert_block_size = siz} -> do+ let chunks = chunksOf (fromIntegral siz) items+ mapM_+ ( \chunk -> do+ let vectorized = V.map V.fromList (V.fromList $ List.transpose chunk)+ let dataBlock = case sample_block of+ Block+ typeinfo@ColumnOrientedBlock+ { columns_with_type = cwt+ } ->+ typeinfo+ { cdata = vectorized+ }+ x -> error ("unexpected packet type: " ++ show x)+ sendData tcp "" (Just dataBlock)+ )+ chunks+ sendData tcp "" Nothing+ buf2 <- refill buf+ runStateT (receivePacket info) buf2+ return "1"++-- | Read data from stream.+receiveData :: ServerInfo -> Reader Block.Block+receiveData info@ServerInfo {revision = revision} = do+ _ <-+ if revision >= _DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES+ then readBinaryStr+ else return ""+ Block.readBlockInputStream info+++-- | Transform received query data into Clickhouse type+receiveResult :: ServerInfo+ -- ^ Server information+ -> QueryInfo+ -- ^ Query information+ -> Reader (Either String CKResult)+ -- ^ Receive either error message or query result.+receiveResult info query_info = do+ packets <- packetGen+ let onlyDataPacket = filter isBlock packets+ let errors = (\(ErrorMessage str) -> str) <$> filter isError packets+ case errors of+ [] -> do+ let dataVectors = Database.ClickHouseDriver.Column.transpose . Block.cdata . queryData <$> onlyDataPacket+ let newQueryInfo = Prelude.foldl updateQueryInfo query_info packets+ return $ Right $ CKResult (V.concat dataVectors) newQueryInfo+ xs -> do+ return $ Left $ Prelude.concat xs+ where+ updateQueryInfo :: QueryInfo -> Packet -> QueryInfo+ updateQueryInfo q (Progress prog) =+ storeProgress q prog+ updateQueryInfo q (StreamProfileInfo profile) =+ storeProfile q profile+ updateQueryInfo q _ = q++ isError :: Packet -> Bool+ isError (ErrorMessage _) = True+ isError _ = False++ isBlock :: Packet -> Bool+ isBlock Block {queryData = Block.ColumnOrientedBlock {cdata = d}} =+ V.length d > 0 && V.length (d ! 0) > 0+ isBlock _ = False++ packetGen :: Reader [Packet]+ packetGen = do+ packet <- receivePacket info+ case packet of+ EndOfStream -> return []+ error@(ErrorMessage _) -> return [error]+ _ -> do+ next <- packetGen+ return (packet : next)++-- | Receive data packet from server +receivePacket :: ServerInfo -> Reader Packet+receivePacket info = do+ packet_type <- readVarInt+ -- The pattern matching does not support match with variable name,+ -- so here we use number instead.+ case packet_type of+ 1 -> receiveData info >>= (return . Block) -- Data+ 2 -> Error.readException Nothing >>= (return . ErrorMessage . show) -- Exception+ 3 -> readProgress (revision info) >>= (return . Progress) -- Progress+ 5 -> return EndOfStream -- End of Stream+ 6 -> readBlockStreamProfileInfo >>= (return . StreamProfileInfo) --Profile+ 7 -> receiveData info >>= (return . Block) -- Total+ 8 -> receiveData info >>= (return . Block) -- Extreme+ -- 10 -> return undefined -- Log+ 11 -> do+ -- MultiStrings message+ first <- readBinaryStr+ second <- readBinaryStr+ return $ MultiString (first, second)+ 0 -> return Hello -- Hello+ _ -> do+ closeBufferSocket+ error $+ show+ Error.ServerException+ { code = Error._UNKNOWN_PACKET_FROM_SERVER,+ message = "Unknown packet from server",+ nested = Nothing+ }++closeConnection :: TCPConnection -> IO ()+closeConnection TCPConnection {tcpSocket = sock} = TCP.closeSock sock++-- | write client information and server infomation to protocols+writeInfo ::+ (MonoidMap ByteString w) =>+ ClientInfo ->+ ServerInfo ->+ Writer w+writeInfo+ ( ClientInfo+ client_name+ interface+ client_version_major+ client_version_minor+ client_version_patch+ client_revision+ initial_user+ initial_query_id+ initial_address+ quota_key+ query_kind+ )+ ServerInfo {revision = server_revision, display_name = host_name}+ | server_revision < _DBMS_MIN_REVISION_WITH_CLIENT_INFO =+ error "Method writeInfo is called for unsupported server revision"+ | otherwise = do+ writeVarUInt 1+ writeBinaryStr initial_user+ writeBinaryStr initial_query_id+ writeBinaryStr initial_address+ writeVarUInt (if interface == HTTP then 0 else 1)+ writeBinaryStr "" -- os_user. Seems that haskell modules don't have support of getting system username yet.+ writeBinaryStr host_name+ writeBinaryStr _CLIENT_NAME --client_name+ writeVarUInt client_version_major+ writeVarUInt client_version_minor+ writeVarUInt client_revision+ when+ (server_revision >= _DBMS_MIN_REVISION_WITH_QUOTA_KEY_IN_CLIENT_INFO)+ $ writeBinaryStr quota_key+ when + (server_revision >= _DBMS_MIN_REVISION_WITH_VERSION_PATCH)+ $ writeVarUInt client_version_patch++-------------------------------------------------------------------------------------------------------------------+---Helpers+{-# INLINE closeBufferSocket #-}+closeBufferSocket :: Reader ()+closeBufferSocket = do+ buf <- get+ let sock = Database.ClickHouseDriver.IO.BufferedReader.socket buf+ forM_ sock TCP.closeSock
+ src/Database/ClickHouseDriver/Defines.hs view
@@ -0,0 +1,130 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++-- | This module defines constants for internal use. ++module Database.ClickHouseDriver.Defines where++import Data.ByteString.Internal++{-# INLINE _DEFAULT_PORT #-}+_DEFAULT_PORT = "9000"++{-# INLINE _DEFAULT_SECURE_PORT #-}+_DEFAULT_SECURE_PORT = "9440"++{-# INLINE _DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES #-}+_DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES = 50264 :: Word++{-# INLINE _DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS #-}+_DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS = 51554 :: Word++{-# INLINE _DBMS_MIN_REVISION_WITH_BLOCK_INFO #-}+_DBMS_MIN_REVISION_WITH_BLOCK_INFO = 51903++-- Legacy above.++{-# INLINE _DBMS_MIN_REVISION_WITH_CLIENT_INFO #-}+_DBMS_MIN_REVISION_WITH_CLIENT_INFO = 54032 :: Word++{-# INLINE _DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE #-}+_DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE = 54058 :: Word++{-# INLINE _DBMS_MIN_REVISION_WITH_QUOTA_KEY_IN_CLIENT_INFO #-}+_DBMS_MIN_REVISION_WITH_QUOTA_KEY_IN_CLIENT_INFO = 54060 :: Word++_DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME = 54372 :: Word++_DBMS_MIN_REVISION_WITH_VERSION_PATCH = 54401 :: Word++_DBMS_MIN_REVISION_WITH_SERVER_LOGS = 54406 :: Word++_DBMS_MIN_REVISION_WITH_COLUMN_DEFAULTS_METADATA = 54410 :: Word++_DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO :: Word+_DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO = 54420 :: Word++_DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS :: Word+_DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS = 54429 :: Word++-- Timeouts+_DBMS_DEFAULT_CONNECT_TIMEOUT_SEC :: Integer+_DBMS_DEFAULT_CONNECT_TIMEOUT_SEC = 10++_DBMS_DEFAULT_TIMEOUT_SEC :: Integer+_DBMS_DEFAULT_TIMEOUT_SEC = 300++_DBMS_DEFAULT_SYNC_REQUEST_TIMEOUT_SEC :: Integer+_DBMS_DEFAULT_SYNC_REQUEST_TIMEOUT_SEC = 5++_DEFAULT_COMPRESS_BLOCK_SIZE :: Integer+_DEFAULT_COMPRESS_BLOCK_SIZE = 1048576++_DEFAULT_INSERT_BLOCK_SIZE :: Word+_DEFAULT_INSERT_BLOCK_SIZE = 1048576 :: Word++_DBMS_NAME :: ByteString+_DBMS_NAME = "ClickHouse" :: ByteString++{-# INLINE _CLIENT_NAME #-}+_CLIENT_NAME :: ByteString+_CLIENT_NAME = "haskell-driver" :: ByteString++{-# INLINE _CLIENT_VERSION_MAJOR #-}+_CLIENT_VERSION_MAJOR :: Word+_CLIENT_VERSION_MAJOR = 18 :: Word++{-# INLINE _CLIENT_VERSION_MINOR #-}+_CLIENT_VERSION_MINOR :: Word+_CLIENT_VERSION_MINOR = 10 :: Word++{-# INLINE _CLIENT_VERSION_PATCH #-}+_CLIENT_VERSION_PATCH :: Word+_CLIENT_VERSION_PATCH = 3 :: Word++{-# INLINE _CLIENT_REVISION #-}+_CLIENT_REVISION :: Word+_CLIENT_REVISION = 54429 :: Word++_STRINGS_ENCODING :: ByteString+_STRINGS_ENCODING = "utf-8" :: ByteString++{-# INLINE _DEFAULT_HTTP_PORT #-}+_DEFAULT_HTTP_PORT :: Int+_DEFAULT_HTTP_PORT = 8123 :: Int++{-# INLINE _BUFFER_SIZE #-}+_BUFFER_SIZE :: Int+_BUFFER_SIZE = 1048576 :: Int++{-# INLINE _DEFAULT_HOST #-}+_DEFAULT_HOST :: [Char]+_DEFAULT_HOST = "localhost"++{-# INLINE _DEFAULT_PING_WAIT_TIME #-}+_DEFAULT_PING_WAIT_TIME :: Int+_DEFAULT_PING_WAIT_TIME = 10000 :: Int++{-# INLINE _DEFAULT_USERNAME #-}+_DEFAULT_USERNAME :: ByteString+_DEFAULT_USERNAME = "default" :: ByteString++{-# INLINE _DEFAULT_HOST_NAME #-}+_DEFAULT_HOST_NAME :: ByteString+_DEFAULT_HOST_NAME = "localhost" :: ByteString++{-# INLINE _DEFAULT_PASSWORD #-}+_DEFAULT_PASSWORD :: ByteString+_DEFAULT_PASSWORD = "" :: ByteString++{-# INLINE _DEFAULT_PORT_NAME #-}+_DEFAULT_PORT_NAME :: ByteString+_DEFAULT_PORT_NAME = "9000" :: ByteString++{-# INLINE _DEFAULT_DATABASE#-}+_DEFAULT_DATABASE :: ByteString+_DEFAULT_DATABASE = "default" :: ByteString++{-# INLINE _DEFAULT_COMPRESSION_SETTING #-}+_DEFAULT_COMPRESSION_SETTING :: Bool+_DEFAULT_COMPRESSION_SETTING = False
+ src/Database/ClickHouseDriver/Error.hs view
@@ -0,0 +1,804 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.+---------------------------------------------------------------------------+-- This file contains the details of handling error messages from server.+-- For internal use only+{-# LANGUAGE OverloadedStrings #-}++module Database.ClickHouseDriver.Error+ ( readException,+ ClickhouseException (..),+ _UNEXPECTED_PACKET_FROM_SERVER,+ _UNKNOWN_PACKET_FROM_SERVER,+ )+where++import Database.ClickHouseDriver.IO.BufferedReader+ ( Reader,+ readBinaryInt32,+ readBinaryStr,+ readBinaryUInt8,+ )+import Data.ByteString.Char8 (unpack)++--import Debug.Trace++data ClickhouseException = ServerException+ { message :: String,+ code :: !Integer,+ nested :: Maybe ClickhouseException+ }++instance Show ClickhouseException where+ show (ServerException message code nested) =+ "Code " ++ show code ++ "."+ ++ ( case nested of+ Nothing -> ""+ Just s -> show s+ )+ ++ " "+ ++ message++readException :: Maybe String -> Reader ClickhouseException+readException additional = do+ code <- readBinaryInt32+ name <- readBinaryStr+ messange <- readBinaryStr+ stack_trace <- readBinaryStr+ has_nested <- (== 1) <$> readBinaryUInt8+ let hasAdditional =+ ( case additional of+ Nothing -> ""+ Just msg -> msg ++ "."+ )+ ++ if name /= "DB::Exception"+ then unpack name+ else "" ++ "."+ let new_message = hasAdditional <> show messange <> ". Stack trace:\n\n" <> show stack_trace+ if has_nested+ then do+ nested <- readException Nothing+ return $ ServerException new_message (fromIntegral code) (Just nested)+ else do+ return $ ServerException new_message (fromIntegral code) Nothing++_UNSUPPORTED_METHOD = 1++_UNSUPPORTED_PARAMETER = 2++_UNEXPECTED_END_OF_FILE = 3++_EXPECTED_END_OF_FILE = 4++_CANNOT_PARSE_TEXT = 6++_INCORRECT_NUMBER_OF_COLUMNS = 7++_THERE_IS_NO_COLUMN = 8++_SIZES_OF_COLUMNS_DOESNT_MATCH = 9++_NOT_FOUND_COLUMN_IN_BLOCK = 10++_POSITION_OUT_OF_BOUND = 11++_PARAMETER_OUT_OF_BOUND = 12++_SIZES_OF_COLUMNS_IN_TUPLE_DOESNT_MATCH = 13++_DUPLICATE_COLUMN = 15++_NO_SUCH_COLUMN_IN_TABLE = 16++_DELIMITER_IN_STRING_LITERAL_DOESNT_MATCH = 17++_CANNOT_INSERT_ELEMENT_INTO_CONSTANT_COLUMN = 18++_SIZE_OF_FIXED_STRING_DOESNT_MATCH = 19++_NUMBER_OF_COLUMNS_DOESNT_MATCH = 20++_CANNOT_READ_ALL_DATA_FROM_TAB_SEPARATED_INPUT = 21++_CANNOT_PARSE_ALL_VALUE_FROM_TAB_SEPARATED_INPUT = 22++_CANNOT_READ_FROM_ISTREAM = 23++_CANNOT_WRITE_TO_OSTREAM = 24++_CANNOT_PARSE_ESCAPE_SEQUENCE = 25++_CANNOT_PARSE_QUOTED_STRING = 26++_CANNOT_PARSE_INPUT_ASSERTION_FAILED = 27++_CANNOT_PRINT_FLOAT_OR_DOUBLE_NUMBER = 28++_CANNOT_PRINT_INTEGER = 29++_CANNOT_READ_SIZE_OF_COMPRESSED_CHUNK = 30++_CANNOT_READ_COMPRESSED_CHUNK = 31++_ATTEMPT_TO_READ_AFTER_EOF = 32++_CANNOT_READ_ALL_DATA = 33++_TOO_MANY_ARGUMENTS_FOR_FUNCTION = 34++_TOO_LESS_ARGUMENTS_FOR_FUNCTION = 35++_BAD_ARGUMENTS = 36++_UNKNOWN_ELEMENT_IN_AST = 37++_CANNOT_PARSE_DATE = 38++_TOO_LARGE_SIZE_COMPRESSED = 39++_CHECKSUM_DOESNT_MATCH = 40++_CANNOT_PARSE_DATETIME = 41++_NUMBER_OF_ARGUMENTS_DOESNT_MATCH = 42++_ILLEGAL_TYPE_OF_ARGUMENT = 43++_ILLEGAL_COLUMN = 44++_ILLEGAL_NUMBER_OF_RESULT_COLUMNS = 45++_UNKNOWN_FUNCTION = 46++_UNKNOWN_IDENTIFIER = 47++_NOT_IMPLEMENTED = 48++_LOGICAL_ERROR = 49++_UNKNOWN_TYPE = 50++_EMPTY_LIST_OF_COLUMNS_QUERIED = 51++_COLUMN_QUERIED_MORE_THAN_ONCE = 52++_TYPE_MISMATCH = 53++_STORAGE_DOESNT_ALLOW_PARAMETERS = 54++_STORAGE_REQUIRES_PARAMETER = 55++_UNKNOWN_STORAGE = 56++_TABLE_ALREADY_EXISTS = 57++_TABLE_METADATA_ALREADY_EXISTS = 58++_ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER = 59++_UNKNOWN_TABLE = 60++_ONLY_FILTER_COLUMN_IN_BLOCK = 61++_SYNTAX_ERROR = 62++_UNKNOWN_AGGREGATE_FUNCTION = 63++_CANNOT_READ_AGGREGATE_FUNCTION_FROM_TEXT = 64++_CANNOT_WRITE_AGGREGATE_FUNCTION_AS_TEXT = 65++_NOT_A_COLUMN = 66++_ILLEGAL_KEY_OF_AGGREGATION = 67++_CANNOT_GET_SIZE_OF_FIELD = 68++_ARGUMENT_OUT_OF_BOUND = 69++_CANNOT_CONVERT_TYPE = 70++_CANNOT_WRITE_AFTER_END_OF_BUFFER = 71++_CANNOT_PARSE_NUMBER = 72++_UNKNOWN_FORMAT = 73++_CANNOT_READ_FROM_FILE_DESCRIPTOR = 74++_CANNOT_WRITE_TO_FILE_DESCRIPTOR = 75++_CANNOT_OPEN_FILE = 76++_CANNOT_CLOSE_FILE = 77++_UNKNOWN_TYPE_OF_QUERY = 78++_INCORRECT_FILE_NAME = 79++_INCORRECT_QUERY = 80++_UNKNOWN_DATABASE = 81++_DATABASE_ALREADY_EXISTS = 82++_DIRECTORY_DOESNT_EXIST = 83++_DIRECTORY_ALREADY_EXISTS = 84++_FORMAT_IS_NOT_SUITABLE_FOR_INPUT = 85++_RECEIVED_ERROR_FROM_REMOTE_IO_SERVER = 86++_CANNOT_SEEK_THROUGH_FILE = 87++_CANNOT_TRUNCATE_FILE = 88++_UNKNOWN_COMPRESSION_METHOD = 89++_EMPTY_LIST_OF_COLUMNS_PASSED = 90++_SIZES_OF_MARKS_FILES_ARE_INCONSISTENT = 91++_EMPTY_DATA_PASSED = 92++_UNKNOWN_AGGREGATED_DATA_VARIANT = 93++_CANNOT_MERGE_DIFFERENT_AGGREGATED_DATA_VARIANTS = 94++_CANNOT_READ_FROM_SOCKET = 95++_CANNOT_WRITE_TO_SOCKET = 96++_CANNOT_READ_ALL_DATA_FROM_CHUNKED_INPUT = 97++_CANNOT_WRITE_TO_EMPTY_BLOCK_OUTPUT_STREAM = 98++_UNKNOWN_PACKET_FROM_CLIENT = 99++_UNKNOWN_PACKET_FROM_SERVER = 100++_UNEXPECTED_PACKET_FROM_CLIENT = 101++_UNEXPECTED_PACKET_FROM_SERVER = 102++_RECEIVED_DATA_FOR_WRONG_QUERY_ID = 103++_TOO_SMALL_BUFFER_SIZE = 104++_CANNOT_READ_HISTORY = 105++_CANNOT_APPEND_HISTORY = 106++_FILE_DOESNT_EXIST = 107++_NO_DATA_TO_INSERT = 108++_CANNOT_BLOCK_SIGNAL = 109++_CANNOT_UNBLOCK_SIGNAL = 110++_CANNOT_MANIPULATE_SIGSET = 111++_CANNOT_WAIT_FOR_SIGNAL = 112++_THERE_IS_NO_SESSION = 113++_CANNOT_CLOCK_GETTIME = 114++_UNKNOWN_SETTING = 115++_THERE_IS_NO_DEFAULT_VALUE = 116++_INCORRECT_DATA = 117++_ENGINE_REQUIRED = 119++_CANNOT_INSERT_VALUE_OF_DIFFERENT_SIZE_INTO_TUPLE = 120++_UNKNOWN_SET_DATA_VARIANT = 121++_INCOMPATIBLE_COLUMNS = 122++_UNKNOWN_TYPE_OF_AST_NODE = 123++_INCORRECT_ELEMENT_OF_SET = 124++_INCORRECT_RESULT_OF_SCALAR_SUBQUERY = 125++_CANNOT_GET_RETURN_TYPE = 126++_ILLEGAL_INDEX = 127++_TOO_LARGE_ARRAY_SIZE = 128++_FUNCTION_IS_SPECIAL = 129++_CANNOT_READ_ARRAY_FROM_TEXT = 130++_TOO_LARGE_STRING_SIZE = 131++_CANNOT_CREATE_TABLE_FROM_METADATA = 132++_AGGREGATE_FUNCTION_DOESNT_ALLOW_PARAMETERS = 133++_PARAMETERS_TO_AGGREGATE_FUNCTIONS_MUST_BE_LITERALS = 134++_ZERO_ARRAY_OR_TUPLE_INDEX = 135++_UNKNOWN_ELEMENT_IN_CONFIG = 137++_EXCESSIVE_ELEMENT_IN_CONFIG = 138++_NO_ELEMENTS_IN_CONFIG = 139++_ALL_REQUESTED_COLUMNS_ARE_MISSING = 140++_SAMPLING_NOT_SUPPORTED = 141++_NOT_FOUND_NODE = 142++_FOUND_MORE_THAN_ONE_NODE = 143++_FIRST_DATE_IS_BIGGER_THAN_LAST_DATE = 144++_UNKNOWN_OVERFLOW_MODE = 145++_QUERY_SECTION_DOESNT_MAKE_SENSE = 146++_NOT_FOUND_FUNCTION_ELEMENT_FOR_AGGREGATE = 147++_NOT_FOUND_RELATION_ELEMENT_FOR_CONDITION = 148++_NOT_FOUND_RHS_ELEMENT_FOR_CONDITION = 149++_NO_ATTRIBUTES_LISTED = 150++_INDEX_OF_COLUMN_IN_SORT_CLAUSE_IS_OUT_OF_RANGE = 151++_UNKNOWN_DIRECTION_OF_SORTING = 152++_ILLEGAL_DIVISION = 153++_AGGREGATE_FUNCTION_NOT_APPLICABLE = 154++_UNKNOWN_RELATION = 155++_DICTIONARIES_WAS_NOT_LOADED = 156++_ILLEGAL_OVERFLOW_MODE = 157++_TOO_MANY_ROWS = 158++_TIMEOUT_EXCEEDED = 159++_TOO_SLOW = 160++_TOO_MANY_COLUMNS = 161++_TOO_DEEP_SUBQUERIES = 162++_TOO_DEEP_PIPELINE = 163++_READONLY = 164++_TOO_MANY_TEMPORARY_COLUMNS = 165++_TOO_MANY_TEMPORARY_NON_CONST_COLUMNS = 166++_TOO_DEEP_AST = 167++_TOO_BIG_AST = 168++_BAD_TYPE_OF_FIELD = 169++_BAD_GET = 170++_BLOCKS_HAVE_DIFFERENT_STRUCTURE = 171++_CANNOT_CREATE_DIRECTORY = 172++_CANNOT_ALLOCATE_MEMORY = 173++_CYCLIC_ALIASES = 174++_CHUNK_NOT_FOUND = 176++_DUPLICATE_CHUNK_NAME = 177++_MULTIPLE_ALIASES_FOR_EXPRESSION = 178++_MULTIPLE_EXPRESSIONS_FOR_ALIAS = 179++_THERE_IS_NO_PROFILE = 180++_ILLEGAL_FINAL = 181++_ILLEGAL_PREWHERE = 182++_UNEXPECTED_EXPRESSION = 183++_ILLEGAL_AGGREGATION = 184++_UNSUPPORTED_MYISAM_BLOCK_TYPE = 185++_UNSUPPORTED_COLLATION_LOCALE = 186++_COLLATION_COMPARISON_FAILED = 187++_UNKNOWN_ACTION = 188++_TABLE_MUST_NOT_BE_CREATED_MANUALLY = 189++_SIZES_OF_ARRAYS_DOESNT_MATCH = 190++_SET_SIZE_LIMIT_EXCEEDED = 191++_UNKNOWN_USER = 192++_WRONG_PASSWORD = 193++_REQUIRED_PASSWORD = 194++_IP_ADDRESS_NOT_ALLOWED = 195++_UNKNOWN_ADDRESS_PATTERN_TYPE = 196++_SERVER_REVISION_IS_TOO_OLD = 197++_DNS_ERROR = 198++_UNKNOWN_QUOTA = 199++_QUOTA_DOESNT_ALLOW_KEYS = 200++_QUOTA_EXPIRED = 201++_TOO_MANY_SIMULTANEOUS_QUERIES = 202++_NO_FREE_CONNECTION = 203++_CANNOT_FSYNC = 204++_NESTED_TYPE_TOO_DEEP = 205++_ALIAS_REQUIRED = 206++_AMBIGUOUS_IDENTIFIER = 207++_EMPTY_NESTED_TABLE = 208++_SOCKET_TIMEOUT = 209++_NETWORK_ERROR = 210++_EMPTY_QUERY = 211++_UNKNOWN_LOAD_BALANCING = 212++_UNKNOWN_TOTALS_MODE = 213++_CANNOT_STATVFS = 214++_NOT_AN_AGGREGATE = 215++_QUERY_WITH_SAME_ID_IS_ALREADY_RUNNING = 216++_CLIENT_HAS_CONNECTED_TO_WRONG_PORT = 217++_TABLE_IS_DROPPED = 218++_DATABASE_NOT_EMPTY = 219++_DUPLICATE_INTERSERVER_IO_ENDPOINT = 220++_NO_SUCH_INTERSERVER_IO_ENDPOINT = 221++_ADDING_REPLICA_TO_NON_EMPTY_TABLE = 222++_UNEXPECTED_AST_STRUCTURE = 223++_REPLICA_IS_ALREADY_ACTIVE = 224++_NO_ZOOKEEPER = 225++_NO_FILE_IN_DATA_PART = 226++_UNEXPECTED_FILE_IN_DATA_PART = 227++_BAD_SIZE_OF_FILE_IN_DATA_PART = 228++_QUERY_IS_TOO_LARGE = 229++_NOT_FOUND_EXPECTED_DATA_PART = 230++_TOO_MANY_UNEXPECTED_DATA_PARTS = 231++_NO_SUCH_DATA_PART = 232++_BAD_DATA_PART_NAME = 233++_NO_REPLICA_HAS_PART = 234++_DUPLICATE_DATA_PART = 235++_ABORTED = 236++_NO_REPLICA_NAME_GIVEN = 237++_FORMAT_VERSION_TOO_OLD = 238++_CANNOT_MUNMAP = 239++_CANNOT_MREMAP = 240++_MEMORY_LIMIT_EXCEEDED = 241++_TABLE_IS_READ_ONLY = 242++_NOT_ENOUGH_SPACE = 243++_UNEXPECTED_ZOOKEEPER_ERROR = 244++_CORRUPTED_DATA = 246++_INCORRECT_MARK = 247++_INVALID_PARTITION_VALUE = 248++_NOT_ENOUGH_BLOCK_NUMBERS = 250++_NO_SUCH_REPLICA = 251++_TOO_MANY_PARTS = 252++_REPLICA_IS_ALREADY_EXIST = 253++_NO_ACTIVE_REPLICAS = 254++_TOO_MANY_RETRIES_TO_FETCH_PARTS = 255++_PARTITION_ALREADY_EXISTS = 256++_PARTITION_DOESNT_EXIST = 257++_UNION_ALL_RESULT_STRUCTURES_MISMATCH = 258++_CLIENT_OUTPUT_FORMAT_SPECIFIED = 260++_UNKNOWN_BLOCK_INFO_FIELD = 261++_BAD_COLLATION = 262++_CANNOT_COMPILE_CODE = 263++_INCOMPATIBLE_TYPE_OF_JOIN = 264++_NO_AVAILABLE_REPLICA = 265++_MISMATCH_REPLICAS_DATA_SOURCES = 266++_STORAGE_DOESNT_SUPPORT_PARALLEL_REPLICAS = 267++_CPUID_ERROR = 268++_INFINITE_LOOP = 269++_CANNOT_COMPRESS = 270++_CANNOT_DECOMPRESS = 271++_AIO_SUBMIT_ERROR = 272++_AIO_COMPLETION_ERROR = 273++_AIO_READ_ERROR = 274++_AIO_WRITE_ERROR = 275++_INDEX_NOT_USED = 277++_LEADERSHIP_LOST = 278++_ALL_CONNECTION_TRIES_FAILED = 279++_NO_AVAILABLE_DATA = 280++_DICTIONARY_IS_EMPTY = 281++_INCORRECT_INDEX = 282++_UNKNOWN_DISTRIBUTED_PRODUCT_MODE = 283++_UNKNOWN_GLOBAL_SUBQUERIES_METHOD = 284++_TOO_LESS_LIVE_REPLICAS = 285++_UNSATISFIED_QUORUM_FOR_PREVIOUS_WRITE = 286++_UNKNOWN_FORMAT_VERSION = 287++_DISTRIBUTED_IN_JOIN_SUBQUERY_DENIED = 288++_REPLICA_IS_NOT_IN_QUORUM = 289++_LIMIT_EXCEEDED = 290++_DATABASE_ACCESS_DENIED = 291++_LEADERSHIP_CHANGED = 292++_MONGODB_CANNOT_AUTHENTICATE = 293++_INVALID_BLOCK_EXTRA_INFO = 294++_RECEIVED_EMPTY_DATA = 295++_NO_REMOTE_SHARD_FOUND = 296++_SHARD_HAS_NO_CONNECTIONS = 297++_CANNOT_PIPE = 298++_CANNOT_FORK = 299++_CANNOT_DLSYM = 300++_CANNOT_CREATE_CHILD_PROCESS = 301++_CHILD_WAS_NOT_EXITED_NORMALLY = 302++_CANNOT_SELECT = 303++_CANNOT_WAITPID = 304++_TABLE_WAS_NOT_DROPPED = 305++_TOO_DEEP_RECURSION = 306++_TOO_MANY_BYTES = 307++_UNEXPECTED_NODE_IN_ZOOKEEPER = 308++_FUNCTION_CANNOT_HAVE_PARAMETERS = 309++_INVALID_SHARD_WEIGHT = 317++_INVALID_CONFIG_PARAMETER = 318++_UNKNOWN_STATUS_OF_INSERT = 319++_VALUE_IS_OUT_OF_RANGE_OF_DATA_TYPE = 321++_BARRIER_TIMEOUT = 335++_UNKNOWN_DATABASE_ENGINE = 336++_DDL_GUARD_IS_ACTIVE = 337++_UNFINISHED = 341++_METADATA_MISMATCH = 342++_SUPPORT_IS_DISABLED = 344++_TABLE_DIFFERS_TOO_MUCH = 345++_CANNOT_CONVERT_CHARSET = 346++_CANNOT_LOAD_CONFIG = 347++_CANNOT_INSERT_NULL_IN_ORDINARY_COLUMN = 349++_INCOMPATIBLE_SOURCE_TABLES = 350++_AMBIGUOUS_TABLE_NAME = 351++_AMBIGUOUS_COLUMN_NAME = 352++_INDEX_OF_POSITIONAL_ARGUMENT_IS_OUT_OF_RANGE = 353++_ZLIB_INFLATE_FAILED = 354++_ZLIB_DEFLATE_FAILED = 355++_BAD_LAMBDA = 356++_RESERVED_IDENTIFIER_NAME = 357++_INTO_OUTFILE_NOT_ALLOWED = 358++_TABLE_SIZE_EXCEEDS_MAX_DROP_SIZE_LIMIT = 359++_CANNOT_CREATE_CHARSET_CONVERTER = 360++_SEEK_POSITION_OUT_OF_BOUND = 361++_CURRENT_WRITE_BUFFER_IS_EXHAUSTED = 362++_CANNOT_CREATE_IO_BUFFER = 363++_RECEIVED_ERROR_TOO_MANY_REQUESTS = 364++_OUTPUT_IS_NOT_SORTED = 365++_SIZES_OF_NESTED_COLUMNS_ARE_INCONSISTENT = 366++_TOO_MANY_FETCHES = 367++_BAD_CAST = 368++_ALL_REPLICAS_ARE_STALE = 369++_DATA_TYPE_CANNOT_BE_USED_IN_TABLES = 370++_INCONSISTENT_CLUSTER_DEFINITION = 371++_SESSION_NOT_FOUND = 372++_SESSION_IS_LOCKED = 373++_INVALID_SESSION_TIMEOUT = 374++_CANNOT_DLOPEN = 375++_CANNOT_PARSE_UUID = 376++_ILLEGAL_SYNTAX_FOR_DATA_TYPE = 377++_DATA_TYPE_CANNOT_HAVE_ARGUMENTS = 378++_UNKNOWN_STATUS_OF_DISTRIBUTED_DDL_TASK = 379++_CANNOT_KILL = 380++_HTTP_LENGTH_REQUIRED = 381++_CANNOT_LOAD_CATBOOST_MODEL = 382++_CANNOT_APPLY_CATBOOST_MODEL = 383++_PART_IS_TEMPORARILY_LOCKED = 384++_MULTIPLE_STREAMS_REQUIRED = 385++_NO_COMMON_TYPE = 386++_EXTERNAL_LOADABLE_ALREADY_EXISTS = 387++_CANNOT_ASSIGN_OPTIMIZE = 388++_INSERT_WAS_DEDUPLICATED = 389++_CANNOT_GET_CREATE_TABLE_QUERY = 390++_EXTERNAL_LIBRARY_ERROR = 391++_QUERY_IS_PROHIBITED = 392++_THERE_IS_NO_QUERY = 393++_QUERY_WAS_CANCELLED = 394++_FUNCTION_THROW_IF_VALUE_IS_NON_ZERO = 395++_TOO_MANY_ROWS_OR_BYTES = 396++_QUERY_IS_NOT_SUPPORTED_IN_MATERIALIZED_VIEW = 397++_CANNOT_PARSE_DOMAIN_VALUE_FROM_STRING = 441++_KEEPER_EXCEPTION = 999++_POCO_EXCEPTION = 1000++_STD_EXCEPTION = 1001++_UNKNOWN_EXCEPTION = 1002++_CONDITIONAL_TREE_PARENT_NOT_FOUND = 2001++_ILLEGAL_PROJECTION_MANIPULATOR = 2002
+ src/Database/ClickHouseDriver/HTTP.hs view
@@ -0,0 +1,35 @@+module Database.ClickHouseDriver.HTTP (+ module Database.ClickHouseDriver.HTTP.Types,+ module Database.ClickHouseDriver.HTTP.Client,+ module Database.ClickHouseDriver.HTTP.Connection+) where++import Database.ClickHouseDriver.HTTP.Types+ ( Cmd,+ Format(..),+ Haxl,+ HttpConnection(..),+ HttpParams(..),+ JSONResult )+import Database.ClickHouseDriver.HTTP.Client+ ( defaultHttpClient,+ defaultHttpPool,+ exec,+ getByteString,+ getJSON,+ getJsonM,+ getText,+ getTextM,+ httpClient,+ insertFromFile,+ insertMany,+ insertOneRow,+ ping,+ runQuery,+ setupEnv )+import Database.ClickHouseDriver.HTTP.Connection+ ( HttpConnection(..),+ createHttpPool,+ defaultHttpConnection,+ httpConnect,+ httpConnectDb )
+ src/Database/ClickHouseDriver/HTTP/Client.hs view
@@ -0,0 +1,279 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++{-# LANGUAGE ApplicativeDo #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE CPP #-}++-- | This module provides implementation of user's API++module Database.ClickHouseDriver.HTTP.Client+ ( + -- * Setting+ setupEnv,+ runQuery,+ -- * Query+ getByteString,+ getJSON,+ getText,+ getTextM,+ getJsonM,+ insertOneRow,+ insertMany,+ ping,+ exec,+ insertFromFile,+ -- * Connection+ defaultHttpClient,+ httpClient,+ defaultHttpPool+ )+where++import Database.ClickHouseDriver.Column ( ClickhouseType )+import Database.ClickHouseDriver.Defines as Defines+ ( _DEFAULT_HTTP_PORT, _DEFAULT_HOST )+import Database.ClickHouseDriver.HTTP.Connection+ ( defaultHttpConnection,+ createHttpPool, httpConnectDb)+import Database.ClickHouseDriver.HTTP.Helpers+ ( extract, genURL, toString )+import Database.ClickHouseDriver.HTTP.Types ( Format(..), JSONResult, HttpConnection(..))+import Control.Concurrent.Async ( mapConcurrently )+import Control.Exception ( SomeException, try )+import Control.Monad.State.Lazy ( MonadIO(..) )+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as LBS+import Data.ByteString.Lazy.Builder (char8, lazyByteString,+ toLazyByteString)+import qualified Data.ByteString.Lazy.Char8 as C8+import Data.Hashable ( Hashable(hashWithSalt) )+import qualified Data.Text as T+import Data.Text.Encoding ( decodeUtf8 )+import Data.Typeable ( Typeable )+import Haxl.Core+ ( putFailure,+ putSuccess,+ dataFetch,+ initEnv,+ runHaxl,+ stateEmpty,+ stateSet,+ BlockedFetch(..),+ DataSource(fetch),+ DataSourceName(..),+ PerformFetch(SyncFetch),+ Env(userEnv),+ GenHaxl,+ ShowP(..),+ StateKey(State) )+import Network.HTTP.Client (RequestBody (..),+ httpLbs, method,+ parseRequest,+ requestBody,+ responseBody,+ streamFile)+import Text.Printf ( printf )+import Data.Pool ( withResource, Pool )+import Data.Time.Clock ( NominalDiffTime ) +import Data.Default.Class (def)++{-Implementation in Haxl-}+--+data HttpClient a where+ FetchByteString :: String -> HttpClient BS.ByteString+ FetchJSON :: String -> HttpClient BS.ByteString+ FetchCSV :: String -> HttpClient BS.ByteString+ FetchText :: String -> HttpClient BS.ByteString+ Ping :: HttpClient BS.ByteString++deriving instance Show (HttpClient a)++deriving instance Typeable HttpClient++deriving instance Eq (HttpClient a)++instance ShowP HttpClient where showp = show++instance Hashable (HttpClient a) where+ hashWithSalt salt (FetchByteString cmd) = hashWithSalt salt cmd+ hashWithSalt salt (FetchJSON cmd) = hashWithSalt salt cmd+ hashWithSalt salt (FetchCSV cmd) = hashWithSalt salt cmd+ hashWithSalt salt Ping = hashWithSalt salt ("ok"::BS.ByteString)++instance DataSourceName HttpClient where+ dataSourceName _ = "ClickhouseDataSource"++instance DataSource u HttpClient where+ fetch (settings) _flags _usrenv = SyncFetch $ \blockedFetches -> do+ printf "Fetching %d queries.\n" (length blockedFetches)+ res <- mapConcurrently (fetchData settings) blockedFetches+ case res of+ [()] -> return ()++instance StateKey HttpClient where+ data State HttpClient = SingleHttp HttpConnection+ | HttpPool (Pool HttpConnection)++class HttpEnvironment a where+ toEnv :: a->State HttpClient+ pick :: a-> IO HttpConnection++instance HttpEnvironment HttpConnection where+ toEnv = SingleHttp+ pick = return++instance HttpEnvironment (Pool HttpConnection) where+ toEnv = HttpPool+ pick pool = withResource pool $ return++-- | fetch function+fetchData ::+ State HttpClient -> --Connection configuration+ BlockedFetch HttpClient -> --fetched data+ IO ()+fetchData (settings) fetches = do+ let (queryWithType, var) = case fetches of+ BlockedFetch (FetchJSON query) var' -> (query ++ " FORMAT JSON", var')+ BlockedFetch (FetchCSV query) var' -> (query ++ " FORMAT CSV", var')+ BlockedFetch (FetchByteString query) var' -> (query, var')+ BlockedFetch Ping var' -> ("ping", var')+ e <- Control.Exception.try $ do+ case settings of+ SingleHttp http@(HttpConnection _ mng) -> do+ url <- genURL http queryWithType+ req <- parseRequest url+ ans <- responseBody <$> httpLbs req mng+ return $ LBS.toStrict ans+ HttpPool pool -> + withResource pool $ \conn@(HttpConnection _ mng)->do+ url <- genURL conn queryWithType+ req <- parseRequest url+ ans <- responseBody <$> httpLbs req mng+ return $ LBS.toStrict ans+ either+ (putFailure var)+ (putSuccess var)+ (e :: Either SomeException (BS.ByteString))+ +-- | Fetch data from ClickHouse client in the text format.+getByteString :: String -> GenHaxl u w BS.ByteString+getByteString = dataFetch . FetchByteString++getText :: String -> GenHaxl u w T.Text+getText cmd = fmap decodeUtf8 (getByteString cmd)++-- | Fetch data from ClickHouse client in the JSON format.+getJSON :: String -> GenHaxl u w JSONResult+getJSON cmd = fmap extract (dataFetch $ FetchJSON cmd)++-- | Fetch data from Clickhouse client with commands warped in a Traversable monad.+getTextM :: (Monad m, Traversable m) => m String -> GenHaxl u w (m T.Text)+getTextM = mapM getText++-- | Fetch data from Clickhouse client in the format of JSON +getJsonM :: (Monad m, Traversable m) => m String -> GenHaxl u w (m JSONResult)+getJsonM = mapM getJSON++-- | actual function used by user to perform fetching command+exec :: (HttpEnvironment a)=>String->Env a w->IO (Either C8.ByteString String)+exec cmd' env = do+ let cmd = C8.pack cmd'+ conn@HttpConnection{httpManager=mng} <- pick $ userEnv env+ url <- genURL conn ""+ req <- parseRequest url+ ans <- responseBody <$> httpLbs req{ method = "POST"+ , requestBody = RequestBodyLBS cmd} mng+ if ans /= ""+ then return $ Left ans -- error message+ else return $ Right "Created successfully"++-- | insert one row+insertOneRow :: (HttpEnvironment a)=> String+ -> [ClickhouseType]+ -> Env a w+ -> IO (Either C8.ByteString String)+insertOneRow table_name arr environment = do+ let row = toString arr+ let cmd = C8.pack ("INSERT INTO " ++ table_name ++ " VALUES " ++ row)+ settings@HttpConnection{httpManager=mng} <- pick $ userEnv environment+ url <- genURL settings ""+ req <- parseRequest url+ ans <- responseBody <$> httpLbs req{ method = "POST"+ , requestBody = RequestBodyLBS cmd} mng+ if ans /= ""+ then return $ Left ans -- error messagethe hellenic republic+ else return $ Right "Inserted successfully"++-- | insert one or more rows +insertMany :: (HttpEnvironment a)=> String+ -> [[ClickhouseType]]+ -> Env a w+ -> IO(Either C8.ByteString String)+insertMany table_name rows environment = do+ let rowsString = map (lazyByteString . C8.pack . toString) rows+ comma = char8 ','+ preset = lazyByteString $ C8.pack $ "INSERT INTO " <> table_name <> " VALUES "+ togo = preset <> (foldl1 (\x y-> x <> comma <> y) rowsString)+ settings@HttpConnection{httpManager=mng} <- pick $ userEnv environment+ url <- genURL settings ""+ req <- parseRequest url+ ans <- responseBody <$> httpLbs req{method = "POST"+ , requestBody = RequestBodyLBS $ toLazyByteString togo} mng+ print "inserted successfully"+ if ans /= ""+ then return $ Left ans+ else return $ Right "Successful insertion"++-- | insert data from +insertFromFile :: (HttpEnvironment a)=> String+ ->Format+ ->FilePath+ ->Env a w+ ->IO(Either C8.ByteString String)+insertFromFile table_name format file environment = do+ fileReqBody <- streamFile file+ settings@HttpConnection{httpManager=mng} <- pick $ userEnv environment+ url <- genURL settings ("INSERT INTO " <> table_name + <> case format of+ CSV->" FORMAT CSV"+ JSON->" FORMAT JSON"+ TUPLE->" VALUES")+ req <- parseRequest url+ ans <- responseBody <$> httpLbs req {method = "POST"+ , requestBody = fileReqBody} mng+ if ans /= ""+ then return $ Left ans -- error message+ else return $ Right "Inserted successfully"++ping :: GenHaxl u w BS.ByteString+ping = dataFetch $ Ping++-- | Default environment+setupEnv :: (MonadIO m, HttpEnvironment a)=>a->m (Env a w)+setupEnv csetting = liftIO $ initEnv (stateSet (toEnv csetting) stateEmpty) csetting++defaultHttpClient :: (MonadIO m)=>m (Env HttpConnection w)+defaultHttpClient = liftIO $ defaultHttpConnection >>= setupEnv++defaultHttpPool :: (MonadIO m)=>Int->NominalDiffTime->Int->m(Env (Pool HttpConnection) w)+defaultHttpPool numStripes idleTime maxResources + = liftIO $ createHttpPool def numStripes idleTime maxResources >>= setupEnv++httpClient :: (MonadIO m)=> String->String-> m(Env HttpConnection w)+httpClient user password = liftIO $ httpConnectDb user password Defines._DEFAULT_HTTP_PORT Defines._DEFAULT_HOST Nothing >>= setupEnv++-- | rename runHaxl function.+{-# INLINE runQuery #-}+runQuery :: (MonadIO m)=> Env u w -> GenHaxl u w a -> m a+runQuery env haxl = liftIO $ runHaxl env haxl
+ src/Database/ClickHouseDriver/HTTP/Connection.hs view
@@ -0,0 +1,81 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++-- | Connection pool for HTTP connection. User should import Database.ClickHouseDriver.HTTP instead++module Database.ClickHouseDriver.HTTP.Connection (+ httpConnect,+ httpConnectDb,+ defaultHttpConnection,+ HttpConnection(..),+ createHttpPool+) where+ +import Network.HTTP.Client+ ( defaultManagerSettings, newManager)+import Database.ClickHouseDriver.HTTP.Types+ ( HttpConnection(..),+ HttpParams(HttpParams, httpUsername, httpPort, httpPassword,+ httpHost, httpDatabase) ) +import Data.Default.Class ( Default(..) )+import Data.Pool ( createPool, Pool )+import Data.Time.Clock ( NominalDiffTime )++#define DEFAULT_USERNAME "default"+#define DEFAULT_HOST_NAME "localhost"+#define DEFAULT_PASSWORD ""+--TODO change default password to ""++defaultHttpConnection :: IO (HttpConnection)+defaultHttpConnection = httpConnect DEFAULT_USERNAME DEFAULT_PASSWORD 8123 DEFAULT_HOST_NAME++instance Default HttpParams where+ def = HttpParams{+ httpHost = DEFAULT_HOST_NAME,+ httpPassword = DEFAULT_PASSWORD,+ httpPort = 8123,+ httpUsername = DEFAULT_USERNAME,+ httpDatabase = Nothing+ }++createHttpPool :: HttpParams+ ->Int+ ->NominalDiffTime+ ->Int+ ->IO(Pool HttpConnection)+createHttpPool HttpParams{+ httpHost=host,+ httpPassword = password,+ httpPort = port,+ httpUsername = user,+ httpDatabase = db+ } + = createPool(+ do+ httpConnectDb user password port host db+ )(\_->return ())+++httpConnect :: String->String->Int->String->IO(HttpConnection)+httpConnect user password port host = + httpConnectDb user password port host Nothing++httpConnectDb :: String->String->Int->String->Maybe String->IO(HttpConnection)+httpConnectDb user password port host database = do+ mng <- newManager defaultManagerSettings+ return HttpConnection {+ httpParams = HttpParams {+ httpHost = host,+ httpPassword = password,+ httpPort = port,+ httpUsername = user,+ httpDatabase = database+ },+ httpManager = mng+ }
+ src/Database/ClickHouseDriver/HTTP/Helpers.hs view
@@ -0,0 +1,94 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++{-# LANGUAGE FlexibleInstances #-}++-- | Miscellaneous helper functions. User should not import it. ++module Database.ClickHouseDriver.HTTP.Helpers+ ( extract,+ genURL,+ toString+ )+where++import Database.ClickHouseDriver.Column+ ( ClickhouseType(CKNull, CKTuple, CKArray, CKString, CKInt32) )+import Database.ClickHouseDriver.HTTP.Connection+ ( HttpConnection(HttpConnection, httpParams) )+import Database.ClickHouseDriver.HTTP.Types ( Cmd, JSONResult, HttpParams(..))+import Database.ClickHouseDriver.IO.BufferedWriter ( writeIn )+import Control.Monad.Writer ( WriterT(runWriterT) )+import qualified Data.Aeson as JP+import Data.Attoparsec.ByteString ( IResult(Done, Fail), parse )+import qualified Data.ByteString.Char8 as C8+import qualified Data.HashMap.Strict as HM+import Data.Text (pack)+import Data.Vector (toList)+import qualified Network.URI.Encode as NE+import Data.Maybe ( fromMaybe )++-- | Trim JSON data+extract :: C8.ByteString -> JSONResult+extract val = getData $ parse JP.json val+ where+ getData (Fail e _ _) = Left e+ getData (Done _ (JP.Object x)) = Right $ getData' x+ getData _ = Right []++ getData' = map getObject . maybeArrToList . HM.lookup (pack "data")++ maybeArrToList Nothing = []+ maybeArrToList (Just x) = toList . getArray $ x++ getArray (JP.Array arr) = arr+ getObject (JP.Object x) = x++genURL :: HttpConnection->Cmd->IO String+genURL HttpConnection {+ httpParams = HttpParams{+ httpHost = host,+ httpPassword = pw, + httpPort = port, + httpUsername = usr,+ httpDatabase = db+ }+ }+ cmd = do+ (_,basicUrl) <- runWriterT $ do+ writeIn "http://"+ writeIn usr+ writeIn ":"+ writeIn pw+ writeIn "@"+ writeIn host+ writeIn ":"+ writeIn $ show port + writeIn "/"+ if cmd == "ping" then return () else writeIn "?query="+ writeIn $ dbUrl db+ let res = basicUrl ++ NE.encode cmd+ return res++-- | serialize column type into sql string+toString :: [ClickhouseType]->String+toString ck = "(" ++ toStr ck ++ ")"++toStr :: [ClickhouseType]->String+toStr [] = ""+toStr (x:[]) = toStr' x+toStr (x:xs) = toStr' x ++ "," ++ toStr xs++toStr' :: ClickhouseType->String+toStr' (CKInt32 n) = show n+toStr' (CKString str) = "'" ++ C8.unpack str ++ "'"+toStr' (CKArray arr) = "[" ++ (toStr $ toList arr) ++ "]"+toStr' (CKTuple arr) = "(" ++ (toStr $ toList arr) ++ ")"+toStr' CKNull = "null"+toStr' _ = error "unsupported writing type"++dbUrl :: (Maybe String) -> String+dbUrl = fromMaybe "" . fmap ("?database=" ++)
+ src/Database/ClickHouseDriver/HTTP/Types.hs view
@@ -0,0 +1,46 @@+-- | Definition of types++module Database.ClickHouseDriver.HTTP.Types+ ( JSONResult (..),+ Cmd,+ Haxl,+ Format(..),+ HttpConnection(..),+ HttpParams(..)+ )+where++import Data.Aeson (Value)+import Data.ByteString (ByteString)+import Data.HashMap.Strict (HashMap)+import Data.Text (Text)+import Haxl.Core (GenHaxl) +import Network.HTTP.Client ( Manager ) +++type JSONResult = Either ByteString [HashMap Text Value]++type Cmd = String++type Haxl a = GenHaxl () a++data Format = CSV | JSON | TUPLE+ deriving Eq++data HttpParams + = HttpParams+ {+ httpHost :: !String,+ httpPort :: {-# UNPACK #-} !Int,+ httpUsername :: !String,+ httpPassword :: !String,+ httpDatabase :: Maybe String+ }++data HttpConnection+ = HttpConnection+ { httpParams :: !HttpParams,+ -- ^ basic parameters+ httpManager :: !Manager+ -- ^ http manager + }
+ src/Database/ClickHouseDriver/IO/BufferedReader.hs view
@@ -0,0 +1,215 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE OverloadedStrings #-}++-- | Tools to analyze protocol and deserialize data sent from server. This module is for internal use only.++module Database.ClickHouseDriver.IO.BufferedReader+ ( readBinaryStrWithLength,+ readVarInt',+ readBinaryStr',+ readBinaryStr,+ readVarInt,+ readBinaryInt8,+ readBinaryInt16,+ readBinaryInt64,+ readBinaryInt32,+ readBinaryUInt8,+ readBinaryUInt128,+ readBinaryUInt64,+ readBinaryUInt32,+ readBinaryUInt16,+ Reader,+ Buffer(..),+ createBuffer,+ refill+ )+where++import Control.Monad.State.Lazy ( StateT(StateT) )+import Data.Binary+ ( Word8, Word16, Word32, Word64, Binary, decode )+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as L+import qualified Data.ByteString.Unsafe as UBS+import Data.DoubleWord (Word128 (..))+import Data.Int ( Int8, Int16, Int32, Int64 )+import Data.Maybe ( fromJust, isNothing )+import Foreign.C ( CString )+import qualified Network.Simple.TCP as TCP+import Network.Socket ( Socket )++-- | Buffer is for receiving data from TCP stream. Whenever all bytes are read, it automatically+-- refill from the stream.+data Buffer = Buffer {+ bufSize :: !Int,+ bytesData :: ByteString,+ socket :: Maybe Socket+}++-- | create buffer with size and socket.+createBuffer :: Int->Socket->IO Buffer+createBuffer size sock = do+ receive <- TCP.recv sock size -- receive data+ return Buffer{+ bufSize = size, -- set the size + bytesData = if isNothing receive then "" else fromJust receive, + socket = Just sock+ }++-- | refill buffer from stream+refill :: Buffer->IO Buffer +refill Buffer{socket = Just sock, bufSize = size} = do+ newData' <- TCP.recv sock size+ let newBuffer = case newData' of+ Just newData -> Buffer {+ bufSize = size,+ bytesData = newData,+ socket = Just sock+ }+ Nothing -> error "Network error"+ return newBuffer+refill Buffer{socket=Nothing} = error "empty socket"++type Reader a = StateT Buffer IO a++readBinaryStrWithLength' :: Int+ -- ^ length of string+ -> Buffer+ -- ^ buffer to read+ -> IO (ByteString, Buffer)+ -- ^ (the string read from buffer, buffer after reading)+readBinaryStrWithLength' n buf@Buffer{bufSize=size, bytesData=str, socket=sock} = do+ let l = BS.length str+ let (part, tail) = BS.splitAt n str+ if n > l+ then do+ newbuff <- refill buf+ (unread, altbuff) <- readBinaryStrWithLength' (n - l) newbuff+ return (part <> unread, altbuff)+ else do+ return (part, Buffer size tail sock)++readVarInt' :: Buffer+ -- ^ buffer to be read+ -> IO (Word, Buffer)+ -- ^ (the word read from buffer, the buffer after reading)+readVarInt' buf@Buffer{bufSize=size,bytesData=str, socket=sock} = do+ let l = fromIntegral $ BS.length str+ skip <- UBS.unsafeUseAsCString str (\x -> c_count x l)+ if skip == 0+ then do+ varint' <- UBS.unsafeUseAsCString str (\x->c_read_varint 0 x l)+ new_buf <- refill buf+ let new_str = bytesData new_buf+ varint <- UBS.unsafeUseAsCString new_str (\x->c_read_varint varint' x l)+ skip2 <- UBS.unsafeUseAsCString new_str (\x->c_count x l)+ let tail = BS.drop (fromIntegral skip) new_str+ return (varint, Buffer size tail sock)+ else do+ varint <- UBS.unsafeUseAsCString str (\x -> c_read_varint 0 x l)+ let tail = BS.drop (fromIntegral skip) str+ return (varint, Buffer size tail sock)+-- | read binary string from buffer.+-- It first read the integer(n) in front of the desired string,+-- then it read n bytes to capture the whole string.+readBinaryStr' :: Buffer + -- ^ Buffer to be read+ -> IO (ByteString, Buffer)+ -- ^ (the string read from Buffer, the buffer after reading)+readBinaryStr' str = do+ (len, tail) <- readVarInt' str -- + (head, tail') <- readBinaryStrWithLength' (fromIntegral len) tail+ + return (head, tail')++-- | read n bytes and then transform into a binary type such as bytestring, Int8, UInt16 etc.+readBinaryHelper :: Binary a => Int -> Buffer -> IO (a, Buffer)+readBinaryHelper fmt str = do+ (cut, tail) <- readBinaryStrWithLength' fmt str+ let v = decode ((L.fromStrict. BS.reverse) cut)+ return (v, tail)++class Readable a where+ readIn :: Reader a++instance Readable Word where+ readIn = StateT readVarInt'++instance Readable ByteString where+ readIn = StateT readBinaryStr'++instance Readable Int8 where+ readIn = StateT $ readBinaryHelper 1++instance Readable Int16 where+ readIn = StateT $ readBinaryHelper 2++instance Readable Int32 where+ readIn = StateT $ readBinaryHelper 4++instance Readable Int64 where+ readIn = StateT $ readBinaryHelper 8++instance Readable Word8 where+ readIn = StateT $ readBinaryHelper 1++instance Readable Word16 where+ readIn = StateT $ readBinaryHelper 2++instance Readable Word32 where+ readIn = StateT $ readBinaryHelper 4++instance Readable Word64 where+ readIn = StateT $ readBinaryHelper 8++readVarInt :: Reader Word+readVarInt = readIn++readBinaryStrWithLength :: Int->Reader ByteString+readBinaryStrWithLength n = StateT (readBinaryStrWithLength' $ fromIntegral n)++readBinaryStr :: Reader ByteString+readBinaryStr = readIn++readBinaryInt8 :: Reader Int8+readBinaryInt8 = readIn++readBinaryInt16 :: Reader Int16+readBinaryInt16 = readIn++readBinaryInt32 :: Reader Int32+readBinaryInt32 = readIn++readBinaryInt64 :: Reader Int64+readBinaryInt64 = readIn++readBinaryUInt32 :: Reader Word32+readBinaryUInt32 = readIn++readBinaryUInt8 :: Reader Word8+readBinaryUInt8 = readIn++readBinaryUInt16 :: Reader Word16+readBinaryUInt16 = readIn++readBinaryUInt64 :: Reader Word64+readBinaryUInt64 = readIn++readBinaryUInt128 :: Reader Word128+readBinaryUInt128 = do+ hi <- readBinaryUInt64+ lo <- readBinaryUInt64+ return $ Word128 hi lo++-- | read bytes in the little endian format and transform into integer, see CBits/varuint.c+foreign import ccall unsafe "varuint.h read_varint" c_read_varint :: Word->CString -> Word -> IO Word++-- | Helper of c_read_varint. it counts how many bits it needs to read. +foreign import ccall unsafe "varuint.h count_read" c_count :: CString -> Word -> IO Word
+ src/Database/ClickHouseDriver/IO/BufferedWriter.hs view
@@ -0,0 +1,139 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++-- | Tools to serialize data sent server. This module is for internal use only.++module Database.ClickHouseDriver.IO.BufferedWriter+ ( writeBinaryStr,+ writeBinaryFixedLengthStr,+ writeVarUInt,+ c_write_varint,+ writeBinaryInt8,+ writeBinaryInt16,+ writeBinaryInt32,+ writeBinaryInt64,+ writeBinaryUInt8,+ writeBinaryUInt16,+ writeBinaryUInt32,+ writeBinaryUInt64,+ writeBinaryUInt128,+ writeIn,+ transform,+ Writer,+ MonoidMap,+ )+where++import Control.Monad.IO.Class ( MonadIO(liftIO) )+import Control.Monad.Writer (WriterT, tell)+import qualified Data.Binary as Binary+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import Data.ByteString.Builder+ ( Builder, toLazyByteString, byteString )+import qualified Data.ByteString.Lazy as L+import Data.ByteString.Lazy.Builder (lazyByteString)+import Data.ByteString.Unsafe+ ( unsafePackCString, unsafePackCStringLen )+import Data.DoubleWord (Word128 (..))+import Data.Int ( Int8, Int16, Int32, Int64 )+import Data.Word ( Word8, Word16, Word32, Word64 )+import Foreign.C ( CString )++-- Monoid Homomorphism.+class (Monoid w, Monoid m)=>MonoidMap w m where+ transform :: w->m++instance MonoidMap ByteString L.ByteString where+ transform = L.fromStrict++instance MonoidMap L.ByteString ByteString where+ transform = L.toStrict++instance MonoidMap ByteString Builder where+ transform = byteString++instance MonoidMap L.ByteString Builder where+ transform = lazyByteString++instance MonoidMap Builder L.ByteString where+ transform = toLazyByteString++instance MonoidMap Builder ByteString where+ transform = L.toStrict . toLazyByteString++instance (Monoid w)=>MonoidMap w w where+ transform = id++-- | The writer monad writes bytestring builders and combine them as a monoid. +type Writer w = WriterT w IO ()++writeBinaryFixedLengthStr :: (MonoidMap ByteString w)=>Word->ByteString->Writer w+writeBinaryFixedLengthStr len str = do+ let l = fromIntegral $ BS.length str+ if len /= l+ then error "Error: the length of the given bytestring does not equal to the given length"+ else do+ writeIn str++writeBinaryStr :: (MonoidMap ByteString w)=>ByteString->Writer w+writeBinaryStr str = do+ let l = BS.length str+ writeVarUInt (fromIntegral l)+ writeIn str++writeVarUInt ::(MonoidMap ByteString w)=>Word->Writer w+writeVarUInt n = do+ varuint <- liftIO $ leb128 n+ writeIn varuint + where+ leb128 :: Word->IO ByteString+ leb128 0 = do+ ostr' <- c_write_varint 0+ unsafePackCStringLen (ostr', 1)+ leb128 n = do+ ostr' <- c_write_varint n+ unsafePackCString ostr'++writeBinaryUInt8 :: (MonoidMap L.ByteString w)=>Word8->Writer w+writeBinaryUInt8 = tell . transform . L.reverse . Binary.encode++writeBinaryInt8 :: (MonoidMap L.ByteString w)=>Int8->Writer w+writeBinaryInt8 = tell . transform . L.reverse . Binary.encode++writeBinaryInt16 :: (MonoidMap L.ByteString w)=>Int16->Writer w+writeBinaryInt16 = tell . transform . L.reverse . Binary.encode++writeBinaryInt32 :: (MonoidMap L.ByteString w)=>Int32->Writer w+writeBinaryInt32 = tell . transform . L.reverse . Binary.encode++writeBinaryInt64 :: (MonoidMap L.ByteString w)=>Int64->Writer w+writeBinaryInt64 = tell . transform . L.reverse . Binary.encode++writeBinaryUInt16 :: (MonoidMap L.ByteString w)=>Word16->Writer w+writeBinaryUInt16 = tell . transform . L.reverse . Binary.encode++writeBinaryUInt32 :: (MonoidMap L.ByteString w)=>Word32->Writer w+writeBinaryUInt32 = tell . transform . L.reverse . Binary.encode++writeBinaryUInt64 :: (MonoidMap L.ByteString w)=>Word64->Writer w+writeBinaryUInt64 = tell . transform . L.reverse . Binary.encode++writeBinaryUInt128 :: (MonoidMap L.ByteString w)=>Word128->Writer w+writeBinaryUInt128 (Word128 hi lo) = do+ writeBinaryUInt64 hi+ writeBinaryUInt64 lo++writeIn :: (MonoidMap m w)=>m->Writer w+writeIn = tell . transform++foreign import ccall unsafe "varuint.h write_varint" c_write_varint :: Word -> IO CString
+ src/Database/ClickHouseDriver/Include/bigint.h view
@@ -0,0 +1,13 @@+#ifndef __BIG_INT__+#define __BIG_INT__+#include<stdio.h>++double word128_division(__int64_t hi, __int64_t lo, int scale);++__int64_t low_bits_128(double x,int scale);+__int64_t hi_bits_128(double x,int scale);++__int64_t low_bits_negative_128(double x, int scale);+__int64_t hi_bits_negative_128(double x, int scale);++#endif
+ src/Database/ClickHouseDriver/Include/datetime.h view
@@ -0,0 +1,10 @@+#ifndef __CK_DATE_TIME__+#define __CK_DATE_TIME__++#include<time.h>++char * convert_time(time_t original_time, char * timezone, size_t length);+time_t parse_time(char * time_string, char * timezone, size_t length1, size_t length2);+time_t convert_time_from_int32(time_t time);++#endif
+ src/Database/ClickHouseDriver/Include/varuint.h view
@@ -0,0 +1,12 @@+#ifndef __VARUINT__+#define __VARUINT__++#include<stdio.h>+#include<string.h>+#include<stdlib.h>++const char * write_varint(u_int16_t number);+u_int16_t read_varint(u_int16_t cont ,char * istr, size_t size);+size_t count_read(char * istr, size_t size);++#endif
+ src/Database/ClickHouseDriver/Pool.hs view
@@ -0,0 +1,68 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.+-------------------------------------------------------------------------------+-- This module provides implementation of Connection pool for TCP network+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE NamedFieldPuns #-}+module Database.ClickHouseDriver.Pool +(+ createConnectionPool+) where++import Database.ClickHouseDriver.Connection ( tcpConnect )+import Database.ClickHouseDriver.Defines+ ( _DEFAULT_USERNAME,+ _DEFAULT_HOST_NAME,+ _DEFAULT_PASSWORD,+ _DEFAULT_PORT_NAME,+ _DEFAULT_DATABASE,+ _DEFAULT_COMPRESSION_SETTING )+import Data.Pool ( createPool, Pool )+import Data.Time.Clock ( NominalDiffTime )+import Network.Socket (close)+import Data.Default.Class ( Default(..) )+import Database.ClickHouseDriver.Types+ ( ConnParams(..), TCPConnection(TCPConnection, tcpSocket) )++-- | default connection parameters (settings)+instance Default ConnParams where+ def = ConnParams{+ username' = _DEFAULT_USERNAME+ ,host' = _DEFAULT_HOST_NAME+ ,port' = _DEFAULT_PORT_NAME+ ,password' = _DEFAULT_PASSWORD+ ,compression' = _DEFAULT_COMPRESSION_SETTING+ ,database' = _DEFAULT_DATABASE+ }++-- | Create connection pool+createConnectionPool :: ConnParams+ -- ^ parameters for basic connection. + ->Int+ -- ^ number of stripes+ ->NominalDiffTime+ -- ^ idleTime for each resource when not using.+ ->Int+ -- ^ maximum number of resources.+ ->IO (Pool TCPConnection)+createConnectionPool+ ConnParams+ { username',+ host',+ port',+ password',+ compression',+ database'+ }+ numStripes+ idleTime+ maxResources = createPool (do+ conn <- tcpConnect host' port' username' password' database' compression'+ case conn of+ Left err -> error err+ Right tcp -> return tcp+ ) (\TCPConnection{tcpSocket=sock}->close sock) + numStripes idleTime maxResources
+ src/Database/ClickHouseDriver/QueryProcessingStage.hs view
@@ -0,0 +1,11 @@+module Database.ClickHouseDriver.QueryProcessingStage where+++_FETCH_COLUMNS :: Word+_FETCH_COLUMNS = 0 :: Word++_WITH_MERGEABLE_STATE :: Word+_WITH_MERGEABLE_STATE = 1 :: Word++_COMPLETE :: Word+_COMPLETE = 2 :: Word
+ src/Database/ClickHouseDriver/ServerProtocol.hs view
@@ -0,0 +1,81 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++-- | This module provides constants the++module Database.ClickHouseDriver.ServerProtocol where++import Data.ByteString (ByteString)+import Data.Vector (Vector, fromList, (!?))+++-- Name, version, revision+_HELLO :: Word+_HELLO = 0 :: Word++-- A block of data+_DATA :: Word+_DATA = 1 :: Word++-- The exception during query execution+_EXCEPTION :: Word+_EXCEPTION = 2 :: Word++-- Query execution process rows read, bytes read+_PROGRESS :: Word+_PROGRESS = 3 :: Word++-- ping response+_PONG :: Word+_PONG = 4 :: Word++-- All packets were transimitted+_END_OF_STREAM :: Word+_END_OF_STREAM = 5 :: Word++-- Packet with profiling info+_PROFILE_INFO :: Word+_PROFILE_INFO = 6 :: Word++-- A block with totals+_TOTAL :: Word+_TOTAL = 7 :: Word++-- A block with minimums and maximums+_EXTREMES :: Word+_EXTREMES = 8 :: Word++-- A response to TableStatus request+_TABLES_STATUS_RESPONSE :: Word+_TABLES_STATUS_RESPONSE = 9 :: Word++-- A System logs of the query execution+_LOG :: Word+_LOG = 10 :: Word++-- Columns' description for default values calculation+_TABLE_COLUMNS :: Word+_TABLE_COLUMNS = 11 :: Word++typeStr :: Vector ByteString+typeStr =+ fromList+ [ "Hello",+ "Data",+ "Exception",+ "Progress",+ "Pong",+ "EndOfStream",+ "ProfileInfo",+ "Totals",+ "Extremes",+ "TablesStatusResponse",+ "Log",+ "TableColumns"+ ]++toString :: Int -> ByteString+toString n = + case typeStr !? n of+ Nothing -> "Unknown Packet"+ Just t -> t
+ src/Database/ClickHouseDriver/Types.hs view
@@ -0,0 +1,339 @@+-- Copyright (c) 2014-present, EMQX, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a MIT license,+-- found in the LICENSE file.++{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveGeneric #-}++-- | Implementation of data types for internal use Most users should+-- import "ClickHouseDriver.Core" instead.+--++module Database.ClickHouseDriver.Types+ ( ServerInfo (..),+ TCPConnection (..),+ getServerInfo,+ getClientInfo,+ getClientSetting,+ ClientInfo (..),+ ClientSetting(..),+ Context (..),+ Interface (..),+ QueryKind (..),+ getDefaultClientInfo,+ Packet (..),+ readProgress,+ readBlockStreamProfileInfo,+ QueryInfo(..),+ Progress(..),+ BlockStreamProfileInfo(..),+ storeElasped,+ storeProfile,+ storeProgress,+ defaultProfile,+ defaultProgress,+ defaultQueryInfo,+ ClickhouseType(..),+ BlockInfo(..),+ Block(..),+ CKResult(..),+ writeBlockInfo,+ ConnParams(..),+ setClientInfo,+ setClientSetting,+ setServerInfo+ )+where++import qualified Database.ClickHouseDriver.Defines as Defines+import Database.ClickHouseDriver.IO.BufferedReader+ ( Reader, readVarInt, readBinaryUInt8 )+import Database.ClickHouseDriver.IO.BufferedWriter+ ( Writer, writeVarUInt, writeBinaryUInt8, writeBinaryInt32)+import Data.ByteString (ByteString)+import Data.ByteString.Builder ( Builder )+import Data.Default.Class ( Default(..) )+import Data.Int ( Int8, Int16, Int32, Int64 )+import Data.Vector (Vector)+import Data.Word ( Word8, Word16, Word32, Word64 )+import GHC.Generics ( Generic )+import Network.Socket (SockAddr, Socket)++-----------------------------------------------------------++-----------------------------------------------------------+data BlockInfo = Info+ { is_overflows :: !Bool,+ bucket_num :: {-# UNPACK #-} !Int32+ } + deriving Show++writeBlockInfo :: BlockInfo->Writer Builder+writeBlockInfo Info{is_overflows, bucket_num} = do+ writeVarUInt 1+ writeBinaryUInt8 (if is_overflows then 1 else 0)+ writeVarUInt 2+ writeBinaryInt32 bucket_num+ writeVarUInt 0++data Block = ColumnOrientedBlock+ { columns_with_type :: Vector (ByteString, ByteString),+ cdata :: Vector (Vector ClickhouseType),+ info :: BlockInfo+ }+ deriving Show+------------------------------------------------------------+data ClickhouseType+ = CKInt8 Int8+ | CKInt16 Int16+ | CKInt32 Int32+ | CKInt64 Int64+ | CKInt128 Int64 Int64+ | CKUInt8 Word8+ | CKUInt16 Word16+ | CKUInt32 Word32+ | CKUInt64 Word64+ | CKUInt128 Word64 Word64+ | CKString ByteString+ | CKTuple (Vector ClickhouseType)+ | CKArray (Vector ClickhouseType)+ | CKDecimal Float+ | CKDecimal32 Float+ | CKDecimal64 Double+ | CKDecimal128 Double+ | CKIPv4 (Word8, Word8, Word8, Word8)+ | CKIPv6 (Word16, Word16, Word16, Word16,+ Word16, Word16, Word16, Word16)+ | CKDate {+ year :: !Integer,+ month :: !Int,+ day :: !Int + }+ | CKNull+ deriving (Show, Eq)++----------------------------------------------------------+data ServerInfo = ServerInfo+ { name :: {-# UNPACK #-} !ByteString,+ version_major :: {-# UNPACK #-} !Word,+ version_minor :: {-# UNPACK #-} !Word,+ version_patch :: {-# UNPACK #-} !Word,+ revision :: !Word,+ timezone :: Maybe ByteString,+ display_name :: {-# UNPACK #-} !ByteString+ }+ deriving (Show)++setServerInfo :: Maybe ServerInfo->TCPConnection->TCPConnection+setServerInfo server_info tcp@TCPConnection{context=ctx} + = tcp{context=ctx{server_info=server_info}}+---------------------------------------------------------+data TCPConnection = TCPConnection+ { tcpHost :: {-# UNPACK #-} !ByteString,+ -- ^ host name, default = "localhost" + tcpPort :: {-# UNPACK #-} !ByteString,+ -- ^ port number, default = "8123"+ tcpUsername :: {-# UNPACK #-} !ByteString,+ -- ^ username, default = "default"+ tcpPassword :: {-# UNPACK #-} !ByteString,+ -- ^ password, dafault = ""+ tcpSocket :: !Socket,+ -- ^ socket for communication+ tcpSockAdrr :: !SockAddr,+ context :: !Context,+ -- ^ server and client informations+ tcpCompression :: {-# UNPACK #-} !Word+ -- ^ should the data be compressed or not. Not applied yet. + }+ deriving (Show)++getServerInfo :: TCPConnection->Maybe ServerInfo+getServerInfo TCPConnection{context=Context{server_info=server_info}} = server_info++getClientInfo :: TCPConnection->Maybe ClientInfo+getClientInfo TCPConnection{context=Context{client_info=client_info}} = client_info++getClientSetting :: TCPConnection->Maybe ClientSetting+getClientSetting TCPConnection{context=Context{client_setting=client_setting}} = client_setting+------------------------------------------------------------------+data ClientInfo = ClientInfo+ { client_name :: {-# UNPACK #-} !ByteString,+ interface :: Interface,+ client_version_major :: {-# UNPACK #-} !Word,+ client_version_minor :: {-# UNPACK #-} !Word,+ client_version_patch :: {-# UNPACK #-} !Word,+ client_revision :: {-# UNPACK #-} !Word,+ initial_user :: {-# UNPACK #-} !ByteString,+ initial_query_id :: {-# UNPACK #-} !ByteString,+ initial_address :: {-# UNPACK #-} !ByteString,+ quota_key :: {-# UNPACK #-} !ByteString,+ query_kind :: QueryKind+ }+ deriving (Show)++getDefaultClientInfo :: ByteString -> ClientInfo+getDefaultClientInfo name =+ ClientInfo+ { client_name = name,+ interface = TCP,+ client_version_major = Defines._CLIENT_VERSION_MAJOR,+ client_version_minor = Defines._CLIENT_VERSION_MINOR,+ client_version_patch = Defines._CLIENT_VERSION_PATCH,+ client_revision = Defines._CLIENT_REVISION,+ initial_user = "",+ initial_query_id = "",+ initial_address = "0.0.0.0:0",+ quota_key = "",+ query_kind = INITIAL_QUERY+ }++setClientInfo :: Maybe ClientInfo -> TCPConnection -> TCPConnection+setClientInfo client_info tcp@TCPConnection{context=ctx}+ = tcp{context=ctx{client_info=client_info}}+-------------------------------------------------------------------+data ClientSetting + = ClientSetting {+ insert_block_size ::{-# UNPACK #-} !Word,+ strings_as_bytes :: !Bool,+ strings_encoding ::{-# UNPACK #-} !ByteString+ }+ deriving Show++setClientSetting :: Maybe ClientSetting->TCPConnection->TCPConnection+setClientSetting client_setting tcp@TCPConnection{context=ctx} + = tcp{context=ctx{client_setting=client_setting}}++-------------------------------------------------------------------+data Interface = TCP | HTTP+ deriving (Show, Eq)++data QueryKind = NO_QUERY | INITIAL_QUERY | SECOND_QUERY+ deriving (Show, Eq)++data Context = Context+ { client_info :: Maybe ClientInfo,+ server_info :: Maybe ServerInfo,+ client_setting :: Maybe ClientSetting+ }+ deriving Show++data Packet+ = Block {queryData :: !Block}+ | Progress {prog :: !Progress}+ | StreamProfileInfo {profile :: !BlockStreamProfileInfo}+ | MultiString !(ByteString, ByteString)+ | ErrorMessage !String+ | Hello+ | EndOfStream+ deriving (Show)+------------------------------------------------------------+data Progress = Prog+ { rows :: {-# UNPACK #-} !Word,+ bytes :: {-# UNPACK #-} !Word,+ total_rows :: {-# UNPACK #-} !Word,+ written_rows :: {-# UNPACK #-} !Word,+ written_bytes :: {-# UNPACK #-} !Word+ }+ deriving (Show)++instance Default Progress where+ def = defaultProgress++increment :: Progress -> Progress -> Progress+increment (Prog a b c d e) (Prog a' b' c' d' e') =+ Prog (a + a') (b + b') (c + c') (d + d') (e + e')++readProgress :: Word -> Reader Progress+readProgress server_revision = do+ rows <- readVarInt+ bytes <- readVarInt++ let revision = server_revision+ total_rows <-+ if revision >= Defines._DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS+ then readVarInt+ else return 0+ if revision >= Defines._DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO+ then do+ written_rows <- readVarInt+ written_bytes <- readVarInt+ return $ Prog rows bytes total_rows written_rows written_bytes+ else do+ return $ Prog rows bytes total_rows 0 0++defaultProgress :: Progress+defaultProgress = Prog 0 0 0 0 0+----------------------------------------------------------------------+data BlockStreamProfileInfo = ProfileInfo+ { number_rows :: {-# UNPACK #-} !Word,+ blocks :: {-# UNPACK #-} !Word,+ number_bytes :: {-# UNPACK #-} !Word,+ applied_limit :: !Bool,+ rows_before_limit :: {-# UNPACK #-} !Word,+ calculated_rows_before_limit :: !Bool+ }+ deriving Show++instance Default BlockStreamProfileInfo where+ def = defaultProfile++defaultProfile :: BlockStreamProfileInfo+defaultProfile = ProfileInfo 0 0 0 False 0 False++readBlockStreamProfileInfo :: Reader BlockStreamProfileInfo+readBlockStreamProfileInfo = do+ rows <- readVarInt+ blocks <- readVarInt+ bytes <- readVarInt+ applied_limit <- (>= 0) <$> readBinaryUInt8+ rows_before_limit <- readVarInt+ calculated_rows_before_limit <- (>= 0) <$> readBinaryUInt8+ return $ ProfileInfo rows blocks bytes applied_limit rows_before_limit calculated_rows_before_limit+-----------------------------------------------------------------------+data QueryInfo = QueryInfo + { profile_info :: !BlockStreamProfileInfo,+ progress :: !Progress,+ elapsed :: {-# UNPACK #-} !Word+ } deriving Show++instance Default QueryInfo where+ def = defaultQueryInfo++storeProfile :: QueryInfo->BlockStreamProfileInfo->QueryInfo+storeProfile (QueryInfo _ progress elapsed) new_profile + = QueryInfo new_profile progress elapsed++storeProgress :: QueryInfo->Progress->QueryInfo+storeProgress (QueryInfo profile progress elapsed) new_progress + = QueryInfo profile (increment progress new_progress) elapsed++storeElasped :: QueryInfo->Word->QueryInfo+storeElasped (QueryInfo profile progress _)+ = QueryInfo profile progress ++defaultQueryInfo :: QueryInfo+defaultQueryInfo = + QueryInfo+ { progress = defaultProgress,+ profile_info = defaultProfile,+ elapsed = 0+ }+-------------------------------------------------------------------------+data CKResult = CKResult+ { query_result :: Vector (Vector ClickhouseType),+ query_info :: !QueryInfo+ }+ deriving Show+-------------------------------------------------------------------------+data ConnParams = ConnParams{+ username' :: !ByteString,+ host' :: !ByteString,+ port' :: !ByteString,+ password' :: !ByteString,+ compression' :: !Bool,+ database' :: !ByteString+ }+ deriving (Show, Generic)
+ test/Spec.hs view
@@ -0,0 +1,11 @@+module Main where++import Test.ColumnSpec+import Test.HTTPSpec+import Test.IO++main :: IO()+main = do+ runTests+ httpSpec+ columnSpec
+ test/Test/ColumnSpec.hs view
@@ -0,0 +1,137 @@+{-# LANGUAGE OverloadedStrings #-}+module Test.ColumnSpec (columnSpec) where++import Database.ClickHouseDriver+ ( ClickhouseType(CKNull, CKTuple, CKArray, CKInt8, CKInt16,+ CKString),+ defaultClient,+ insertMany,+ query )+import Test.Hspec+ ( hspec, describe, it, parallel, runIO, shouldBe, Spec )+import Data.Vector (fromList)+++columnSpec :: IO()+columnSpec = hspec $ parallel $ do+ stringAndIntSpec+ arraySpec+ tupleAndEnumSpec+ lowCardinalitySpec+ --ipAndDateSpec+ --aggregateFunctionSpec+ uuidSpec+ --nullableSpec++stringAndIntSpec :: Spec+stringAndIntSpec = describe "test string and int columns" $ do+ conn <- runIO $ defaultClient+ runIO $ query conn ("CREATE TABLE IF NOT EXISTS str_and_int_suite " ++ + "(`str` String, `int` Int16, `fix` FixedString(3))" ++ "ENGINE = Memory") + runIO $ Database.ClickHouseDriver.insertMany conn "INSERT INTO str_and_int_suite VALUES"+ [+ [CKString "teststr1", CKInt16 323, CKString "abc"],+ [CKString "teststr2", CKInt16 456 ,CKString "axy"],+ [CKString "teststr2", CKInt16 220, CKString "ffa"]+ ]+ q <- runIO $ query conn "SELECT * FROM str_and_int_suite" + it "returns query result in standard format" $ do+ show q `shouldBe` "[[CKString \"teststr1\", CKInt16 323, CKString \"abc\"]," +++ "[CKString \"teststr2\", CKInt16 456 ,CKString \"axy\"],[CKString \"teststr2\"," +++ "CKInt16 220, CKString \"ffa\"]]"++arraySpec :: Spec+arraySpec = describe "test array" $ do+ conn <- runIO $ defaultClient+ runIO $ query conn ("CREATE TABLE IF NOT EXISTS array_suite " ++ + "(`id` Int8, `arr` Array(String))" ++ "ENGINE = Memory") + runIO $ Database.ClickHouseDriver.insertMany conn "INSERT INTO array_suite VALUES"+ [+ [CKInt8 1, CKArray $ fromList [CKString "Clickhouse", CKString "Test1"]],+ [CKInt8 2, CKArray $ fromList [CKString "Clickhouse", CKString "Test2"]],+ [CKInt8 3, CKArray $ fromList [CKString "Clickhouse", CKString "Test3"]]+ ]+ q <- runIO $ query conn "SELECT * FROM array_suite" + it "returns query result in standard format" $ do+ show q `shouldBe` (show $ ([+ [CKInt8 1, CKArray $ fromList [CKString "Clickhouse", CKString "Test1"]],+ [CKInt8 2, CKArray $ fromList [CKString "Clickhouse", CKString "Test2"]],+ [CKInt8 3, CKArray $ fromList [CKString "Clickhouse", CKString "Test3"]]+ ]))++tupleAndEnumSpec :: Spec+tupleAndEnumSpec = describe "tuple and enum test" $ do+ conn <- runIO $ defaultClient+ runIO $ query conn ("CREATE TABLE IF NOT EXISTS tuple_enum_suite " ++ + "(`id` Int8, `tup` Tuple(Int8, String), `enum` Enum('hello' = 1, 'world' = 2))" ++ "ENGINE = Memory") + runIO $ Database.ClickHouseDriver.insertMany conn "INSERT INTO tuple_enum_suite VALUES"+ [+ [CKInt8 1, CKTuple $ fromList [CKInt8 11, CKString "Test1"], CKString "hello"],+ [CKInt8 2, CKTuple $ fromList [CKInt8 12, CKString "Test2"], CKString "world"],+ [CKInt8 3, CKTuple $ fromList [CKInt8 13, CKString "Test3"], CKString "hello"]+ ]+ q <- runIO $ query conn "SELECT * FROM tuple_enum_suite" + it "returns query result in standard format" $ do+ show q `shouldBe` (show $ ([+ [CKInt8 1, CKTuple $ fromList [CKInt8 11, CKString "Test1"], CKString "hello"],+ [CKInt8 2, CKTuple $ fromList [CKInt8 12, CKString "Test2"], CKString "world"],+ [CKInt8 3, CKTuple $ fromList [CKInt8 13, CKString "Test3"], CKString "hello"]+ ]))+ ++lowCardinalitySpec :: Spec+lowCardinalitySpec = describe "lowcardinality type test" $ do+ conn <- runIO $ defaultClient+ runIO $ query conn ("CREATE TABLE IF NOT EXISTS lowcardinality_suite " ++ + "(`id` Int8, `lowstr` LowCardinality(String))" ++ "ENGINE = Memory")+ runIO $ Database.ClickHouseDriver.insertMany conn "INSERT INTO lowcardinality_suite VALUES"+ [+ [CKInt8 1, CKString "Clickhouse", CKInt8 123],+ [CKInt8 2, CKString "driver", CKInt8 123],+ [CKInt8 3, CKString "Clickhouse", CKInt8 120]+ ]+ q <- runIO $ query conn "SELECT * FROM lowcardinality_suite" + it "returns query result in standard format" $ do+ show q `shouldBe` "[[CKInt8 1, CKString \"Clickhouse\"]," +++ "[CKInt8 2, CKString \"driver\",]," +++ "[CKInt8 3, CKString \"Clickhouse\"]]"++ipAndDateSpec :: Spec+ipAndDateSpec = undefined++aggregateFunctionSpec :: Spec+aggregateFunctionSpec = undefined++uuidSpec :: Spec+uuidSpec = describe "uuid test" $ do+ conn <- runIO $ defaultClient+ runIO $ query conn ("CREATE TABLE IF NOT EXISTS uuid_suite " ++ + "(`id` Int8, `uuid` UUID)" ++ "ENGINE = Memory") + runIO $ Database.ClickHouseDriver.insertMany conn "INSERT INTO uuid_suite VALUES"+ [+ [CKInt8 1, CKString "123e4567-e89b-12d3-a456-426614174000"],+ [CKInt8 2, CKString "123e4567-e89b-12d3-a456-426614174000"],+ [CKInt8 3, CKString "123e4567-e89b-12d3-a456-426614174000"]+ ]+ q <- runIO $ query conn "SELECT * FROM uuid_suite" + it "returns query result in standard format" $ do+ show q `shouldBe` "[[CKInt8 1,CKString \"123e4567-e89b-12d3-a456-426614174000\"]," ++ + "[CKInt8 2,CKString \"123e4567-e89b-12d3-a456-426614174000\"]," +++ "[CKInt8 3,CKString \"123e4567-e89b-12d3-a456-426614174000\"]]"++nullableSpec :: Spec+nullableSpec = describe "nullable type test" $ do+ conn <- runIO $ defaultClient+ runIO $ query conn ("CREATE TABLE IF NOT EXISTS nullable_suite " ++ + "(`id` Int8, `nullableStr` Nullable(String), `nullableInt` Nullable(Int16), `nullableArr` Array(Nullable(Int16)))" ++ "ENGINE = Memory") + runIO $ Database.ClickHouseDriver.insertMany conn "INSERT INTO nullable_suie VALUES"+ [+ [CKInt8 1, CKString "string", CKNull, CKArray $ fromList [CKNull, CKNull, CKInt16 155]],+ [CKInt8 2, CKNull, CKInt16 1100, CKArray $ fromList [CKNull,CKInt16 110, CKNull]],+ [CKInt8 3, CKString "string", CKNull, CKArray $ fromList [CKInt16 220, CKInt16 110, CKNull]]+ ]+ q <- runIO $ query conn "SELECT * FROM nullable_suite" + it "returns query result in standard format" $ do+ show q `shouldBe` "[[CKInt8 1, CKString \"17ddc5d-e556-4d27-95dd-a34d84e46a50\"]," +++ "[CKInt8 2, CKString \"17ddc5d-0000-4d27-95dd-a34d84e46a50\"]," +++ "[CKInt8 3, CKString \"17ddc5d-e556-4d27-0000-a34d84e46a50\"]]"
+ test/Test/HTTPSpec.hs view
@@ -0,0 +1,70 @@+{-# LANGUAGE OverloadedStrings #-}++module Test.HTTPSpec (httpSpec) where++import Database.ClickHouseDriver+import Database.ClickHouseDriver.HTTP+import Data.ByteString as B+import Data.ByteString.Char8 as C8+import qualified Data.HashMap.Strict as HM+import Haxl.Core+import Test.HUnit+import Test.Hspec+import Data.Either+++httpSpec :: IO()+httpSpec = hspec $ parallel $ do+ query1+ query2+ query3+ queryTCP+ lowCardinalityTest++query1 :: Spec+query1 = describe "show databases" $ do+ deSetting <- runIO $ defaultHttpConnection+ env <- runIO $ setupEnv deSetting+ res <- runIO $ runQuery env (getByteString "SHOW DATABASES")+ it "returns query result in text format" $ do+ res `shouldBe` C8.pack "_temporary_and_external_tables\ndefault\nsystem\n"++query2 :: Spec+query2 = describe "select 1" $ do+ deSetting <- runIO $ defaultHttpConnection+ env <- runIO $ setupEnv deSetting+ res <- runIO $ runQuery env (getByteString "SELECT 1")+ it "returns query result in text format" $ do+ res `shouldBe` C8.pack "1\n"++query3 :: Spec+query3 = describe "format in JSON" $ do+ deSetting <- runIO $ defaultHttpConnection+ env <- runIO $ setupEnv deSetting+ res <- runIO $ runQuery env (getJSON "SELECT * FROM default.test_table")+ let check = case res of+ Right (x : xs) -> C8.pack $ show (HM.lookup "id" x)+ _ -> C8.pack "error"+ it "returns query result in JSON" $ do+ check `shouldBe` C8.pack "Just (String \"0000000011\")"++queryTCP :: Spec+queryTCP = describe "select 1" $ do+ conn <- runIO $ defaultClient+ res <- runIO $ query conn "SELECT 1" + it "returns result in ClickhouseType" $ do+ (show res) `shouldBe` ("Right [[CKUInt8 1]]")++lowCardinalityTest :: Spec+lowCardinalityTest = describe "lowCardinality" $ do+ conn <- runIO $ defaultClient+ res <- runIO $ query conn "SELECT * FROM crd3" + it "returns result in ClickhouseType" $ do+ (show res) `shouldBe` ("Right " ++ show [[CKString "abc",CKString "myString",CKNull],[CKString"xyz",CKString "Noctis",CKString "Ross"],[CKString "123",CKString "Alice",CKNull],[CKString "456",CKString "Bob",CKString "Walter"]])++comprehensiveTest :: Spec+comprehensiveTest = describe "array string number etc." $ do+ conn <- runIO $ defaultClient+ res <- runIO $ query conn "SELECT * FROM big" + it "returns result in ClickhouseType" $ do+ (show res) `shouldBe` (show "")
+ test/Test/IO.hs view
@@ -0,0 +1,83 @@+module Test.IO where++import Database.ClickHouseDriver.IO.BufferedWriter+import Database.ClickHouseDriver.IO.BufferedReader+import Test.QuickCheck+import Test.QuickCheck.Monadic+import qualified Data.ByteString as BS+import Control.Monad.Writer hiding (Writer)+import Control.Monad.State+import Data.ByteString (ByteString)+import Data.Int+import qualified Data.ByteString.Char8 as C8++prop_BinaryStr :: String->Property+prop_BinaryStr xs = monadicIO $ do+ let wtr = writeBinaryStr :: ByteString->Writer ByteString+ let packed = C8.pack xs+ a <- run (execWriterT $ wtr packed)+ let buf = Buffer {+ bufSize = BS.length a,+ socket = Nothing,+ bytesData = a+ }+ (res,_) <- run (runStateT readBinaryStr buf)+ assert (res == packed)++prop_Strs :: [String]->Property+prop_Strs xs = monadicIO $ do+ let wtr = writeBinaryStr :: ByteString->Writer ByteString+ let packed = fmap C8.pack xs + a <- run (execWriterT $ (mapM_ wtr packed))+ let buf = Buffer {+ bufSize = BS.length a,+ socket = Nothing,+ bytesData = a+ }+ (res, _) <- run (runStateT (mapM (\x->readBinaryStr) xs) buf)+ assert (res == packed)++prop_int8 :: [Int8]->Property+prop_int8 xs = monadicIO $ do+ let wtr = writeBinaryInt8 :: Int8->Writer ByteString+ a <- run (execWriterT $ (mapM_ wtr xs))+ let buf = Buffer {+ bufSize = BS.length a,+ socket = Nothing,+ bytesData = a+ }+ (res, _) <- run (runStateT (mapM (\x->readBinaryInt8) xs) buf)+ assert (res == xs)++prop_int32 :: [Int32]->Property+prop_int32 xs = monadicIO $ do+ let wtr = writeBinaryInt32 :: Int32->Writer ByteString+ a <- run (execWriterT $ (mapM_ wtr xs))+ let buf = Buffer {+ bufSize = BS.length a,+ socket = Nothing,+ bytesData = a+ }+ (res, _) <- run (runStateT (mapM (\x->readBinaryInt32) xs) buf)+ assert (res == xs)++prop_fix_str :: [String]->Property+prop_fix_str xs = monadicIO $ do+ let wtr = writeBinaryFixedLengthStr :: Word->ByteString->Writer ByteString+ let packed = fmap C8.pack xs+ a <- run $ execWriterT $ (mapM_ (\str->wtr (fromIntegral $ BS.length str) str ) packed)+ let buf = Buffer {+ bufSize = BS.length a,+ socket = Nothing,+ bytesData = a+ }+ (res,_) <- run $ runStateT (mapM (\str->readBinaryStrWithLength (BS.length str)) packed) buf+ assert (res == packed)++runTests = do+ print "test"+ quickCheck prop_BinaryStr+ quickCheck prop_int8+ quickCheck prop_int32+ quickCheck prop_Strs+ quickCheck prop_fix_str