In-memory and file datastores

Author: GearsAD

src/BigData/BigData.jl

@@ -1,5 +1,5 @@
 include("entities/AbstractDataStore.jl")
-include("entities/DataEntries.jl")
+include("entities/AbstractBigDataEntries.jl")
 include("entities/InMemoryDataStore.jl")
 include("entities/FileDataStore.jl")
 

src/BigData/entities/DataEntries.jl renamed to src/BigData/entities/AbstractBigDataEntries.jl

@@ -7,7 +7,7 @@ reproducible key using userId_robotId_sessionId_variableId_key.
 """
 mutable struct GeneralBigDataEntry <: AbstractBigDataEntry
     key::Symbol
-    originalKey::Symbol
+    storeKey::Symbol # Could swap this to string, but using it as an index later, it's better as a symbol I believe.
     createdTimestamp::DateTime
     lastUpdatedTimestamp::DateTime
     mimeType::String
@@ -16,7 +16,7 @@ end
 ## TODO: Move this
 function ==(a::GeneralBigDataEntry, b::GeneralBigDataEntry)
     return a.key == b.key &&
-        a.originalKey == b.originalKey &&
+        a.storeKey == b.storeKey &&
         a.mimeType == b.mimeType &&
         Dates.value(a.createdTimestamp - b.createdTimestamp) < 1000 &&
         Dates.value(a.lastUpdatedTimestamp - b.lastUpdatedTimestamp) < 1000 #1 second
@@ -27,24 +27,24 @@ end
 """
     $(SIGNATURES)
 Generates a unique key for the entry - userId_robotId_sessionId_variable_key.
-Simple.
+Simple symbol.
 """
-function uniqueKey(dfg::G, v::V, key::Symbol) where {G <: AbstractDFG, V <: AbstractDFGVariable}
+function uniqueKey(dfg::G, v::V, key::Symbol)::Symbol where {G <: AbstractDFG, V <: AbstractDFGVariable}
     key = join(String.([dfg.userId, dfg.robotId, dfg.sessionId, label(v), String(key)]), "_")
     return Symbol(key)
 end
 
 GeneralBigDataEntry(key::Symbol,
-                    originalKey::Symbol;
+                    storeKey::Symbol;
                     mimeType::String="application/octet-stream") =
-                        GeneralBigDataEntry(key, originalKey, now(), now(), mimeType)
+                        GeneralBigDataEntry(key, storeKey, now(), now(), mimeType)
 
 function GeneralBigDataEntry(dfg::G,
                              var::V,
-                             originalKey::Symbol;
+                             key::Symbol;
                              mimeType::String="application/octet-stream") where {G <: AbstractDFG, V <: AbstractDFGVariable}
-    return GeneralBigDataEntry(uniqueKey(dfg, var, originalKey),
-                               originalKey,
+    return GeneralBigDataEntry(key,
+                               uniqueKey(dfg, var, key),
                                mimeType=mimeType)
 end
 

src/BigData/services/FileDataStore.jl

@@ -1,5 +1,5 @@
 searchdir(path,key) = filter(x->occursin(key,x), readdir(path))
-filename(store::FileDataStore, entry::E) where E <: AbstractBigDataEntry = "$(store.folder)/$(entry.key).dat"
+filename(store::FileDataStore, entry::E) where E <: AbstractBigDataEntry = "$(store.folder)/$(entry.storeKey).dat"
 function readentry(store::FileDataStore, entry::E) where E <: AbstractBigDataEntry
     open(filename(store, entry)) do f
         return read(f)
@@ -12,32 +12,34 @@ function writeentry(store::FileDataStore, entry::E, data::Vector{UInt8}) where E
 end
 
 function getBigData(store::FileDataStore, entry::E)::Union{Vector{UInt8}, Nothing} where {E <: AbstractBigDataEntry}
-    length(searchdir(store.folder, String(entry.key))) !=1 && (@warn "Could not find unique file for key '$(entry.key)'."; return nothing)
+    length(searchdir(store.folder, String(entry.storeKey)*".dat")) !=1 && (@warn "Could not find unique file for key '$(entry.storeKey)'."; return nothing)
     return readentry(store, entry)
 end
 
 function addBigData!(store::FileDataStore, entry::E, data::Vector{UInt8})::Vector{UInt8} where {E <: AbstractBigDataEntry}
-    length(searchdir(store.folder, String(entry.key))) !=0 && @warn "Key '$(entry.key)' already exists, overwriting!."
+    length(searchdir(store.folder, String(entry.storeKey)*".dat")) !=0 && @warn "Key '$(entry.storeKey)' already exists, overwriting!."
     writeentry(store, entry, data)
     # Update timestamp
     entry.lastUpdatedTimestamp = now()
     return getBigData(store, entry)
 end
 
 function updateBigData!(store::FileDataStore, entry::E, data::Vector{UInt8})::Union{Vector{UInt8}, Nothing} where {E <: AbstractBigDataEntry}
-    length(searchdir(store.folder, String(entry.key))) !=1 && (@warn "Could not find unique file for key '$(entry.key)'."; return nothing)
+    length(searchdir(store.folder, String(entry.storeKey)*".dat")) !=1 && (@warn "Could not find unique file for key '$(entry.storeKey)'."; return nothing)
     writeentry(store, entry, data)
     # Update timestamp
     entry.lastUpdatedTimestamp = now()
     return getBigData(store, entry)
 end
 
 function deleteBigData!(store::FileDataStore, entry::E)::Vector{UInt8} where {E <: AbstractBigDataEntry}
-    data = getData(store, v, entry.key)
+    data = getBigData(store, entry)
     data == nothing && return nothing
     rm(filename(store, entry))
     return data
 end
+
+# TODO: Manifest file
 #
 # function listStoreEntries(store::FileDataStore)::Vector{E} where {E <: AbstractBigDataEntry}
 #     return collect(values(store.entries))

src/BigData/services/InMemoryDataStore.jl

@@ -1,29 +1,29 @@
 function getBigData(store::InMemoryDataStore{T, E}, entry::E)::Union{T, Nothing} where {T, E <: AbstractBigDataEntry}
-    !haskey(store.data, entry.key) && return nothing
-    return store.data[entry.key]
+    !haskey(store.data, entry.storeKey) && return nothing
+    return store.data[entry.storeKey]
 end
 
 function addBigData!(store::InMemoryDataStore{T, E}, entry::E, data::T)::T where {T, E <: AbstractBigDataEntry}
-    haskey(store.entries, entry.key) && @warn "Key '$(entry.key)' already exists in the data store, overwriting!"
-    store.entries[entry.key] = entry
+    haskey(store.entries, entry.storeKey) && @warn "Key '$(entry.storeKey)' already exists in the data store, overwriting!"
+    store.entries[entry.storeKey] = entry
     # Update timestamp
     entry.lastUpdatedTimestamp = now()
-    return store.data[entry.key] = data
+    return store.data[entry.storeKey] = data
 end
 
 function updateBigData!(store::InMemoryDataStore{T, E}, entry::E, data::T)::Union{T, Nothing} where {T, E <: AbstractBigDataEntry}
-    !haskey(store.entries, entry.key) && (@error "Key '$(entry.key)' doesn't exist in the data store!"; return nothing)
-    store.entries[entry.key] = entry
+    !haskey(store.entries, entry.storeKey) && (@error "Key '$(entry.storeKey)' doesn't exist in the data store!"; return nothing)
+    store.entries[entry.storeKey] = entry
     # Update timestamp
     entry.lastUpdatedTimestamp = now()
-    return store.data[entry.key] = data
+    return store.data[entry.storeKey] = data
 end
 
 function deleteBigData!(store::InMemoryDataStore{T, E}, entry::E)::T where {T, E <: AbstractBigDataEntry}
     data = getBigData(store, entry)
     data == nothing && return nothing
-    delete!(store.data, entry.key)
-    delete!(store.entries, entry.key)
+    delete!(store.data, entry.storeKey)
+    delete!(store.entries, entry.storeKey)
     return data
 end
 

src/BigData/services/MongoDataStore.jl

@@ -1,4 +1,8 @@
 
+"""
+    @(SIGNATURES)
+Save a DFG to a folder. Will create/overwrite folder if it exists.
+"""
 function saveDFG(dfg::G, folder::String) where G <: AbstractDFG
     variables = getVariables(dfg)
     factors = getFactors(dfg)
@@ -30,6 +34,19 @@ function saveDFG(dfg::G, folder::String) where G <: AbstractDFG
     end
 end
 
+"""
+    @(SIGNATURES)
+Load a DFG from a saved folder. You will need to provide the IIFModule as the second
+parameter.
+
+Example:
+using DistributedFactorGraphs, IncrementalInference
+# Create a DFG - can make one directly, e.g. GraphsDFG{NoSolverParams}() or use IIF:
+dfg = initfg()
+# Load the graph
+loadDFG("/tmp/savedgraph", IncrementalInference, dfg)
+# Use it as normal.
+"""
 function loadDFG(folder::String, iifModule, dfgLoadInto::G) where G <: AbstractDFG
     variables = DFGVariable[]
     factors = DFGFactor[]

src/FileDFG/services/FileDFG.jl

@@ -171,7 +171,7 @@ end
 
 """
     $(SIGNATURES)
-Add Big Data Entry to distrubuted factor graph.
+Add Big Data Entry to distributed factor graph.
 Should be extended if DFG variable is not returned by reference.
 """
 function addBigDataEntry!(dfg::AbstractDFG, label::Symbol, bde::AbstractBigDataEntry)::AbstractDFGVariable
@@ -194,27 +194,34 @@ end
     $(SIGNATURES)
 Update big data entry
 """
-function updateBigDataEntry!(var::AbstractDFGVariable,  bde::AbstractBigDataEntry)::AbstractDFGVariable
-    !haskey(var.bigData,bde.key) && (@error "$(bde.key) does not exist in variable!"; return false)
+function updateBigDataEntry!(var::AbstractDFGVariable,  bde::AbstractBigDataEntry)::Union{Nothing, AbstractDFGVariable}
+    !haskey(var.bigData,bde.key) && (@error "$(bde.key) does not exist in variable!"; return nothing)
     var.bigData[bde.key] = bde
     return var
 end
-function updateBigDataEntry!(dfg::AbstractDFG, label::Symbol,  bde::AbstractBigDataEntry)::AbstractDFGVariable
+function updateBigDataEntry!(dfg::AbstractDFG, label::Symbol,  bde::AbstractBigDataEntry)::Union{Nothing, AbstractDFGVariable}
+    !isVariable(dfg, label) && return nothing
     return updateBigDataEntry!(getVariable(dfg, label), bde)
 end
 
 """
     $(SIGNATURES)
-Delete big data entry
+Delete big data entry from the factor graph.
+Note this doesn't remove it from any data stores.
 """
-function deleteBigDataEntry!(var::AbstractDFGVariable, key::Symbol)::AbstractBigDataEntry #users responsibility to delete big data in db before deleting entry
+function deleteBigDataEntry!(var::AbstractDFGVariable, key::Symbol)::Union{Nothing, AbstractDFGVariable} #users responsibility to delete big data in db before deleting entry
     bde = getBigDataEntry(var, key)
+    bde == nothing && return nothing
     delete!(var.bigData, key)
-    return bde
+    return var
+end
+function deleteBigDataEntry!(dfg::AbstractDFG, label::Symbol, key::Symbol)::Union{Nothing, AbstractDFGVariable} #users responsibility to delete big data in db before deleting entry
+    !isVariable(dfg, label) && return nothing
+    return deleteBigDataEntry!(getVariable(dfg, label), key)
 end
 
-function deleteBigDataEntry!(dfg::AbstractDFG, label::Symbol, key::Symbol)::AbstractBigDataEntry #users responsibility to delete big data in db before deleting entry
-    deleteBigDataEntry!(getVariable(dfg, label), key)
+function deleteBigDataEntry!(var::AbstractDFGVariable, entry::AbstractBigDataEntry)::Union{Nothing, AbstractDFGVariable} #users responsibility to delete big data in db before deleting entry
+    return deleteBigDataEntry!(var, entry.key)
 end
 
 """
@@ -225,7 +232,8 @@ function getBigDataEntries(var::AbstractDFGVariable)::Vector{AbstractBigDataEntr
     #or should we return the iterator, Base.ValueIterator{Dict{Symbol,AbstractBigDataEntry}}?
     collect(values(var.bigData))
 end
-function getBigDataEntries(dfg::AbstractDFG, label::Symbol)::Vector{AbstractBigDataEntry}
+function getBigDataEntries(dfg::AbstractDFG, label::Symbol)::Union{Nothing, Vector{AbstractBigDataEntry}}
+    !isVariable(dfg, label) && return nothing
     #or should we return the iterator, Base.ValueIterator{Dict{Symbol,AbstractBigDataEntry}}?
     getBigDataEntries(getVariable(dfg, label))
 end
@@ -238,6 +246,7 @@ getBigDataKeys
 function getBigDataKeys(var::AbstractDFGVariable)::Vector{Symbol}
     collect(keys(var.bigData))
 end
-function getBigDataKeys(dfg::AbstractDFG, label::Symbol)::Vector{Symbol}
+function getBigDataKeys(dfg::AbstractDFG, label::Symbol)::Union{Nothing, Vector{Symbol}}
+    !isVariable(dfg, label) && return nothing
     getBigDataKeys(getVariable(dfg, label))
 end

src/services/DFGVariable.jl

@@ -0,0 +1,51 @@
+using DistributedFactorGraphs
+using Test
+
+# Build a basic graph.
+
+dfg = GraphsDFG{NoSolverParams}()
+#add types for softtypes
+struct TestInferenceVariable1 <: InferenceVariable end
+v1 = DFGVariable(:a, TestInferenceVariable1())
+v2 = DFGVariable(:b, TestInferenceVariable1())
+f1 = DFGFactor{Int, :Symbol}(:f1)
+#add tags for filters
+append!(v1.tags, [:VARIABLE, :POSE])
+append!(v2.tags, [:VARIABLE, :LANDMARK])
+append!(f1.tags, [:FACTOR])
+# @testset "Creating Graphs" begin
+addVariable!(dfg, v1)
+addVariable!(dfg, v2)
+addFactor!(dfg, [v1, v2], f1)
+
+# Stores to test
+testStores = [InMemoryDataStore(), FileDataStore("/tmp/dfgFilestore")]
+
+for testStore in testStores
+    # Create a data store and a dataset
+    ds = testStore
+    # ds = FileDataStore("/tmp/filestore")
+    dataset = rand(UInt8, 1000)
+    dataset2 = rand(UInt8, 1000)
+
+    entry1 = GeneralBigDataEntry(dfg, v1, :test1)
+    # Set it in the store
+    @test addBigData!(ds, entry1, dataset) == dataset
+    @test getBigData(ds, entry1) == dataset
+    # Now add it to the variable
+    @test addBigDataEntry!(v1, entry1) == v1
+    @test entry1 in getBigDataEntries(v1)
+    # Update test
+    copyEntry = deepcopy(entry1)
+    @test updateBigData!(ds, entry1, dataset2) == dataset2
+    # Data updated?
+    @test getBigData(ds, entry1) == dataset2
+    # Timestamp updated?
+    @test entry1.lastUpdatedTimestamp > copyEntry.lastUpdatedTimestamp
+    # Delete data
+    @test deleteBigData!(ds, entry1) == dataset2
+    # Delete entry
+    @test deleteBigDataEntry!(v1, entry1) == v1
+    #TODO
+    #listStoreEntries(ds)
+end

test/DataStoreTests.jl

@@ -18,6 +18,13 @@
     updateVariable!(dfg, verts[7])
     updateVariable!(dfg, verts[8])
 
+    # Add some bigData to x1, x2
+    addBigDataEntry!(verts[1], GeneralBigDataEntry(:testing, :testing; mimeType="application/nuthin!"))
+    addBigDataEntry!(verts[2], FileBigDataEntry(:testing2, "/dev/null"))
+    #call update to set it on cloud
+    updateVariable!(dfg, verts[1])
+    updateVariable!(dfg, verts[2])
+
     facts = map(n -> addFactor!(dfg, [verts[n], verts[n+1]], LinearConditional(Normal(50.0,2.0))), 1:(numNodes-1))
 
     # Save and load the graph to test.
@@ -26,6 +33,7 @@
 
     copyDfg = DistributedFactorGraphs._getDuplicatedEmptyDFG(dfg)
     retDFG = loadDFG(saveFolder, Main, copyDfg)
+
     @test symdiff(ls(dfg), ls(retDFG)) == []
     @test symdiff(lsf(dfg), lsf(retDFG)) == []
     for var in ls(dfg)
@@ -34,4 +42,10 @@
     for fact in lsf(dfg)
         @test getFactor(dfg, fact) == getFactor(retDFG, fact)
     end
+
+    @test length(getBigDataEntries(getVariable(retDFG, :x1))) == 1
+    @test typeof(getBigDataEntry(getVariable(retDFG, :x1),:testing)) == GeneralBigDataEntry
+    @test length(getBigDataEntries(getVariable(retDFG, :x2))) == 1
+    @test typeof(getBigDataEntry(getVariable(retDFG, :x2),:testing2)) == FileBigDataEntry
+
 end

test/fileDFGTests.jl

@@ -209,20 +209,21 @@ end
 
         #add
         v1 = getVariable(dfg, :a)
-        @test addBigDataEntry!(v1, de1)
-        @test addBigDataEntry!(dfg, :a, de2)
-        @test addBigDataEntry!(v1, de1)
+        @test addBigDataEntry!(v1, de1) == v1
+        @test addBigDataEntry!(dfg, :a, de2) == v1
+        @test addBigDataEntry!(v1, de1) == v1
+        @test de2 in getBigDataEntries(v1)
 
         #get
         @test deepcopy(de1) == getBigDataEntry(v1, :key1)
         @test deepcopy(de2) == getBigDataEntry(dfg, :a, :key2)
-        @test_throws Any getBigDataEntry(v2, :key1)
-        @test_throws Any getBigDataEntry(dfg, :b, :key1)
+        @test getBigDataEntry(v2, :key1) == nothing
+        @test getBigDataEntry(dfg, :b, :key1) == nothing
 
         #update
-        @test updateBigDataEntry!(dfg, :a, de2_update)
+        @test updateBigDataEntry!(dfg, :a, de2_update) == v1
         @test deepcopy(de2_update) == getBigDataEntry(dfg, :a, :key2)
-        @test !updateBigDataEntry!(dfg, :b, de2_update)
+        @test updateBigDataEntry!(dfg, :b, de2_update) == nothing
 
         #list
         entries = getBigDataEntries(dfg, :a)
@@ -234,10 +235,10 @@ end
         @test getBigDataKeys(dfg, :b) == Symbol[]
 
         #delete
-        @test deepcopy(de1) == deleteBigDataEntry!(v1, :key1)
+        @test deleteBigDataEntry!(v1, :key1) == v1
         @test getBigDataKeys(v1) == Symbol[:key2]
         #delete from dfg
-        @test deepcopy(de2_update) == deleteBigDataEntry!(dfg, :a, :key2)
+        @test deleteBigDataEntry!(dfg, :a, :key2) == v1
         @test getBigDataKeys(v1) == Symbol[]
     end
 end