avro 0.3.4.3 → 0.3.5.0

raw patch · 8 files changed

+219/−198 lines, 8 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• avro.cabal +3/−4
• src/Data/Avro.hs +1/−0
• src/Data/Avro/Decode/Lazy.hs +63/−29
• src/Data/Avro/Encode.hs +76/−21
• src/Data/Avro/Schema.hs +0/−1
• test/Avro/Decode/Lazy/RawBlocksSpec.hs +76/−0
• test/Avro/SchemaSpec.hs +0/−38
• test/data/internal-bindings.avsc +0/−105

@@ -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

@@ -72,6 +72,7 @@   ( FromAvro(..)   , ToAvro(..)   , HasAvroSchema(..)+  , Schema   , Avro   , (.:)   , (.=), record, fixed

@@ -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 =

@@ -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)

@@ -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,

@@ -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)

@@ -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

@@ -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" : []-        }-      ]-    }-  ]-}