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: 83c35a316bced7d77c45fea0d0d8b228cdff0c26f57a07eedb698798addc4697 +-- hash: df77e046e65bd987123d33da69b0e843f877bd76dea4406b0e1fd22d9fd5a8f8 name: avro -version: 0.3.5.0 +version: 0.3.5.1 synopsis: Avro serialization support for Haskell description: Avro serialization and deserialization support for Haskell category: Data @@ -56,6 +56,8 @@ Data.Avro.Decode.Lazy.Convert Data.Avro.Decode.Lazy.Deconflict Data.Avro.Decode.Lazy.LazyValue + Data.Avro.Decode.Strict + Data.Avro.Decode.Strict.Internal Data.Avro.DecodeRaw Data.Avro.Deconflict Data.Avro.Deriving @@ -122,6 +124,7 @@ Avro.Codec.ZigZagSpec Avro.Decode.Lazy.ContainerSpec Avro.Decode.Lazy.RawBlocksSpec + Avro.Decode.Lazy.RawValuesSpec Avro.Decode.Lazy.ValuesSpec Avro.Deconflict.Reader Avro.Deconflict.Writer diff --git a/src/Data/Avro.hs b/src/Data/Avro.hs --- a/src/Data/Avro.hs +++ b/src/Data/Avro.hs @@ -88,34 +88,35 @@ , schemaOf ) where -import Control.Arrow (first) -import qualified Data.Avro.Decode as D -import Data.Avro.Deconflict as C -import qualified Data.Avro.Encode as E -import Data.Avro.Schema as S -import Data.Avro.Types as T -import qualified Data.Binary.Get as G -import qualified Data.Binary.Put as P -import qualified Data.ByteString as B -import Data.ByteString.Lazy (ByteString) -import qualified Data.ByteString.Lazy as BL -import Data.Foldable (toList) -import qualified Data.HashMap.Strict as HashMap +import Control.Arrow (first) +import qualified Data.Avro.Decode as D +import qualified Data.Avro.Decode.Lazy as DL +import Data.Avro.Deconflict as C +import qualified Data.Avro.Encode as E +import Data.Avro.Schema as S +import Data.Avro.Types as T +import qualified Data.Binary.Get as G +import qualified Data.Binary.Put as P +import qualified Data.ByteString as B +import Data.ByteString.Lazy (ByteString) +import qualified Data.ByteString.Lazy as BL +import Data.Foldable (toList) +import qualified Data.HashMap.Strict as HashMap import Data.Int -import Data.List.NonEmpty (NonEmpty (..)) -import qualified Data.Map as Map -import Data.Monoid ((<>)) +import Data.List.NonEmpty (NonEmpty (..)) +import qualified Data.Map as Map +import Data.Monoid ((<>)) import Data.Tagged -import Data.Text (Text) -import qualified Data.Text as Text -import qualified Data.Text.Lazy as TL -import qualified Data.Vector as V +import Data.Text (Text) +import qualified Data.Text as Text +import qualified Data.Text.Lazy as TL +import qualified Data.Vector as V import Data.Word -import Prelude as P +import Prelude as P -import Data.Avro.FromAvro -import Data.Avro.HasAvroSchema -import Data.Avro.ToAvro +import Data.Avro.FromAvro +import Data.Avro.HasAvroSchema +import Data.Avro.ToAvro type Avro a = (FromAvro a, ToAvro a) diff --git a/src/Data/Avro/Decode.hs b/src/Data/Avro/Decode.hs --- a/src/Data/Avro/Decode.hs +++ b/src/Data/Avro/Decode.hs @@ -41,10 +41,12 @@ import Data.Avro.Decode.Get import Data.Avro.DecodeRaw -import Data.Avro.Schema as S -import qualified Data.Avro.Types as T +import Data.Avro.Schema as S +import qualified Data.Avro.Types as T import Data.Avro.Zag +import Data.Avro.Decode.Strict.Internal + -- |Decode bytes into a 'Value' as described by Schema. decodeAvro :: Schema -> BL.ByteString -> Either String (T.Value Type) decodeAvro sch = either (\(_,_,s) -> Left s) (\(_,_,a) -> Right a) . runGetOrFail (getAvroOf sch) @@ -92,60 +94,3 @@ | Just x <- code = fail ("Unrecognized codec: " <> BC.unpack x) | otherwise = return return - -{-# INLINABLE getAvroOf #-} -getAvroOf :: Schema -> Get (T.Value Type) -getAvroOf ty0 = go ty0 - where - env = S.buildTypeEnvironment envFail ty0 - envFail t = fail $ "Named type not in schema: " <> show t - - go :: Type -> Get (T.Value Type) - go ty = - case ty of - Null -> return T.Null - Boolean -> T.Boolean <$> getAvro - Int -> T.Int <$> getAvro - Long -> T.Long <$> getAvro - Float -> T.Float <$> getAvro - Double -> T.Double <$> getAvro - Bytes -> T.Bytes <$> getAvro - String -> T.String <$> getAvro - Array t -> - do vals <- getBlocksOf t - return $ T.Array (V.fromList $ mconcat vals) - Map t -> - do kvs <- getKVBlocks t - return $ T.Map (HashMap.fromList $ mconcat kvs) - NamedType tn -> env tn >>= go - Record {..} -> - do let getField Field {..} = (fldName,) <$> go fldType - T.Record ty . HashMap.fromList <$> mapM getField fields - Enum {..} -> - do val <- getLong - let sym = fromMaybe "" (symbolLookup val) -- empty string for 'missing' symbols (alternative is an error or exception) - pure (T.Enum ty (fromIntegral val) sym) - Union ts unionLookup -> - do i <- getLong - case unionLookup i of - Nothing -> fail $ "Decoded Avro tag is outside the expected range for a Union. Tag: " <> show i <> " union of: " <> show (P.map typeName $ NE.toList ts) - Just t -> T.Union ts t <$> go t - Fixed {..} -> T.Fixed ty <$> G.getByteString (fromIntegral size) - - getKVBlocks :: Type -> Get [[(Text,T.Value Type)]] - getKVBlocks t = - do blockLength <- abs <$> getLong - if blockLength == 0 - then return [] - else do vs <- replicateM (fromIntegral blockLength) ((,) <$> getString <*> go t) - (vs:) <$> getKVBlocks t - {-# INLINE getKVBlocks #-} - - getBlocksOf :: Type -> Get [[T.Value Type]] - getBlocksOf t = - do blockLength <- abs <$> getLong - if blockLength == 0 - then return [] - else do vs <- replicateM (fromIntegral blockLength) (go t) - (vs:) <$> getBlocksOf t - {-# INLINE getBlocksOf #-} 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 @@ -19,50 +19,55 @@ -- * Lower level interface , getContainerValues , getContainerValuesWith + , getContainerValuesBytes + , getContainerValuesBytes' , getAvroOf , GetAvro(..) ) where -import qualified Codec.Compression.Zlib as Z -import Control.Monad (foldM, replicateM, when) -import qualified Data.Aeson as A -import qualified Data.Array as Array -import Data.Binary.Get (Get, runGetOrFail) -import qualified Data.Binary.Get as G -import Data.Binary.IEEE754 as IEEE +import qualified Codec.Compression.Zlib as Z +import Control.Monad (foldM, replicateM, when) +import qualified Data.Aeson as A +import qualified Data.Array as Array +import Data.Binary.Get (Get, runGetOrFail) +import qualified Data.Binary.Get as G +import Data.Binary.IEEE754 as IEEE import Data.Bits -import Data.ByteString (ByteString) -import qualified Data.ByteString.Lazy as BL -import qualified Data.ByteString.Lazy.Char8 as BL -import Data.Either (isRight) -import qualified Data.HashMap.Strict as HashMap +import Data.ByteString (ByteString) +import qualified Data.ByteString.Lazy as BL +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', unfoldr) -import qualified Data.List.NonEmpty as NE -import qualified Data.Map as Map +import Data.List (foldl', unfoldr) +import qualified Data.List.NonEmpty as NE +import qualified Data.Map as Map import Data.Maybe -import Data.Monoid ((<>)) -import qualified Data.Set as Set -import Data.Tagged (Tagged, untag) -import Data.Text (Text) -import qualified Data.Text as Text -import qualified Data.Text.Encoding as Text -import qualified Data.Vector as V -import Prelude as P +import Data.Monoid ((<>)) +import qualified Data.Set as Set +import Data.Tagged (Tagged, untag) +import Data.Text (Text) +import qualified Data.Text as Text +import qualified Data.Text.Encoding as Text +import qualified Data.Vector as V +import Prelude as P -import qualified Data.Avro.Decode.Lazy.LazyValue as T +import qualified Data.Avro.Decode.Lazy.LazyValue as T import Data.Avro.DecodeRaw -import Data.Avro.HasAvroSchema (schema) -import Data.Avro.Schema as S +import Data.Avro.HasAvroSchema (schema) +import Data.Avro.Schema as S +import qualified Data.Avro.Types as TypesStrict import Data.Avro.Zag -import Data.Avro.Decode.Get -import Data.Avro.Decode.Lazy.Convert (toStrictValue) -import Data.Avro.Decode.Lazy.Deconflict as C -import Data.Avro.FromAvro +import qualified Data.Avro.Decode.Strict.Internal as DecodeStrict -import Debug.Trace +import Data.Avro.Decode.Get +import Data.Avro.Decode.Lazy.Convert (toStrictValue) +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 @@ -167,8 +172,8 @@ 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) + Right Nothing -> Nothing + Left err -> Just (Left err, Nothing) getNextBlock :: BL.ByteString -> (BL.ByteString -> Get BL.ByteString) @@ -195,7 +200,7 @@ checkMarker sync bs = case BL.splitAt nrSyncBytes bs of (marker, _) | marker /= sync -> Left "Invalid marker, does not match sync bytes." - (_, rest) -> Right rest + (_, rest) -> Right rest getContainerValuesWith :: (Schema -> BL.ByteString -> (BL.ByteString, T.LazyValue Type)) -> BL.ByteString @@ -218,6 +223,50 @@ in either (\(rest,_,s) -> (rest, T.Error s)) (\(rest,_,a) -> (rest, a)) res {-# INLINE decodeGet #-} +-- | Splits container into a list of individual avro-encoded values. +-- +-- This is particularly useful when slicing up containers into one or more +-- smaller files. By extracting the original bytestring it is possible to +-- avoid re-encoding data. +getContainerValuesBytes :: BL.ByteString -> Either String (Schema, [Either String BL.ByteString]) +getContainerValuesBytes = + extractContainerValues readBytes + where + readBytes sch = do + start <- G.bytesRead + end <- G.lookAhead (DecodeStrict.getAvroOf sch >> G.bytesRead) + G.getLazyByteString (end-start) + +-- | Splits container into a list of individual avro-encoded values. +-- This version provides both encoded and decoded values. +-- +-- This is particularly useful when slicing up containers into one or more +-- smaller files. By extracting the original bytestring it is possible to +-- avoid re-encoding data. +getContainerValuesBytes' :: BL.ByteString -> Either String (Schema, [Either String (TypesStrict.Value S.Type, BL.ByteString)]) +getContainerValuesBytes' = + extractContainerValues readBytes + where + readBytes sch = do + start <- G.bytesRead + (val, end) <- G.lookAhead (DecodeStrict.getAvroOf sch >>= (\v -> (v, ) <$> G.bytesRead)) + res <- G.getLazyByteString (end-start) + pure (val, res) + +extractContainerValues :: (Schema -> Get a) -> BL.ByteString -> Either String (Schema, [Either String a]) +extractContainerValues f bs = + case decodeRawBlocks bs of + Left err -> Left err + Right (sch, blocks) -> Right (sch, blocks >>= decodeBlock sch) + where + decodeBlock _ (Left err) = undefined + decodeBlock sch (Right (nrObj, bytes)) = snd $ consumeN (fromIntegral nrObj) (decodeValue sch) bytes + + decodeValue sch bytes = + case G.runGetOrFail (f sch) bytes of + Left (bs', _, err) -> (bs', Left err) + Right (bs', _, res) -> (bs', Right res) + consumeN :: Int64 -> (a -> (a, b)) -> a -> (a, [b]) consumeN n f a = if n == 0 @@ -262,7 +311,7 @@ Left (bs', _, err) -> (bs', T.Error err) Right (bs', _, i) -> case symbolLookup i of - Nothing -> (bs', T.Error ("Unknown value {" <> show i <> "} for enum " <> Text.unpack (typeName ty) )) + Nothing -> (bs', T.Error ("Unknown value {" <> show i <> "} for enum " <> Text.unpack (typeName ty) )) Just sym -> (bs', T.Enum ty (fromIntegral i) sym) Union ts unionLookup -> diff --git a/src/Data/Avro/Decode/Strict.hs b/src/Data/Avro/Decode/Strict.hs new file mode 100644 --- /dev/null +++ b/src/Data/Avro/Decode/Strict.hs @@ -0,0 +1,3 @@ +module Data.Avro.Decode.Strict +where + diff --git a/src/Data/Avro/Decode/Strict/Internal.hs b/src/Data/Avro/Decode/Strict/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Data/Avro/Decode/Strict/Internal.hs @@ -0,0 +1,96 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiWayIf #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} +module Data.Avro.Decode.Strict.Internal +where + +import qualified Codec.Compression.Zlib as Z +import Control.Monad (replicateM, when) +import qualified Data.Aeson as A +import qualified Data.Array as Array +import Data.Binary.Get (Get, runGetOrFail) +import qualified Data.Binary.Get as G +import Data.Binary.IEEE754 as IEEE +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.HashMap.Strict as HashMap +import Data.Int +import Data.List (foldl') +import qualified Data.List.NonEmpty as NE +import qualified Data.Map as Map +import Data.Maybe +import Data.Monoid ((<>)) +import qualified Data.Set as Set +import Data.Text (Text) +import qualified Data.Text as Text +import qualified Data.Text.Encoding as Text +import qualified Data.Vector as V +import Prelude as P + +import Data.Avro.Decode.Get +import Data.Avro.DecodeRaw +import Data.Avro.Schema as S +import qualified Data.Avro.Types as T +import Data.Avro.Zag + +{-# INLINABLE getAvroOf #-} +getAvroOf :: Schema -> Get (T.Value Type) +getAvroOf ty0 = go ty0 + where + env = S.buildTypeEnvironment envFail ty0 + envFail t = fail $ "Named type not in schema: " <> show t + + go :: Type -> Get (T.Value Type) + go ty = + case ty of + Null -> return T.Null + Boolean -> T.Boolean <$> getAvro + Int -> T.Int <$> getAvro + Long -> T.Long <$> getAvro + Float -> T.Float <$> getAvro + Double -> T.Double <$> getAvro + Bytes -> T.Bytes <$> getAvro + String -> T.String <$> getAvro + Array t -> + do vals <- getBlocksOf t + return $ T.Array (V.fromList $ mconcat vals) + Map t -> + do kvs <- getKVBlocks t + return $ T.Map (HashMap.fromList $ mconcat kvs) + NamedType tn -> env tn >>= go + Record {..} -> + do let getField Field {..} = (fldName,) <$> go fldType + T.Record ty . HashMap.fromList <$> mapM getField fields + Enum {..} -> + do val <- getLong + let sym = fromMaybe "" (symbolLookup val) -- empty string for 'missing' symbols (alternative is an error or exception) + pure (T.Enum ty (fromIntegral val) sym) + Union ts unionLookup -> + do i <- getLong + case unionLookup i of + Nothing -> fail $ "Decoded Avro tag is outside the expected range for a Union. Tag: " <> show i <> " union of: " <> show (P.map typeName $ NE.toList ts) + Just t -> T.Union ts t <$> go t + Fixed {..} -> T.Fixed ty <$> G.getByteString (fromIntegral size) + + getKVBlocks :: Type -> Get [[(Text,T.Value Type)]] + getKVBlocks t = + do blockLength <- abs <$> getLong + if blockLength == 0 + then return [] + else do vs <- replicateM (fromIntegral blockLength) ((,) <$> getString <*> go t) + (vs:) <$> getKVBlocks t + {-# INLINE getKVBlocks #-} + + getBlocksOf :: Type -> Get [[T.Value Type]] + getBlocksOf t = + do blockLength <- abs <$> getLong + if blockLength == 0 + then return [] + else do vs <- replicateM (fromIntegral blockLength) (go t) + (vs:) <$> getBlocksOf t + {-# INLINE getBlocksOf #-} diff --git a/test/Avro/Decode/Lazy/RawValuesSpec.hs b/test/Avro/Decode/Lazy/RawValuesSpec.hs new file mode 100644 --- /dev/null +++ b/test/Avro/Decode/Lazy/RawValuesSpec.hs @@ -0,0 +1,60 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Avro.Decode.Lazy.RawValuesSpec +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.Avro.Schema (resultToEither) +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.RawValuesSpec" $ do + + it "should decode empty container" $ do + empty <- A.encodeContainer ([] :: [[Endpoint]]) + DL.getContainerValuesBytes empty `shouldBe` Right (schema'Endpoint, []) + + it "should decode container with one block" $ do + let msgs = mkEndpoint <$> [1, 2] + container <- A.encodeContainer [msgs] + let Right (sch, vals) = DL.getContainerValuesBytes container + sch `shouldBe` schema'Endpoint + let results = resultToEither . A.decode <$> rights vals + rights results `shouldBe` msgs + + + it "should decode container with multiple blocks" $ do + let msgs = mkEndpoint <$> [1..19] + container <- A.encodeContainer (chunksOf 4 msgs) + let Right (sch, vals) = DL.getContainerValuesBytes container + sch `shouldBe` schema'Endpoint + let results = resultToEither . A.decode <$> rights vals + rights results `shouldBe` msgs + +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)