Add an Ordered Dict Row Table Blob Store (#1075)

* Add a Ordered Dict Row Table Blob Store

* add RowBlobStore constructors for tables

* FolderStore label as kwarg

Author: Affie

Project.toml

@@ -25,6 +25,7 @@ SHA = "ea8e919c-243c-51af-8825-aaa63cd721ce"
 SparseArrays = "2f01184e-e22b-5df5-ae63-d93ebab69eaf"
 StaticArrays = "90137ffa-7385-5640-81b9-e52037218182"
 StructTypes = "856f2bd8-1eba-4b0a-8007-ebc267875bd4"
+Tables = "bd369af6-aec1-5ad0-b16a-f7cc5008161c"
 Tar = "a4e569a6-e804-4fa4-b0f3-eef7a1d5b13e"
 TensorCast = "02d47bb6-7ce6-556a-be16-bb1710789e2b"
 TimeZones = "f269a46b-ccf7-5d73-abea-4c690281aa53"
@@ -52,8 +53,8 @@ Graphs = "1.4"
 InteractiveUtils = "1.10"
 JSON3 = "1"
 LinearAlgebra = "1.10"
-ManifoldsBase = "0.14, 0.15"
 Manifolds = "0.9"
+ManifoldsBase = "0.14, 0.15"
 OrderedCollections = "1.4"
 Pkg = "1.4, 1.5"
 ProgressMeter = "1"
@@ -64,9 +65,10 @@ SparseArrays = "1.10"
 StaticArrays = "1"
 Statistics = "1.10"
 StructTypes = "1"
+Tables = "v1.11.1"
 Tar = "1.9"
-Test = "1.10"
 TensorCast = "0.3.3, 0.4"
+Test = "1.10"
 TimeZones = "1.3.1"
 UUIDs = "1.10"
 julia = "1.10"

src/DataBlobs/services/BlobEntry.jl

@@ -130,6 +130,16 @@ function getBlobEntryFirst(dfg::AbstractDFG, label::Symbol, key::Regex)
     return getBlobEntryFirst(getVariable(dfg, label), key)
 end
 
+# TODO Consider autogenerating all methods of the form:
+# verbNoun(dfg::DFGVariable, label::Symbol, args...; kwargs...) = verbNoun(getVariable(dfg, label), args...; kwargs...)
+# with something like:
+# getvariablemethod = [
+#     :getBlobEntryFirst,
+# ]
+# for met in methodstooverload  
+#     @eval DistributedFactorGraphs $met(dfg::AbstractDFG, label::Symbol, args...; kwargs...) = $met(getVariable(dfg, label), args...; kwargs...)
+# end
+
 function getBlobEntry(dfg::AbstractDFG, label::Symbol, key::Union{Symbol, UUID})
     return getBlobEntry(getVariable(dfg, label), key)
 end

src/DataBlobs/services/BlobStores.jl

@@ -181,13 +181,13 @@ struct FolderStore{T} <: AbstractBlobStore{T}
     folder::String
 end
 
-function FolderStore(foldername::String; createfolder = true)
+function FolderStore(foldername::String; label = :default_folder_store, createfolder = true)
     if createfolder && !isdir(foldername)
         @info "Folder '$foldername' doesn't exist - creating."
         # create new folder
         mkpath(foldername)
     end
-    return FolderStore{Vector{UInt8}}(:default_folder_store, foldername)
+    return FolderStore{Vector{UInt8}}(label, foldername)
 end
 
 blobfilename(store::FolderStore, blobId::UUID) = joinpath(store.folder, string(blobId))
@@ -254,7 +254,7 @@ struct InMemoryBlobStore{T} <: AbstractBlobStore{T}
 end
 
 function InMemoryBlobStore{T}(storeKey::Symbol) where {T}
-    return InMemoryBlobStore{Vector{UInt8}}(storeKey, Dict{UUID, T}())
+    return InMemoryBlobStore{T}(storeKey, Dict{UUID, T}())
 end
 function InMemoryBlobStore(storeKey::Symbol = :default_inmemory_store)
     return InMemoryBlobStore{Vector{UInt8}}(storeKey)
@@ -338,3 +338,128 @@ end
 
 deleteBlob!(store::LinkStore, ::BlobEntry) = deleteBlob!(store)
 deleteBlob!(store::LinkStore, ::UUID) = deleteBlob!(store)
+
+##==============================================================================
+## RowBlobStore Ordered Dict Row Table Blob Store
+##==============================================================================
+
+# RowBlob
+# T must be compatable with the AbstactRow iterator
+# struct and named tuple as examples
+struct RowBlob{T} <: Tables.AbstractRow
+    id::UUID
+    blob::T
+end
+
+function RowBlob(::Type{T}, nt::NamedTuple) where {T}
+    id = nt.id
+    blob = T(nt[keys(nt)[2:end]])
+    return RowBlob(id, blob)
+end
+
+function Tables.getcolumn(row::RowBlob, i::Int)
+    return i == 1 ? getfield(row, :id) : Tables.getcolumn(getfield(row, :blob), i - 1)
+end
+function Tables.getcolumn(row::RowBlob, nm::Symbol)
+    return nm == :id ? getfield(row, :id) : Tables.getcolumn(getfield(row, :blob), nm)
+end
+function Tables.columnnames(row::RowBlob)
+    return (:id, Tables.columnnames(getfield(row, :blob))...)
+end
+
+## RowBlobStore
+
+struct RowBlobStore{T} <: AbstractBlobStore{T}
+    key::Symbol
+    blobs::OrderedDict{UUID, RowBlob{T}}
+end
+
+function RowBlobStore{T}(storeKey::Symbol) where {T}
+    return RowBlobStore{T}(storeKey, OrderedDict{UUID, RowBlob{T}}())
+end
+function RowBlobStore(storeKey::Symbol, T::DataType)
+    return RowBlobStore{T}(storeKey)
+end
+
+function RowBlobStore(storeKey::Symbol, T::DataType, table)
+    store = RowBlobStore(storeKey, T)
+    for nt in Tables.namedtupleiterator(table)
+        row = DFG.RowBlob(T, nt)
+        store.blobs[row.id] = row
+    end
+    return store
+end
+
+# Tables interface
+Tables.istable(::Type{RowBlobStore{T}}) where {T} = true
+Tables.rowaccess(::Type{RowBlobStore{T}}) where {T} = true
+Tables.rows(store::RowBlobStore) = values(store.blobs)
+#TODO
+# Tables.materializer(::Type{RowBlobStore{T}}) where T = Tables.rowtable
+
+##
+function getBlob(store::RowBlobStore, blobId::UUID)
+    return getfield(store.blobs[blobId], :blob)
+end
+
+function addBlob!(store::RowBlobStore{T}, blobId::UUID, blob::T) where {T}
+    if haskey(store.blobs, blobId)
+        error("Key '$blobId' blob already exists.")
+    end
+    store.blobs[blobId] = RowBlob(blobId, blob)
+    return blobId
+end
+
+function updateBlob!(store::RowBlobStore{T}, blobId::UUID, blob::T) where {T}
+    if haskey(store.blobs, blobId)
+        @warn "Key '$blobId' doesn't exist."
+    end
+    return store.blobs[blobId] = RowBlob(blobId, blob)
+end
+
+function deleteBlob!(store::RowBlobStore, blobId::UUID)
+    return getfield(pop!(store.blobs, blobId), :blob)
+end
+
+hasBlob(store::RowBlobStore, blobId::UUID) = haskey(store.blobs, blobId)
+
+listBlobs(store::RowBlobStore) = collect(keys(store.blobs))
+
+# TODO also see about wrapping a table directly
+##
+if false
+    rb = RowBlob(uuid4(), (a = [1, 2], b = [3, 4]))
+
+    Tables.columnnames(rb)
+
+    tstore = RowBlobStore(:namedtuple, @NamedTuple{a::Vector{Int}, b::Vector{Int}})
+
+    addBlob!(tstore, uuid4(), (a = [1, 2], b = [3, 4]))
+    addBlob!(tstore, uuid4(), (a = [5, 6], b = [7, 8]))
+    addBlob!(tstore, uuid4(), (a = [9, 10], b = [11, 12]))
+
+    rowtbl = Tables.rowtable(tstore)
+    coltbl = Tables.columntable(rowtbl)
+
+    Tables.rows(tstore)
+    tbl = Tables.rowtable(tstore)
+
+    first(Tables.namedtupleiterator(tstore))
+
+    # Tables.materializer(tstore)
+
+    ##
+    struct Foo
+        a::Float64
+        b::Float64
+    end
+
+    sstore = RowBlobStore(:struct_Foo, Foo)
+
+    addBlob!(sstore, uuid4(), Foo(1, 2))
+    addBlob!(sstore, uuid4(), Foo(3, 4))
+    addBlob!(sstore, uuid4(), Foo(5, 6))
+
+    Tables.rowtable(sstore)
+end
+##

src/DataBlobs/services/HelpersDataWrapEntryBlob.jl

@@ -122,6 +122,7 @@ function getData(
     return de => db
 end
 
+#FIXME Should `addData!`` not return entry=>blob pair?
 function addData!(
     dfg::AbstractDFG,
     label::Symbol,
@@ -171,7 +172,6 @@ function addData!(
     )
 end
 
-#FIXME id used wrong
 function addData!(
     dfg::AbstractDFG,
     blobstore::AbstractBlobStore,
@@ -182,14 +182,15 @@ function addData!(
     description = "",
     metadata = "",
     mimeType::String = "application/octet-stream",
-    id::Union{UUID, Nothing} = nothing, #only assign if blobstore issued you an id
+    id::Union{UUID, Nothing} = nothing,
+    blobId::Union{UUID, Nothing} = nothing, #only assign if blobstore issued you an id
     originId::UUID = uuid4(),
     hashfunction = sha256,
 )
     #
-    @warn "ID's and origin IDs should be reconciled here in DFG.addData!." maxlog = 50
     entry = BlobEntry(;
         id,
+        blobId,
         originId,
         label = bLbl,
         blobstore = blobstore.key,
@@ -204,6 +205,40 @@ function addData!(
     return addData!(dfg, blobstore, vLbl, entry, blob; hashfunction)
 end
 
+function addData!(
+    dfg::AbstractDFG,
+    blobstore::AbstractBlobStore{T},
+    vLbl::Symbol,
+    blobLabel::Symbol,
+    blob::T,
+    timestamp = now(localzone());
+    description = "",
+    metadata = "",
+    mimeType::String = "application/octet-stream",
+    origin = buildSourceString(dfg, vLbl),
+    # hashfunction = sha256,
+) where {T}
+    #
+    # checkhash && assertHash(entry, blob; hashfunction)
+    blobId = addBlob!(blobstore, blob)
+
+    entry = BlobEntry(;
+        blobId,
+        originId = blobId,
+        label = blobLabel,
+        blobstore = blobstore.key,
+        # hash = string(bytes2hex(hashfunction(blob))),
+        hash = "",
+        origin,
+        description,
+        mimeType,
+        metadata,
+        timestamp,
+    )
+    addBlobEntry!(dfg, vLbl, entry)
+    return entry => blob
+end
+
 function updateData!(
     dfg::AbstractDFG,
     label::Symbol,

src/DistributedFactorGraphs.jl

@@ -38,6 +38,7 @@ using OrderedCollections
 export OrderedDict
 
 using CSV
+using Tables
 
 # used for @defVariable
 import ManifoldsBase

src/services/AbstractDFG.jl

@@ -1302,7 +1302,7 @@ function getNeighborhood(
     solvable::Int = 0,
 )
     # find neighbors at distance to add
-    neighbors = Symbol[]
+    neighbors = Set{Symbol}()
     if distance > 0
         for l in variableFactorLabels
             union!(neighbors, getNeighborhood(dfg, l, distance))