diff --git a/avro.cabal b/avro.cabal --- a/avro.cabal +++ b/avro.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 307ace222c16e9660fc8fd8f1d3b126a4254daea39bb616b93782a86b81d73b0 +-- hash: 83c35a316bced7d77c45fea0d0d8b228cdff0c26f57a07eedb698798addc4697 name: avro -version: 0.3.4.3 +version: 0.3.5.0 synopsis: Avro serialization support for Haskell description: Avro serialization and deserialization support for Haskell category: Data @@ -23,7 +23,6 @@ test/data/deconflict/writer.avsc test/data/enums-object.json test/data/enums.avsc - test/data/internal-bindings.avsc test/data/karma.avsc test/data/logical.avsc test/data/maybe.avsc @@ -122,6 +121,7 @@ Avro.Codec.TextSpec Avro.Codec.ZigZagSpec Avro.Decode.Lazy.ContainerSpec + Avro.Decode.Lazy.RawBlocksSpec Avro.Decode.Lazy.ValuesSpec Avro.Deconflict.Reader Avro.Deconflict.Writer @@ -137,7 +137,6 @@ Avro.THSimpleSpec Avro.THUnionSpec Avro.ToAvroSpec - Avro.SchemaSpec DecodeContainer Example1 Paths_avro diff --git a/src/Data/Avro.hs b/src/Data/Avro.hs --- a/src/Data/Avro.hs +++ b/src/Data/Avro.hs @@ -72,6 +72,7 @@ ( FromAvro(..) , ToAvro(..) , HasAvroSchema(..) + , Schema , Avro , (.:) , (.=), record, fixed diff --git a/src/Data/Avro/Decode/Lazy.hs b/src/Data/Avro/Decode/Lazy.hs --- a/src/Data/Avro/Decode/Lazy.hs +++ b/src/Data/Avro/Decode/Lazy.hs @@ -13,6 +13,9 @@ , decodeContainerWithSchema , decodeContainerWithSchema' + -- * Bypass decoding + , decodeRawBlocks + -- * Lower level interface , getContainerValues , getContainerValuesWith @@ -30,11 +33,11 @@ import Data.Bits import Data.ByteString (ByteString) import qualified Data.ByteString.Lazy as BL -import qualified Data.ByteString.Lazy.Char8 as BC +import qualified Data.ByteString.Lazy.Char8 as BL import Data.Either (isRight) import qualified Data.HashMap.Strict as HashMap import Data.Int -import Data.List (foldl') +import Data.List (foldl', unfoldr) import qualified Data.List.NonEmpty as NE import qualified Data.Map as Map import Data.Maybe @@ -58,6 +61,8 @@ import Data.Avro.Decode.Lazy.Deconflict as C import Data.Avro.FromAvro +import Debug.Trace + -- | Decodes the container as a lazy list of values of the requested type. -- -- The schema for the requested type will be de-conflicted with the schema @@ -132,21 +137,54 @@ -- -- The "outer" error represents the error in opening the container itself -- (including problems like reading schemas embedded into the container.) -getContainerValues :: BC.ByteString -> Either String (Schema, [[T.LazyValue Type]]) +getContainerValues :: BL.ByteString -> Either String (Schema, [[T.LazyValue Type]]) getContainerValues = getContainerValuesWith getAvroOf {-# INLINABLE getContainerValues #-} -getContainerValuesWith :: (Schema -> BL.ByteString -> (BL.ByteString, T.LazyValue Type)) - -> BL.ByteString - -> Either String (Schema, [[T.LazyValue Type]]) -getContainerValuesWith schemaToGet bs = +-- | Reads the container as a list of blocks without decoding them into actual values. +-- +-- This can be useful for streaming / splitting / merging Avro containers without +-- paying the cost for Avro encoding/decoding. +-- +-- Each block is returned as a raw 'ByteString' annotated with the number of Avro values +-- that are contained in this block. +-- +-- The "outer" error represents the error in opening the container itself +-- (including problems like reading schemas embedded into the container.) +decodeRawBlocks :: BL.ByteString -> Either String (Schema, [Either String (Int, BL.ByteString)]) +decodeRawBlocks bs = case runGetOrFail getAvro bs of - Left (bs', _, err) -> - Left err + Left (bs', _, err) -> Left err Right (bs', _, ContainerHeader {..}) -> - Right (containedSchema, snd $ getBlocks (schemaToGet containedSchema) syncBytes bs' decompress) + let blocks = allBlocks syncBytes decompress bs' + in Right (containedSchema, blocks) where - getRawBlock :: (BL.ByteString -> Get BL.ByteString) -> Get (Int, BC.ByteString) + allBlocks sync decompress bytes = + flip unfoldr (Just bytes) $ \acc -> case acc of + Just rest -> next sync decompress rest + Nothing -> Nothing + + next syncBytes decompress bytes = + case getNextBlock syncBytes decompress bytes of + Right (Just (numObj, block, rest)) -> Just (Right (numObj, block), Just rest) + Right Nothing -> Nothing + Left err -> Just (Left err, Nothing) + +getNextBlock :: BL.ByteString + -> (BL.ByteString -> Get BL.ByteString) + -> BL.ByteString + -> Either String (Maybe (Int, BL.ByteString, BL.ByteString)) +getNextBlock sync decompress bs = + if BL.null bs + then Right Nothing + else case runGetOrFail (getRawBlock decompress) bs of + Left (bs', _, err) -> Left err + Right (bs', _, (nrObj, bytes)) -> + case checkMarker sync bs' of + Left err -> Left err + Right rest -> Right $ Just (nrObj, bytes, rest) + where + getRawBlock :: (BL.ByteString -> Get BL.ByteString) -> Get (Int, BL.ByteString) getRawBlock decompress = do nrObj <- getLong >>= sFromIntegral nrBytes <- getLong @@ -159,24 +197,20 @@ (marker, _) | marker /= sync -> Left "Invalid marker, does not match sync bytes." (_, rest) -> Right rest - getBlocks :: (BL.ByteString -> (BL.ByteString, T.LazyValue Type)) -- ^ getValue - -> BL.ByteString -- ^ sync bytes - -> BL.ByteString -- ^ byteString to parse - -> (BL.ByteString -> Get BL.ByteString) -- ^ how to decompress - -> (BL.ByteString, [[T.LazyValue Type]]) - getBlocks getValue sync bs decompress = - if BL.null bs - then (bs, []) - else case runGetOrFail (getRawBlock decompress) bs of - Left (bs', _, err) -> (bs', [[T.Error err]]) - Right (bs', _, (nrObj, bytes)) -> - let (_, vs) = consumeN (fromIntegral nrObj) getValue bytes - in case checkMarker sync bs' of - Left err -> (bs', [[T.Error err]]) - Right bs'' | BL.null bs'' -> (bs'', [vs]) - Right bs'' -> - let (rest, vs') = getBlocks getValue sync bs'' decompress - in (rest, vs : vs') +getContainerValuesWith :: (Schema -> BL.ByteString -> (BL.ByteString, T.LazyValue Type)) + -> BL.ByteString + -> Either String (Schema, [[T.LazyValue Type]]) +getContainerValuesWith schemaToGet bs = + case decodeRawBlocks bs of + Left err -> Left err + Right (sch, blocks) -> Right (sch, decodeBlocks (schemaToGet sch) blocks) + where + decodeBlocks getValue blocks = decodeBlock getValue <$> blocks + decodeBlock getValue v = case v of + Left err -> [T.Error err] + Right (nObj, bytes) -> + let (_, vs) = consumeN (fromIntegral nObj) getValue bytes + in vs decodeGet :: GetAvro a => (a -> T.LazyValue Type) -> BL.ByteString -> (BL.ByteString, T.LazyValue Type) decodeGet f bs = diff --git a/src/Data/Avro/Encode.hs b/src/Data/Avro/Encode.hs --- a/src/Data/Avro/Encode.hs +++ b/src/Data/Avro/Encode.hs @@ -9,7 +9,15 @@ ( -- * High level interface getSchema , encodeAvro - , encodeContainer, encodeContainerWithSync + , encodeContainer + , newSyncBytes + , encodeContainerWithSync + -- * Packing containers + , containerHeaderWithSync + , packContainerBlocks + , packContainerBlocksWithSync + , packContainerValues + , packContainerValuesWithSync -- * Lower level interface , EncodeAvro(..) , Zag(..) @@ -47,31 +55,43 @@ import Prelude as P import System.Entropy (getEntropy) -import Data.Avro.EncodeRaw -import Data.Avro.HasAvroSchema -import Data.Avro.Schema as S -import Data.Avro.Types as T -import Data.Avro.Zag -import Data.Avro.Zig +import Data.Avro.EncodeRaw +import Data.Avro.HasAvroSchema +import Data.Avro.Schema as S +import Data.Avro.Types as T +import Data.Avro.Zag +import Data.Avro.Zig encodeAvro :: EncodeAvro a => a -> BL.ByteString encodeAvro = toLazyByteString . putAvro +-- | Generates a new synchronization marker for encoding Avro containers +newSyncBytes :: IO BL.ByteString +newSyncBytes = BL.fromStrict <$> getEntropy 16 + -- |Encode chunks of objects into a container, using 16 random bytes for -- the synchronization markers. encodeContainer :: EncodeAvro a => Schema -> [[a]] -> IO BL.ByteString encodeContainer sch xss = - do sync <- getEntropy 16 - return $ encodeContainerWithSync sch (BL.fromStrict sync) xss + do sync <- newSyncBytes + return $ encodeContainerWithSync sch sync xss +-- | Creates an Avro container header for a given schema. +containerHeaderWithSync :: Schema -> BL.ByteString -> Builder +containerHeaderWithSync sch syncBytes = + lazyByteString avroMagicBytes <> + putAvro (HashMap.fromList [("avro.schema", A.encode sch), ("avro.codec","null")] :: HashMap Text BL.ByteString) <> + lazyByteString syncBytes + where + avroMagicBytes :: BL.ByteString + avroMagicBytes = "Obj" <> BL.pack [1] + -- |Encode chunks of objects into a container, using the provided -- ByteString as the synchronization markers. encodeContainerWithSync :: EncodeAvro a => Schema -> BL.ByteString -> [[a]] -> BL.ByteString encodeContainerWithSync sch syncBytes xss = toLazyByteString $ - lazyByteString avroMagicBytes <> - putAvro (HashMap.fromList [("avro.schema", A.encode sch), ("avro.codec","null")] :: HashMap Text BL.ByteString) <> - lazyByteString syncBytes <> + containerHeaderWithSync sch syncBytes <> foldMap putBlocks xss where putBlocks ys = @@ -82,14 +102,52 @@ putAvro nrBytes <> lazyByteString theBytes <> lazyByteString syncBytes - avroMagicBytes :: BL.ByteString - avroMagicBytes = "Obj" <> BL.pack [1] --- XXX make an instance 'EncodeAvro Schema' --- Would require a schema schema... --- encodeSchema :: EncodeAvro a => a -> BL.ByteString --- encodeSchema = toLazyByteString . putAvro . getSchema +-- | Packs a new container from a list of already encoded Avro blocks. +-- Each block is denoted as a pair of a number of objects within that block and the block content. +packContainerBlocks :: Schema -> [(Int, BL.ByteString)] -> IO BL.ByteString +packContainerBlocks sch blocks = do + sync <- newSyncBytes + pure $ packContainerBlocksWithSync sch sync blocks +-- | Packs a new container from a list of already encoded Avro blocks. +-- Each block is denoted as a pair of a number of objects within that block and the block content. +packContainerBlocksWithSync :: Schema -> BL.ByteString -> [(Int, BL.ByteString)] -> BL.ByteString +packContainerBlocksWithSync sch syncBytes blocks = + toLazyByteString $ + containerHeaderWithSync sch syncBytes <> + foldMap putBlock blocks + where + putBlock (nrObj, bytes) = + putAvro nrObj <> + putAvro (BL.length bytes) <> + lazyByteString bytes <> + lazyByteString syncBytes + +-- | Packs a container from a given list of already encoded Avro values +-- Each bytestring should represent exactly one one value serialised to Avro. +packContainerValues :: Schema -> [[BL.ByteString]] -> IO BL.ByteString +packContainerValues sch values = do + sync <- newSyncBytes + pure $ packContainerValuesWithSync sch sync values + +-- | Packs a container from a given list of already encoded Avro values +-- Each bytestring should represent exactly one one value serialised to Avro. +packContainerValuesWithSync :: Schema -> BL.ByteString -> [[BL.ByteString]] -> BL.ByteString +packContainerValuesWithSync sch syncBytes values = + toLazyByteString $ + containerHeaderWithSync sch syncBytes <> + foldMap putBlock values + where + putBlock ys = + let nrObj = P.length ys + nrBytes = DL.foldl' (\a b -> BL.length b + a) 0 ys + theBytes = mconcat $ lazyByteString <$> ys + in putAvro nrObj <> + putAvro nrBytes <> + theBytes <> + lazyByteString syncBytes + putAvro :: EncodeAvro a => a -> Builder putAvro = fst . runAvro . avro @@ -106,9 +164,6 @@ class EncodeAvro a where avro :: a -> AvroM - --- class PutAvro a where --- putAvro :: a -> Builder avroInt :: forall a. (FiniteBits a, Integral a, EncodeRaw a) => a -> AvroM avroInt n = AvroM (encodeRaw n, S.Int) diff --git a/src/Data/Avro/Schema.hs b/src/Data/Avro/Schema.hs --- a/src/Data/Avro/Schema.hs +++ b/src/Data/Avro/Schema.hs @@ -544,7 +544,6 @@ Union {..} -> concatMap go options Fixed {..} -> mk name aliases namespace Array {..} -> go item - Map {..} -> go values _ -> [] -- | Checks that two schemas match. This is like equality of schemas, diff --git a/test/Avro/Decode/Lazy/RawBlocksSpec.hs b/test/Avro/Decode/Lazy/RawBlocksSpec.hs new file mode 100644 --- /dev/null +++ b/test/Avro/Decode/Lazy/RawBlocksSpec.hs @@ -0,0 +1,76 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Avro.Decode.Lazy.RawBlocksSpec +where + +import Data.Avro as A +import Data.Avro.Decode.Lazy as DL +import Data.Avro.Decode.Lazy.Convert as TC +import Data.Avro.Deriving +import Data.Avro.Encode (packContainerBlocks, packContainerValues) +import Data.Either (isLeft, isRight, rights) +import Data.List (unfoldr) +import Data.Semigroup ((<>)) +import Data.Text (pack) + +import Test.Hspec + +{-# ANN module ("HLint: ignore Redundant do" :: String) #-} + +deriveAvro "test/data/small.avsc" + +spec :: Spec +spec = describe "Avro.Decode.Lazy.RawBlocksSpec" $ do + + it "should decode empty container" $ do + empty <- A.encodeContainer ([] :: [[Endpoint]]) + DL.decodeRawBlocks empty `shouldBe` Right (schema'Endpoint, []) + + it "should decode container with one block" $ do + container <- A.encodeContainer [mkEndpoint <$> [1, 2]] + let Right (s, bs) = DL.decodeRawBlocks container + s `shouldBe` schema'Endpoint + fmap fst <$> bs `shouldBe` [Right 2] + sequence bs `shouldSatisfy` isRight + + it "should decode container with multiple blocks" $ do + container <- A.encodeContainer (chunksOf 4 $ mkEndpoint <$> [1..10]) + let Right (s, bs) = DL.decodeRawBlocks container + s `shouldBe` schema'Endpoint + fmap fst <$> bs `shouldBe` [Right 4, Right 4, Right 2] + sequence bs `shouldSatisfy` isRight + + it "should repack container" $ do + let srcValues = mkEndpoint <$> [1..19] + srcContainer <- A.encodeContainer (chunksOf 4 srcValues) + let Right (s, bs) = DL.decodeRawBlocks srcContainer + tgtContainer <- packContainerBlocks s (rights bs) + let tgtValues = DL.decodeContainer tgtContainer + let allTgtValues = rights tgtValues + allTgtValues `shouldBe` srcValues + + it "should pack container with individual values" $ do + let srcValues = mkEndpoint <$> [1..19] + let values = A.encode <$> srcValues + container <- packContainerValues schema'Endpoint (chunksOf 4 values) + + let Right (s, bs) = DL.decodeRawBlocks container + s `shouldBe` schema'Endpoint + fst <$> rights bs `shouldBe` [4,4,4,4,3] + + rights (DL.decodeContainer container) `shouldBe` srcValues + +mkEndpoint :: Int -> Endpoint +mkEndpoint i = + Endpoint + { endpointIps = ["192.168.1." <> pack (show i), "127.0.0." <> pack (show i)] + , endpointPorts = [PortRange 1 10, PortRange 11 20] + , endpointOpaque = Opaque "16-b-long-string" + , endpointCorrelation = Opaque "opaq-correlation" + , endpointTag = Left (fromIntegral i) + } + +chunksOf :: Int -> [a] -> [[a]] +chunksOf n = takeWhile (not.null) . unfoldr (Just . splitAt n) diff --git a/test/Avro/SchemaSpec.hs b/test/Avro/SchemaSpec.hs deleted file mode 100644 --- a/test/Avro/SchemaSpec.hs +++ /dev/null @@ -1,38 +0,0 @@ -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TemplateHaskell #-} -module Avro.SchemaSpec -where - -import Data.Avro -import Data.Avro.Deriving (makeSchema) -import Data.Avro.Schema (buildTypeEnvironment, matches) - -import Test.Hspec - -{-# ANN module ("HLint: ignore Redundant do" :: String) #-} - -spec :: Spec -spec = describe "Avro.SchemaSpec" $ do - describe "buildTypeEnvironment" $ - it "should contain definitions for all internal types" $ do - let schema = $(makeSchema "test/data/internal-bindings.avsc") - environment = buildTypeEnvironment err schema - err name = fail $ "Missing " ++ show name ++ " in environment." - expected = - [ "InternalBindings" - , "InField" - , "NestedInField" - , "AliasNestedInField" - , "NestedEnum" - , "NestedFixed" - , "InArray" - , "NestedInArray" - , "InMap" - , "NestedInMap" - , "InUnionA" - , "InUnionB" - ] - definitions <- traverse environment expected - length definitions `shouldBe` length expected diff --git a/test/data/internal-bindings.avsc b/test/data/internal-bindings.avsc deleted file mode 100644 --- a/test/data/internal-bindings.avsc +++ /dev/null @@ -1,105 +0,0 @@ -{ - "name" : "InternalBindings", - "type" : "record", - "doc" : "A test record that includes subdefinitions nested in different ways.", - "fields" : [ - { - "name" : "inField", - "doc" : "A record definition nested in a field of a record.", - "type" : { - "name" : "InField", - "type" : "record", - "fields" : [ - { - "name" : "nestedInField", - "doc" : "A record definition nested in two other records.", - "type" : { - "name" : "NestedInField", - "type" : "record", - "aliases" : ["AliasNestedInField"], - "fields" : [] - } - }, - { - "name" : "nestedEnum", - "doc" : "An enum definition nested in a nested record.", - "type" : { - "type" : "enum", - "name" : "NestedEnum", - "symbols" : ["Foo", "Bar"] - } - }, - { - "name" : "nestedFixed", - "doc" : "A fixed definition nested in a nested record.", - "type" : { - "type" : "fixed", - "size" : 42, - "name" : "NestedFixed" - } - } - ] - } - }, - { - "name" : "inArray", - "doc" : "A record definition nested in an array type.", - "type" : { - "type" : "array", - "items" : { - "name" : "InArray", - "type" : "record", - "fields" : [ - { - "name" : "nestedInArray", - "doc" : "A record definition nested inside a record defined in an array.", - "type" : { - "name" : "NestedInArray", - "type" : "record", - "fields" : [] - } - } - ] - } - } - }, - { - "name" : "inMap", - "doc" : "A record definition nested in a map type.", - "type" : { - "type" : "map", - "values" : { - "name" : "InMap", - "type" : "record", - "fields" : [ - { - "name" : "nestedInMap", - "doc" : "A record definition nested inside a record defined in a map.", - "type" : { - "name" : "NestedInMap", - "type" : "record", - "fields" : [] - } - } - ] - } - } - }, - { - "name" : "inUnion", - "doc" : "Record definitions nested in a union.", - "type" : [ - { - "name" : "InUnionA", - "type" : "record", - "fields" : [] - }, - { - "name" : "InUnionB", - "type" : "record", - "fields" : [] - } - ] - } - ] -}