Zarr v3 Prototype

Project.toml

@@ -6,6 +6,7 @@ version = "0.9.5"
 [deps]
 AWSS3 = "1c724243-ef5b-51ab-93f4-b0a88ac62a95"
 Blosc = "a74b3585-a348-5f62-a45c-50e91977d574"
+CRC32c = "8bf52ea8-c179-5cab-976a-9e18b702a9bc"
 ChunkCodecCore = "0b6fb165-00bc-4d37-ab8b-79f91016dbe1"
 ChunkCodecLibZlib = "4c0bbee4-addc-4d73-81a0-b6caacae83c8"
 ChunkCodecLibZstd = "55437552-ac27-4d47-9aa3-63184e8fd398"
@@ -27,6 +28,7 @@ Blosc = "0.5, 0.6, 0.7"
 ChunkCodecCore = "1"
 ChunkCodecLibZlib = "1"
 ChunkCodecLibZstd = "1"
+CRC32c = "1.10, 1.11"
 DataStructures = "0.17, 0.18, 0.19"
 DateTimes64 = "1"
 DiskArrays = "0.4.2"

docs/src/tutorial.md

@@ -187,7 +187,7 @@
 
 When using a compressor, it can be useful to get some diagnostics on the compression ratio. `ZArrays` provide a `zinfo` function which can be used to print some diagnostics, e.g.:
 
 ```jldoctest compress
 julia> zinfo(z)
 Type                : ZArray
 Data type           : Int32
@@ -197,7 +197,7 @@
 Read-Only           : false
 Compressor          : Zarr.BloscCompressor(0, 3, "zstd", 1)
 Filters             : nothing
-Store type          : Dictionary Storage
+Store type          : Zarr.VersionedStore{2, '.', Zarr.DictStore}(Dictionary Storage)
 No. bytes           : 400000000
 No. bytes stored    : 2412289
 Storage ratio       : 165.81761140559857

src/Codecs/Codecs.jl

@@ -0,0 +1,49 @@
+module Codecs
+
+using JSON: JSON
+
+"""
+    abstract type Codec
+
+The abstract supertype for all Zarr codecs
+
+## Interface
+
+All subtypes of `Codec` SHALL implement the following methods:
+
+- `zencode(a, c::Codec)`: compress the array `a` using the codec `c`.
+- `zdecode(a, c::Codec, T)`: decode the array `a` using the codec `c` 
+  and return an array of type `T`.
+- `JSON.lower(c::Codec)`: return a JSON representation of the codec `c`, which 
+  follows the Zarr specification for that codec.
+- `getCodec(::Type{<:Codec}, d::Dict)`: return a codec object from a given 
+  dictionary `d` which contains the codec's parameters according to the Zarr spec.
+
+Subtypes of `Codec` MAY also implement the following methods:
+
+- `zencode!(encoded, data, c::Codec)`: encode the array `data` using the 
+  codec `c` and store the result in the array `encoded`.
+- `zdecode!(data, encoded, c::Codec)`: decode the array `encoded` 
+  using the codec `c` and store the result in the array `data`.
+
+Finally, an entry MUST be added to the `VN.codectypes` dictionary for each codec type where N is the
+Zarr format version.
+This must also follow the Zarr specification's name for that compressor.  The name of the compressor
+is the key, and the value is the compressor type (e.g. `BloscCodec` or `NoCodec`).
+
+For example, the Blosc codec is named "blosc" in the Zarr spec, so the entry for [`BloscCodec`](@ref) 
+must be added to `codectypes` as `codectypes["blosc"] = BloscCodec`.
+"""
+
+abstract type Codec end
+
+zencode(a, c::Codec) = error("Unimplemented")
+zencode!(encoded, data, c::Codec) = error("Unimplemented")
+zdecode(a, c::Codec, T::Type) = error("Unimplemented")
+zdecode!(data, encoded, c::Codec) = error("Unimplemented")
+JSON.lower(c::Codec) = error("Unimplemented")
+getCodec(::Type{<:Codec}, d::Dict) = error("Unimplemented")
+
+include("V3/V3.jl")
+
+end

src/Codecs/V3/V3.jl

@@ -0,0 +1,103 @@
+module V3Codecs
+
+import ..Codecs: zencode, zdecode, zencode!, zdecode!
+using CRC32c: CRC32c
+
+abstract type V3Codec{In,Out} end
+const codectypes = Dict{String, V3Codec}()
+
+@enum BloscCompressor begin
+    lz4
+    lz4hc
+    blosclz
+    zstd
+    snappy
+    zlib
+end
+
+@enum BloscShuffle begin
+    noshuffle
+    shuffle
+    bitshuffle
+end
+
+struct BloscCodec <: V3Codec{:bytes, :bytes}
+    cname::BloscCompressor
+    clevel::Int64
+    shuffle::BloscShuffle
+    typesize::UInt8
+    blocksize::UInt
+end
+name(::BloscCodec) = "blosc"
+
+struct BytesCodec <: V3Codec{:array, :bytes}
+end
+name(::BytesCodec) = "bytes"
+
+struct CRC32cCodec <: V3Codec{:bytes, :bytes}
+end
+name(::CRC32cCodec) = "crc32c"
+
+struct GzipCodec <: V3Codec{:bytes, :bytes}
+end
+name(::GzipCodec) = "gzip"
+
+
+#=
+zencode(a, c::Codec) = error("Unimplemented")
+zencode!(encoded, data, c::Codec) = error("Unimplemented")
+zdecode(a, c::Codec, T::Type) = error("Unimplemented")
+zdecode!(data, encoded, c::Codec) = error("Unimplemented")
+=#
+
+function crc32c_stream!(output::IO, input::IO; buffer = Vector{UInt8}(undef, 1024*32))
+    hash::UInt32 = 0x00000000
+    while(bytesavailable(input) > 0)
+        sized_buffer = @view(buffer[1:min(length(buffer), bytesavailable(input))])
+        read!(input, sized_buffer)
+        write(output, sized_buffer)
+        hash = CRC32c.crc32c(sized_buffer, hash)
+    end
+    return hash
+end
+function zencode!(encoded::Vector{UInt8}, data::Vector{UInt8}, c::CRC32cCodec)
+    output = IOBuffer(encoded, read=false, write=true)
+    input = IOBuffer(data, read=true, write=false)
+    zencode!(output, input, c)
+    return take!(output)
+end
+function zencode!(output::IO, input::IO, c::CRC32cCodec)
+    hash = crc32c_stream!(output, input)
+    write(output, hash)
+    return output
+end
+function zdecode!(encoded::Vector{UInt8}, data::Vector{UInt8}, c::CRC32cCodec)
+    output = IOBuffer(encoded, read=false, write=true)
+    input = IOBuffer(data, read=true, write=true)
+    zdecode!(output, input, c)
+    return take!(output)
+end
+function zdecode!(output::IOBuffer, input::IOBuffer, c::CRC32cCodec)
+    input_vec = take!(input)
+    truncated_input = IOBuffer(@view(input_vec[1:end-4]); read=true, write=false)
+    hash = crc32c_stream!(output, truncated_input)
+    if input_vec[end-3:end] != reinterpret(UInt8, [hash])
+        throw(IOError("CRC32c hash does not match"))
+    end
+    return output
+end
+
+struct ShardingCodec{N} <: V3Codec{:array, :bytes}
+    chunk_shape::NTuple{N,Int}
+    codecs::Vector{V3Codec}
+    index_codecs::Vector{V3Codec}
+    index_location::Symbol
+end
+name(::ShardingCodec) = "sharding_indexed"
+
+struct TransposeCodec <: V3Codec{:array, :array}
+end
+name(::TransposeCodec) = "transpose"
+
+
+end

src/Compressors/Compressors.jl

@@ -49,10 +49,13 @@ const compressortypes = Dict{Union{String,Nothing}, Type{<: Compressor}}()
 include("blosc.jl")
 include("zlib.jl")
 include("zstd.jl")
+include("v3.jl")
 
 # ## Fallback definitions for the compressor interface
 # Define fallbacks and generic methods for the compressor interface
-getCompressor(compdict::Dict) = getCompressor(compressortypes[compdict["id"]],compdict)
+getCompressor(compdict::Dict) = haskey(compdict, "id") ?
+    getCompressor(compressortypes[compdict["id"]], compdict) :
+    getCompressor(compressortypes[compdict["name"]], compdict["configuration"])
 getCompressor(::Nothing) = NoCompressor()
 
 # Compression when no filter is given
@@ -104,4 +107,4 @@ end
 
 JSON.lower(::NoCompressor) = nothing
 
-compressortypes[nothing] = NoCompressor
+compressortypes[nothing] = NoCompressor

src/Compressors/v3.jl

@@ -0,0 +1,58 @@
+"""
+    Compressor v3{C <: Compressor} <: Compressor
+
+Wrapper to indicate Zarr v3 of a compressor
+"""
+struct Compressor_v3{C} <: Compressor
+    parent::C
+end
+Base.parent(c::Compressor_v3) = c.parent
+
+function zuncompress(a, z::Compressor_v3, T)
+    zuncompress(a, parent(z), T)
+end
+
+function zuncompress!(data::DenseArray, compressed, z::Compressor_v3)
+    zuncompress!(data, compressed, parent(z))
+end
+
+function zcompress(a, z::Compressor_v3)
+    zcompress(a, parent(z))
+end
+
+
+function JSON.lower(c::Compressor_v3{BloscCompressor})
+    p = parent(c)
+    return Dict(
+        "name" => "blosc",
+        "configuration" => Dict(
+            "cname" => p.cname,
+            "clevel" => p.clevel,
+            "shuffle" => p.shuffle,
+# TODO: Evaluate if we can encode typesize
+#            "typesize" => p.typesize,
+            "blocksize" => p.blocksize
+        )
+    )
+end
+
+function JSON.lower(c::Compressor_v3{ZlibCompressor})
+    p = parent(c)
+    return Dict(
+        "name" => "gzip",
+        "configuration" => Dict(
+            "level" => p.clevel
+        )
+    )
+end
+
+function JSON.lower(c::Compressor_v3{ZstdCompressor})
+    p = parent(c)
+    return Dict(
+        "name" => "zstd",
+        "configuration" => Dict(
+            "level" => p.config.compressionlevel,
+            "checksum" => p.config.checksum
+        )
+    )
+end

src/Compressors/zstd.jl

@@ -4,6 +4,7 @@
 This file implements a Zstd compressor via ChunkCodecLibZstd.jl.
 
 =#
+
 using ChunkCodecLibZstd: ZstdEncodeOptions
 using ChunkCodecCore: encode, decode, decode!
 
@@ -51,4 +52,4 @@ function JSON.lower(z::ZstdCompressor)
     end
 end
 
-Zarr.compressortypes["zstd"] = ZstdCompressor
+Zarr.compressortypes["zstd"] = ZstdCompressor

src/Storage/Storage.jl

@@ -108,8 +108,12 @@ function writeattrs(s::AbstractStore, p, att::Dict; indent_json::Bool= false)
   att
 end
 
-is_zgroup(s::AbstractStore, p) = isinitialized(s,_concatpath(p,".zgroup"))
-is_zarray(s::AbstractStore, p) = isinitialized(s,_concatpath(p,".zarray"))
+is_zarr3(s::AbstractStore, p) = isinitialized(s,_concatpath(p,"zarr.json"))
+is_zarr2(s::AbstractStore, p) = is_z2array(s, p) || is_z2group(s,p)
+is_zgroup(s::AbstractStore, p) = is_z2group(s,p)
+is_zarray(s::AbstractStore, p) = is_z2array(s,p)
+is_z2group(s::AbstractStore, p) = isinitialized(s,_concatpath(p,".zgroup"))
+is_z2array(s::AbstractStore, p) = isinitialized(s,_concatpath(p,".zarray"))
 
 isinitialized(s::AbstractStore, p, i::CartesianIndex)=isinitialized(s,p,citostring(i))
 isinitialized(s::AbstractStore, p, i) = isinitialized(s,_concatpath(p,i))
@@ -197,6 +201,7 @@ isemptysub(s::AbstractStore, p) = isempty(subkeys(s,p)) && isempty(subdirs(s,p))
 #during auto-check of storage format when doing zopen
 storageregexlist = Pair[]
 
+include("formattedstore.jl")
 include("directorystore.jl")
 include("dictstore.jl")
 include("s3store.jl")