perf: optimize CDF variable access by removing redundant VVR caching and disabling chunking

Author: Beforerr

src/loading/variable.jl

@@ -26,16 +26,10 @@ function variable(cdf::CDFDataset, name)
     record_dims = record_sizes(vdr)
     dims = (record_dims..., vdr.max_rec + 1)
     N = vdr.z_num_dims + 1
-    vvrs, vvr_type = read_vvrs(vdr)
-    compression = if !isempty(vvrs) #  # vvr records is the ultimative source
-        vvr_type == VVR_ ? NoCompression : variable_compression(vdr)
-    else
-        NoCompression
-    end
     byte_swap = is_big_endian_encoding(cdf.cdr.encoding)
 
     return CDFVariable{T, N, typeof(vdr), typeof(cdf)}(
-        name, vdr, cdf, dims, vvrs, compression, byte_swap
+        name, vdr, cdf, dims, byte_swap
     )
 end
 
@@ -45,15 +39,20 @@ function DiskArrays.readblock!(var::CDFVariable{T, N}, dest::AbstractArray{T}, r
 
     buffer = parent(var.parentdataset)
     RecordSizeType = recordsize_type(var.parentdataset)
-    entries = var.vvrs
+    entries, vvr_type = read_vvrs(var.vdr)
     isempty(entries) && return dest
+    compression = if !isempty(entries) #  # vvr records is the ultimative source
+        vvr_type == VVR_ ? NoCompression : variable_compression(var.vdr)
+    else
+        NoCompression
+    end
 
     record_range = ranges[end]
     other_ranges = ranges[1:(N - 1)]
     dims_without_record = var.dims[1:(N - 1)]
 
     is_full_record = length.(other_ranges) == dims_without_record
-    is_no_compression = var.compression == NoCompression
+    is_no_compression = compression == NoCompression
 
     first_rec = first(record_range)
     last_rec = last(record_range)
@@ -87,7 +86,7 @@ function DiskArrays.readblock!(var::CDFVariable{T, N}, dest::AbstractArray{T}, r
                 dest_view = selectdim(dest, N, dest_range)
                 total_elems = record_size * length(entry)
                 decompressor = take!(decompressors())
-                load_cvvr_data!(dest_view, 1, buffer, entry.offset, total_elems, RecordSizeType, var.compression; decompressor)
+                load_cvvr_data!(dest_view, 1, buffer, entry.offset, total_elems, RecordSizeType, compression; decompressor)
                 put!(decompressors(), decompressor)
             else
                 # partial entry
@@ -101,7 +100,7 @@ function DiskArrays.readblock!(var::CDFVariable{T, N}, dest::AbstractArray{T}, r
                     load_vvr_data!(chunk, 1, buffer, entry.offset, total_elems, RecordSizeType)
                 else
                     decompressor = take!(decompressors())
-                    load_cvvr_data!(chunk, 1, buffer, entry.offset, total_elems, RecordSizeType, var.compression; decompressor)
+                    load_cvvr_data!(chunk, 1, buffer, entry.offset, total_elems, RecordSizeType, compression; decompressor)
                     put!(decompressors(), decompressor)
                 end
 

src/variable.jl

@@ -13,8 +13,6 @@ struct CDFVariable{T, N, V, P} <: AbstractVariable{T, N}
     vdr::V
     parentdataset::P
     dims::NTuple{N, Int}
-    vvrs::Vector{VVREntry}
-    compression::CompressionType
     byte_swap::Bool
 end
 
@@ -30,23 +28,39 @@ function dst_src_ranges(first, last, entry)
     return (dest_first:dest_last, local_first:local_last)
 end
 
-DiskArrays.haschunks(::CDFVariable) = DiskArrays.Chunked()
-function DiskArrays.eachchunk(var::CDFVariable)
+# Codes seem to be faster if we disable chunking
+DiskArrays.haschunks(::CDFVariable) = DiskArrays.Unchunked()
+# DiskArrays.haschunks(::CDFVariable) = DiskArrays.Chunked()
+DiskArrays.eachchunk(var::CDFVariable) = _eachchunk(var)
+
+function _eachchunk(var::CDFVariable)
+    N = ndims(var)
+    chunks = ntuple(N) do i
+        arraysize = var.dims[i]
+        chunksize = max(arraysize, 1) # handle zero-size dimensions
+        DiskArrays.RegularChunks(chunksize, 0, arraysize)
+    end
+    return DiskArrays.GridChunks(chunks)
+end
+
+function _eachchunk_vvrs(var::CDFVariable)
+    vvrs, _ = read_vvrs(var.vdr)
     N = ndims(var)
     chunks = ntuple(N) do i
         if i != N
             DiskArrays.RegularChunks(var.dims[i], 0, var.dims[i])
         else
-            chunksizes = length.(var.vvrs)
-            if length(var.vvrs) > 0
-                chunksizes[end] = @views var.dims[N] - sum(chunksizes[1:end-1])
+            chunksizes = length.(vvrs)
+            if length(vvrs) > 0
+                chunksizes[end] = @views var.dims[N] - sum(chunksizes[1:(end - 1)])
             end
             DiskArrays.IrregularChunks(chunksizes = chunksizes)
         end
     end
     return DiskArrays.GridChunks(chunks)
 end
 
+
 function Base.getproperty(var::CDFVariable, name::Symbol)
     name in fieldnames(CDFVariable) && return getfield(var, name)
     if name == :attrib
@@ -77,4 +91,4 @@ function CPR(var::CDFVariable)
     return CPR(parent(cdf), vdr.cpr_or_spr_offset, recordsize_type(cdf))
 end
 
-is_record_varying(v::CDFVariable) = is_record_varying(v.vdr)
+is_record_varying(v::CDFVariable) = is_record_varying(v.vdr)

test/perf_test.jl

@@ -14,26 +14,29 @@ sum(var)
 full_load(elx_file)
 b0 = @b ds["elb_pef_hs_Epat_eflux"] evals=20
 b1= @b sum(Array(ds["elb_pef_hs_Epat_eflux"])) evals=5
+b12= @b sum(ds["elb_pef_hs_Epat_eflux"]) evals=5
 b2= @b full_load(elx_file)  evals=2
 
 mms_file = data_path(".mms1_scm_srvy_l2_scsrvy_20190301_v2.2.0.cdf")
 ds = CDFDataset(mms_file)
 sum(ds["mms1_scm_acb_gse_scsrvy_srvy_l2"])
+sum(ds["mms1_scm_acb_gse_scsrvy_srvy_l2"][:, 100:100000]) 
 b30 = @b ds["mms1_scm_acb_gse_scsrvy_srvy_l2"] evals=20
 b3= @b sum(Array(ds["mms1_scm_acb_gse_scsrvy_srvy_l2"])) evals=2
 b4= @b sum(ds["mms1_scm_acb_gse_scsrvy_srvy_l2"][:, 100:100000]) evals=5
 b5= @b full_load(mms_file) evals=2
 
-b = [b0, b1, b2, b30, b3, b4, b5]
+b = [b0, b1, b12, b2, b30, b3, b4, b5]
 @info "Benchmarks" b
 
 # ┌ Info: Benchmarks
 # │   b =
-# │    7-element Vector{Chairmarks.Sample}:
-# │     629.150 ns (9 allocs: 784 bytes)
-# │     2.142 μs (25 allocs: 29.328 KiB)
-# │     87.333 μs (3864 allocs: 169.844 KiB)
-# │     2.410 μs (15 allocs: 26.828 KiB)
-# │     9.632 ms (585 allocs: 31.692 MiB)
-# │     474.792 μs (111 allocs: 1.381 MiB)
-# └     273.312 μs (276 allocs: 46.094 KiB)
+# │    8-element Vector{Chairmarks.Sample}:
+# │     539.550 ns (6 allocs: 528 bytes)
+# │     2.083 μs (20 allocs: 29.172 KiB)
+# │     2.000 μs (24 allocs: 29.328 KiB)
+# │     83.896 μs (3777 allocs: 162.469 KiB)
+# │     385.400 ns (7 allocs: 528 bytes)
+# │     9.586 ms (574 allocs: 31.655 MiB)
+# │     467.275 μs (100 allocs: 1.344 MiB)
+# └     20.855 μs (250 allocs: 12.484 KiB)

test/runtests.jl

@@ -57,6 +57,8 @@ end
     @test ds["var_string"] == reference["var_string"]
 
     var = ds["var"]
+    #TODO: find a better and small dataset to really test the chunking
+    @test CommonDataFormat._eachchunk_vvrs(ds["var3d"]) == CommonDataFormat._eachchunk(ds["var3d"])
     @test occursin("compressed", string(var.vdr))
 end
 
@@ -89,4 +91,4 @@ end
     @test string(TT2000(0)) == "2000-01-01T11:58:55.816"
     @test TT2000(0) == TT2000(0) |> bswap
     @test TT2000(0) == DateTime("2000-01-01T11:58:55.816")
-end
+end