DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Initial CRS implementation for RDBMS

Browse Source
This commit is contained in:
Denis Arh
2022-04-23 16:21:55 +02:00
committed by Tomaž Jerman
parent cda884dff7
commit 2d06cf1cf9
12 changed files with 1448 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
compose/crs/driver.go
View File

@@ -47,12 +47,19 @@ type (
// DDL
// Models returns all of the collections the given database already defines
Models(context.Context) (data.ModelSet, error)
// returns all attribute types that driver supports
AttributeTypes() []data.AttributeType
// AddModel requests the driver to support the specified collections
AddModel(context.Context, *data.Model, ...*data.Model) error
// // RemoveModel requests the driver to remove support for the specified collections
// RemoveModel(context.Context, *data.Model, ...*data.Model) error
// AlterModel requests the driver to alter the general collectio parameters
AlterModel(ctx context.Context, old *data.Model, new *data.Model) error
// AlterModelAttribute requests the driver to alter the specified attribute of the given collection
AlterModelAttribute(ctx context.Context, sch *data.Model, old data.Attribute, new data.Attribute, trans ...func(*data.Model, data.Attribute, expr.TypedValue) (expr.TypedValue, bool, error)) error
}

2
compose/crs/model.go
View File

@@ -208,7 +208,7 @@ func (crs *composeRecordStore) loadModules(ctx context.Context, cs cStore) (mm t
return
}
func moduleFieldCodec(f *types.ModuleField) (strat data.StoreStrategy) {
func moduleFieldCodec(f *types.ModuleField) (strat data.StoreCodec) {
// Defaulting to alias
strat = data.StoreCodecAlias{
Ident: f.Name,

22
pkg/data/model.go
View File

@@ -28,7 +28,7 @@ type (
// Store describes the strategy the underlying storage system should
// apply to the underlying value
Store StoreStrategy
Store StoreCodec
// Type describes what the value represents and how it should be
// encoded/decoded
Type attributeType
@@ -215,6 +215,26 @@ func (m Model) Validate() error {
return nil
}
//// AttributeGroups returns attributes grouped by ident
//func (m Model) AttributeGroups() (gg map[string][]*Attribute) {
// gg = make(map[string][]*Attribute)
//
// for _, attr := range m.Attributes {
// // @todo properly check attribute integrity:
// // is embeddable
// // only same types use the same ident
// if gIdent, embeddable := attr.Store.Embeddable(); !embeddable && len(gg[attr.Ident]) > 0 {
// panic("attribute " + attr.Ident + " is not embeddable")
// } else if len(gIdent) > 0 {
// gg[gIdent] = append(gg[gIdent], attr)
// } else {
// gg[attr.Ident] = append(gg[attr.Ident], attr)
// }
// }
//
// return
//}
// Receivers to conform to the interface
func (t TypeID) Type() AttributeType {

53
pkg/data/store_codec.go
View File

@@ -1,22 +1,36 @@
package data
type (
// StoreStrategy defines how values for a specific model attribute
StoreCodecType string
// StoreCodec defines how values for a specific model attribute
// are retrieved or stored
//
// If attribute does not have store strategy set
// store driver should use attribute name to determinate
// source/destination of the value (table column, json doc property)
StoreStrategy any
// StoreCodecJSON defines that values are encoded/decoded into
// a JSON document (stored under ident)
// Path defines exact location inside the doc.
StoreCodecJSON struct {
Ident string
Path []any
// source/destination of the value (table column, json document property)
StoreCodec interface {
Type() StoreCodecType
}
StoreCodecPlain struct{}
// StoreCodecStdRecordValueJSON defines that values are encoded/decoded into
// a JSON simple document { [_: string]: Array<unknown> }
//
// This only solves
// Attribute{Ident: "foo", Store: StoreCodecRecordValueJSON{ Ident: "bar" }
// => "bar"->'foo'->0
StoreCodecStdRecordValueJSON struct {
Ident string
}
// handling complex JSON documents
//StoreCodecJSON struct {
// Ident string
// Path []any
//}
//
// { "@value": ... "@type": .... }
// StoreCodecJSONLD struct { Ident string; Path []any }
@@ -26,8 +40,25 @@ type (
// StoreCodecAlias defines case when value is not stored
// under the same column (in case of an SQL table)
//
// Example: attribute ident is "foo" and database table column is "bar"
// Value of StoreCodecAlias.Ident is used as base
// and value of Attribute.Ident holding StoreCodecAlias is used
// as an alias!
//
// Attribute{Ident: "foo", Store: StoreCodecAlias{ Ident: "bar" }
// => "bar" as "foo"
StoreCodecAlias struct {
Ident string
}
)
const (
StoreCodecPlainType StoreCodecType = "corteza::data-store-codec:plain"
StoreCodecStdRecordValueJSONType StoreCodecType = "corteza::data-store-codec:record-store-json"
StoreCodecAliasType StoreCodecType = "corteza::data-store-codec:alias"
)
func (t StoreCodecType) Is(c StoreCodec) bool { return t == c.Type() }
func (StoreCodecPlain) Type() StoreCodecType { return StoreCodecPlainType }
func (StoreCodecStdRecordValueJSON) Type() StoreCodecType { return StoreCodecStdRecordValueJSONType }
func (StoreCodecAlias) Type() StoreCodecType { return StoreCodecAliasType }

26
pkg/slice/strings.go
View File

@@ -1,5 +1,30 @@
package slice
func ContainsAny[C comparable](haystack []C, needles ...C) bool {
for _, h := range haystack {
for _, n := range needles {
if h == n {
return true
}
}
}
return false
}
func ContainsAll[C comparable](haystack []C, needles ...C) bool {
var matches = 0
for _, h := range haystack {
for _, n := range needles {
if h == n {
matches++
}
}
}
return len(needles) == matches
}
func IntersectStrings(a []string, b []string) []string {
var (
out = make([]string, 0, len(a)+len(b))
@@ -33,6 +58,7 @@ func ToUint64BoolMap(u []uint64) (h map[uint64]bool) {
return
}
// @todo replace with ContainsAny
func HasString(ss []string, s string) bool {
for i := range ss {
if ss[i] == s {

203
store/adapters/rdbms/crs/decoder.go Normal file
View File

@@ -0,0 +1,203 @@
package crs
import (
"database/sql"
"encoding/json"
"fmt"
"strconv"
"time"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/data"
"github.com/google/uuid"
"github.com/spf13/cast"
)
// decodes values into record by iterating over all columns
// and running decode handler on all
func decode(columns []*column, values []any, r *types.Record) (err error) {
for i, c := range columns {
if err = c.decode(c.attributes, values[i], r); err != nil {
return
}
}
return nil
}
func decodeStdRecordValueJSON(aa []*data.Attribute, value any, r *types.Record) (err error) {
rawJson, is := value.(*sql.RawBytes)
if !is {
return fmt.Errorf("incompatible input value type (%T), expecting *sql.RawBytes", value)
}
buf := make(map[string][]any)
if err = json.Unmarshal(*rawJson, &buf); err != nil {
return
}
rvs := make([]*types.RecordValue, 0, len(buf))
// @todo this is too naive
for f, vv := range buf {
if isSystemField(f) {
if err = decodeSystemField(f, value, r); err != nil {
return
}
} else {
for i, v := range vv {
rvs = append(rvs, &types.RecordValue{Name: f, Value: cast.ToString(v), Place: uint(i)})
}
}
}
r.Values = rvs
return
}
// decode non-serialized record value
func decodeRecordValue(aa []*data.Attribute, raw any, r *types.Record) (err error) {
if len(aa) == 0 {
return fmt.Errorf("can not decode value from column without attributes")
}
if len(aa) > 1 {
return fmt.Errorf("can not decode value from a non-embedded column with more than one attribute")
}
var (
rv = &types.RecordValue{Name: aa[0].Ident}
)
switch aux := raw.(type) {
case *uint64:
if aux != nil {
rv.Ref = *aux
rv.Value = strconv.FormatUint(*aux, 10)
}
case *sql.NullTime:
if aux == nil || !aux.Valid {
return
}
rv.Value = aux.Time.Format(time.RFC3339)
case *sql.NullString:
if aux == nil || !aux.Valid {
return
}
rv.Value = aux.String
case *sql.NullBool:
if aux == nil || !aux.Valid {
return
}
if aux.Bool {
rv.Value = "true"
} else {
rv.Value = "false"
}
case *sql.RawBytes:
if raw == nil {
return
}
rv.Value = string(*aux)
case *uuid.UUID:
if raw == nil {
return
}
rv.Value = aux.String()
default:
return
}
r.Values = r.Values.Set(rv)
return nil
}
func decodeSystemField(ident string, raw any, r *types.Record) error {
switch ident {
case sysID:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.ID = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysID)
}
case sysNamespaceID:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.ID = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysNamespaceID)
}
case sysModuleID:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.ID = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysModuleID)
}
case sysCreatedAt:
if tmp, is := raw.(*sql.NullTime); is {
if tmp.Valid {
r.CreatedAt = tmp.Time
}
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysCreatedAt)
}
case sysCreatedBy:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.CreatedBy = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysCreatedBy)
}
case sysUpdatedAt:
if tmp, is := raw.(*sql.NullTime); is {
if tmp.Valid {
r.UpdatedAt = &tmp.Time
}
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysUpdatedAt)
}
case sysUpdatedBy:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.UpdatedBy = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysUpdatedBy)
}
case sysDeletedAt:
if tmp, is := raw.(*sql.NullTime); is {
if tmp.Valid {
r.DeletedAt = &tmp.Time
}
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysDeletedAt)
}
case sysDeletedBy:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.DeletedBy = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysDeletedBy)
}
case sysOwnedBy:
if tmp, is := raw.(*uint64); is && tmp != nil {
r.OwnedBy = *tmp
} else {
return fmt.Errorf("incompatible type %T for system field %s", raw, sysOwnedBy)
}
}
return nil
}

701
store/adapters/rdbms/crs/driver.go Normal file
View File

@@ -0,0 +1,701 @@
package crs
import (
"context"
"database/sql"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/data"
"github.com/cortezaproject/corteza-server/pkg/filter"
"github.com/cortezaproject/corteza-server/store/adapters/rdbms"
"github.com/doug-martin/goqu/v9"
"github.com/doug-martin/goqu/v9/exp"
)
type (
sqlizer interface {
ToSQL() (string, []interface{}, error)
}
connection interface {
ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
QueryRowContext(context.Context, string, ...interface{}) *sql.Row
QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
}
attrExpression interface {
exp.Comparable
exp.Inable
exp.Isable
}
attributeType interface {
Type() data.AttributeType
}
column struct {
ident string
columnType attributeType
attributes []*data.Attribute
encode func([]*data.Attribute, *types.Record) (any, error)
decode func([]*data.Attribute, any, *types.Record) error
}
crs struct {
model *data.Model
conn connection
deepJsonFn func(ident string, pp ...any) exp.LiteralExpression
table exp.IdentifierExpression
// ID column identifier expression
sysColumnID exp.IdentifierExpression
// optional record fields/columns/expressions
sysExprNamespaceID attrExpression
sysExprModuleID attrExpression
sysExprDeletedAt attrExpression
// all columns we're selecting from when
// we're selecting from all columns
columns []*column
}
)
// CRS returns fully initialized model store
//
// It abstracts database table and its columns and provides unified interface
// for fetching and storing records.
func CRS(m *data.Model, conn connection) *crs {
var (
ms = &crs{
model: m,
conn: conn,
}
)
ms.table = exp.NewIdentifierExpression("", m.Ident, "")
ms.deepJsonFn = rdbms.DeepIdentJSON
_ = ms.genColumns()
return ms
}
// generates columns from model attributes.
//
// Important note!
// Attribute name is not always equal to name of the column
// (case of aliased fields or embedded values)
func (ms *crs) genColumns() error {
var (
colIdent string
att *data.Attribute
uniqCols = make(map[string]int)
cols = make([]*column, 0, len(ms.model.Attributes))
)
for a := range ms.model.Attributes {
att = ms.model.Attributes[a]
colIdent = attrColumnIdent(att)
// there are a couple of important system attributes/columns
// we need to locate and link for easier management
switch att.Ident {
case sysID:
ms.sysColumnID = ms.table.Col(colIdent)
case sysNamespaceID:
ms.sysExprNamespaceID = ms.table.Col(colIdent)
case sysModuleID:
ms.sysExprModuleID = ms.table.Col(colIdent)
case sysDeletedAt:
ms.sysExprDeletedAt = ms.table.Col(colIdent)
}
// identify db columns from list of attributes
colIndex, has := uniqCols[colIdent]
if has {
// column already initialized one of the previous iterations
// append attribute and continue with the next one
cols[colIndex].attributes = append(cols[colIndex].attributes, att)
continue
}
colIndex = len(cols)
uniqCols[colIdent] = colIndex
cols = append(cols, &column{ident: colIdent, attributes: []*data.Attribute{att}})
if data.StoreCodecStdRecordValueJSONType.Is(att.Store) {
// when dealing with encoded types there is probably
// a different column that can handle the encoded payload
cols[colIndex].columnType = &data.TypeJSON{}
cols[colIndex].decode = decodeStdRecordValueJSON
cols[colIndex].encode = encodeStdRecordValueJSON
} else {
cols[colIndex].columnType = att.Type
if isSystemField(att.Ident) {
cols[colIndex].decode = func(aa []*data.Attribute, raw any, r *types.Record) error {
return decodeSystemField(aa[0].Ident, raw, r)
}
cols[colIndex].encode = func(aa []*data.Attribute, r *types.Record) (any, error) {
// ignoring attribute slice because it's strictly one field we will be dealing with
// + only interested in the single-value (first) fields
return getSystemFieldValue(r, aa[0].Ident)
}
} else {
cols[colIndex].decode = decodeRecordValue
cols[colIndex].encode = func(aa []*data.Attribute, r *types.Record) (any, error) {
// ignoring attribute slice because it's strictly one field we will be dealing with
// + only interested in the single-value (first) fields
//
// any potential incompatibilities should be resolved before this point!
if v := r.Values.Get(aa[0].Ident, 0); v != nil {
return v.Value, nil
}
return nil, nil
}
}
}
}
ms.columns = cols
return nil
}
func (ms *crs) Create(ctx context.Context, rr ...*types.Record) error {
sql, args, err := ms.insertSql(rr...).ToSQL()
if err != nil {
return err
}
_, err = ms.conn.ExecContext(ctx, sql, args...)
return err
}
func (ms *crs) Update(ctx context.Context, r *types.Record) error {
sql, args, err := ms.updateSql(r).ToSQL()
if err != nil {
return err
}
_, err = ms.conn.ExecContext(ctx, sql, args...)
return err
}
func (ms *crs) Delete(ctx context.Context, r *types.Record) error {
sql, args, err := ms.deleteByIdSql(r.ID).ToSQL()
if err != nil {
return err
}
_, err = ms.conn.ExecContext(ctx, sql, args...)
return err
}
func (ms *crs) Search(ctx context.Context, f types.RecordFilter) (i *iterator, err error) {
// @todo apply filter
// @todo apply cursor constraints
if f.PageCursor != nil {
// Page cursor exists; we need to validate it against used sort
// To cover the case when paging cursor is set but sorting is empty, we collect the sorting instructions
// from the cursor.
// This (extracted sorting info) is then returned as part of response
if f.Sort, err = f.PageCursor.Sort(f.Sort); err != nil {
return
}
}
// Make sure results are always sorted at least by primary key
if f.Sort.Get(sysID) == nil {
f.Sort = append(f.Sort, &filter.SortExpr{
Column: sysID,
Descending: f.Sort.LastDescending(),
})
}
return &iterator{ms: ms}, nil
}
//func Search(ctx context.Context, s querier, m *data.Model, f types.RecordFilter) (set types.RecordSet, _ types.RecordFilter, err error) {
// // Cleanup unwanted cursor values (only relevant is f.PageCursor, next&prev are reset and returned)
// f.PrevPage, f.NextPage = nil, nil
//
// if f.PageCursor != nil {
// // Page cursor exists; we need to validate it against used sort
// // To cover the case when paging cursor is set but sorting is empty, we collect the sorting instructions
// // from the cursor.
// // This (extracted sorting info) is then returned as part of response
// if f.Sort, err = f.PageCursor.Sort(f.Sort); err != nil {
// return
// }
// }
//
// // Make sure results are always sorted at least by primary keys
// if f.Sort.Get("id") == nil {
// f.Sort = append(f.Sort, &filter.SortExpr{
// Column: "id",
// Descending: f.Sort.LastDescending(),
// })
// }
//
// // Cloned sorting instructions for the actual sorting
// // Original are passed to the etchFullPageOfApplications fn used for cursor creation;
// // direction information it MUST keep the initial
// sort := f.Sort.Clone()
//
// // When cursor for a previous page is used it's marked as reversed
// // This tells us to flip the descending flag on all used sort keys
// if f.PageCursor != nil && f.PageCursor.ROrder {
// sort.Reverse()
// }
//
// set, f.PrevPage, f.NextPage, err = fetchFullPage(ctx, s, m, f, sort)
//
// f.PageCursor = nil
// if err != nil {
// return nil, f, err
// }
//
// return set, f, nil
//
//}
//
//func fetchFullPage(ctx context.Context, s querier, m *data.Model, filter types.RecordFilter, sort filter.SortExprSet) (set []*types.Record, prev, next *filter.PagingCursor, err error) {
// var (
// aux []*types.Record
//
// // When cursor for a previous page is used it's marked as reversed
// // This tells us to flip the descending flag on all used sort keys
// reversedOrder = filter.PageCursor != nil && filter.PageCursor.ROrder
//
// // Copy no. of required items to limit
// // Limit will change when doing subsequent queries to fill
// // the set with all required items
// limit = filter.Limit
//
// reqItems = filter.Limit
//
// // cursor to prev. page is only calculated when cursor is used
// hasPrev = filter.PageCursor != nil
//
// // next cursor is calculated when there are more pages to come
// hasNext bool
//
// tryFilter types.RecordFilter
// )
//
// set = make([]*types.Record, 0, rdbms.DefaultSliceCapacity)
//
// for try := 0; try < rdbms.MaxRefetches; try++ {
// // Copy filter & apply custom sorting that might be affected by cursor
// tryFilter = filter
// tryFilter.Sort = sort
//
// if limit > 0 {
// // fetching + 1 to peak ahead if there are more items
// // we can fetch (next-page cursor)
// tryFilter.Limit = limit + 1
// }
//
// if aux, hasNext, err = query(ctx, s, m, tryFilter); err != nil {
// return nil, nil, nil, err
// }
//
// if len(aux) == 0 {
// // nothing fetched
// break
// }
//
// // append fetched items
// set = append(set, aux...)
//
// if reqItems == 0 || !hasNext {
// // no max requested items specified, break out
// break
// }
//
// collected := uint(len(set))
//
// if reqItems > collected {
// // not enough items fetched, try again with adjusted limit
// limit = reqItems - collected
//
// if limit < rdbms.MinEnsureFetchLimit {
// // In case limit is set very low and we've missed records in the first fetch,
// // make sure next fetch limit is a bit higher
// limit = rdbms.MinEnsureFetchLimit
// }
//
// // Update cursor so that it points to the last item fetched
// tryFilter.PageCursor = collectCursorValues(set[collected-1], filter.Sort...)
//
// // Copy reverse flag from sorting
// tryFilter.PageCursor.LThen = filter.Sort.Reversed()
// continue
// }
//
// if reqItems < collected {
// set = set[:reqItems]
// }
//
// break
// }
//
// collected := len(set)
//
// if collected == 0 {
// return nil, nil, nil, nil
// }
//
// if reversedOrder {
// // Fetched set needs to be reversed because we've forced a descending order to get the previous page
// for i, j := 0, collected-1; i < j; i, j = i+1, j-1 {
// set[i], set[j] = set[j], set[i]
// }
//
// // when in reverse-order rules on what cursor to return change
// hasPrev, hasNext = hasNext, hasPrev
// }
//
// if hasPrev {
// prev = collectCursorValues(set[0], filter.Sort...)
// prev.ROrder = true
// prev.LThen = !filter.Sort.Reversed()
// }
//
// if hasNext {
// next = collectCursorValues(set[collected-1], filter.Sort...)
// next.LThen = filter.Sort.Reversed()
// }
//
// return set, prev, next, nil
//}
//
//func collectCursorValues(res *types.Record, cc ...*filter.SortExpr) *filter.PagingCursor {
// var (
// cur = &filter.PagingCursor{LThen: filter.SortExprSet(cc).Reversed()}
//
// hasUnique bool
//
// pkID bool
//
// collect = func(cc ...*filter.SortExpr) {
// for _, c := range cc {
// switch c.Column {
// case sysID:
// cur.Set(c.Column, res.ID, c.Descending)
// pkID = true
// case "createdAt":
// cur.Set(c.Column, res.CreatedAt, c.Descending)
// case "updatedAt":
// cur.Set(c.Column, res.UpdatedAt, c.Descending)
// case "deletedAt":
// cur.Set(c.Column, res.DeletedAt, c.Descending)
// }
// }
// }
// )
//
// collect(cc...)
// if !hasUnique || !pkID {
// collect(&filter.SortExpr{Column: "id", Descending: false})
// }
//
// return cur
//}
//
//func query(ctx context.Context, s querier, m *data.Model, f types.RecordFilter) (_ []*types.Record, more bool, err error) {
// var (
// ok bool
//
// set = make([]*types.Record, 0, rdbms.DefaultSliceCapacity)
// res *types.Record
// rows *sql.Rows
// count uint
// expr, tExpr []goqu.Expression
//
// sortExpr []exp.OrderedExpression
//
// sql *goqu.SelectDataset
// )
//
// if err != nil {
// err = fmt.Errorf("could generate filter expression for Application: %w", err)
// return
// }
//
// // paging feature is enabled
// if f.PageCursor != nil {
// if tExpr, err = rdbms.Cursor(f.PageCursor); err != nil {
// return
// } else {
// expr = append(expr, tExpr...)
// }
// }
//
// sql, err = ms.searchSql(f)
// sql.Where(expr...)
//
// //query, := searchSql(m, f).Where(expr...)
//
// // sorting feature is enabled
// if sortExpr, err = rdbms.Order(f.Sort, nil); err != nil {
// err = fmt.Errorf("could generate order expression for Application: %w", err)
// return
// }
//
// if len(sortExpr) > 0 {
// sql = sql.Order(sortExpr...)
// }
//
// if f.Limit > 0 {
// sql = sql.Limit(f.Limit)
// }
//
// rows, err = s.Query(ctx, sql)
// if err != nil {
// err = fmt.Errorf("could not query Application: %w", err)
// return
// }
//
// if err = rows.Err(); err != nil {
// err = fmt.Errorf("could not query Application: %w", err)
// return
// }
//
// defer func() {
// closeError := rows.Close()
// if err == nil {
// // return error from close
// err = closeError
// }
// }()
//
// for rows.Next() {
// if err = rows.Err(); err != nil {
// err = fmt.Errorf("could not query Application: %w", err)
// return
// }
//
// listOfSelectedColumns := make([]any, 10)
//
// // @todo this does not work!
//
// if err = rows.Scan(listOfSelectedColumns...); err != nil {
// err = fmt.Errorf("could not scan rows for Application: %w", err)
// return
// }
//
// count++
// // @todo convert listOfSelectedCOlumns into *types.Record
//
// //if res, err = aux.decode(); err != nil {
// // err = fmt.Errorf("could not decode Application: %w", err)
// // return
// //}
//
// // check fn set, call it and see if it passed the test
// // if not, skip the item
// if f.Check != nil {
// if ok, err = f.Check(res); err != nil {
// return
// } else if !ok {
// continue
// }
// }
//
// set = append(set, res)
// }
//
// return set, f.Limit > 0 && count >= f.Limit, err
//
//}
// constructs SQL for selecting records from a table, converting parts of record filter into conditions
func (ms *crs) searchSql(f types.RecordFilter) *goqu.SelectDataset {
var (
cnd []exp.Expression
)
{
// Add module & namespace constraints when model expects (has configured attributes) for them
//
// This covers both scenarios:
// 1) Model is configured to store records in a dedicated table
// without module and/or namespace attributes
//
// 2) Model has module and/or namespace attribute and saves records
// from different modules in the same table
if ms.sysExprNamespaceID != nil {
cnd = append(cnd, ms.sysExprNamespaceID.Eq(f.NamespaceID))
}
if ms.sysExprModuleID != nil {
cnd = append(cnd, ms.sysExprModuleID.Eq(f.ModuleID))
}
}
{
// Limit by LabeledIDs (list of record IDs)
cnd = append(cnd, ms.sysColumnID.In(f.LabeledIDs))
}
{
// If module supports soft-deletion (= delete-at attribute is present)
// we need to make sure we respect it
if ms.sysExprDeletedAt != nil {
switch f.Deleted {
case filter.StateExclusive:
// only not-null values
cnd = append(cnd, ms.sysExprDeletedAt.IsNotNull())
case filter.StateExcluded:
// exclude all non-null values
cnd = append(cnd, ms.sysExprDeletedAt.IsNull())
}
}
}
// @todo ql (AST) => []exp.Expression
return ms.selectSql().Where(cnd...)
}
func (ms *crs) lookupByIdSql(id uint) *goqu.SelectDataset {
return ms.selectSql().
Where(ms.sysColumnID.Eq(id)).
Limit(1)
}
func (ms *crs) selectSql() *goqu.SelectDataset {
// working around a bug inside goqu lib that adds
// * to the list of columns to be selected
// even if we clear the columns first
q := goqu.From(ms.table).Select(ms.table.Col(ms.columns[0].ident))
for _, col := range ms.columns[1:] {
q = q.SelectAppend(ms.table.Col(col.ident))
}
return q
}
func (ms *crs) insertSql(rr ...*types.Record) (_ *goqu.InsertDataset) {
var (
err error
rows = make([]any, len(rr))
)
for i, r := range rr {
rows[i], err = encode(ms.columns, r)
if err != nil {
return (&goqu.InsertDataset{}).SetError(err)
}
}
return goqu.
Insert(ms.table).
Rows(rows...)
}
// updateSql generates SQL command for updating record
//
// Integrity check (ie module, namespace, dates, owners change) is out of scope
// for this layer
func (ms *crs) updateSql(r *types.Record) *goqu.UpdateDataset {
encValues, err := encode(ms.columns, r, sysID)
if err != nil {
return (&goqu.UpdateDataset{}).SetError(err)
}
return goqu.
Update(ms.table).
Set(encValues).
Where(ms.sysColumnID.Eq(r.ID))
}
func (ms *crs) deleteByIdSql(id uint64) *goqu.DeleteDataset {
return goqu.Delete(ms.table).Where(ms.sysColumnID.Eq(id))
}
//func (ms *crs) attrToExpr(ident string) (exp.Comparable, bool) {
// if !ms.model.HasAttribute(ident) {
// return nil, false
// }
//
// attr := ms.model.Attributes.FindByIdent(ident)
// switch s := attr.Store.(type) {
// case *data.StoreCodecAlias:
// // using column directly
// return exp.NewIdentifierExpression("", ms.model.Ident, s.Ident), true
//
// case *data.StoreCodecStdRecordValueJSON:
// // using JSON to handle embedded values
// return ms.deepJsonFn(ms.model.Ident, s.Ident, 0), true
// }
//
// return exp.NewIdentifierExpression("", ms.model.Ident, ident), true
//}
func attrColumnIdent(att *data.Attribute) string {
switch ss := att.Store.(type) {
case *data.StoreCodecStdRecordValueJSON:
return ss.Ident
case *data.StoreCodecAlias:
return ss.Ident
default:
return att.Ident
}
}
//func extractColumns(m *data.Model) []exp.IdentifierExpression {
// var (
// col string
// uniq = make(map[string]bool)
// ii = make([]exp.IdentifierExpression, 0, len(aa))
// )
//
// for _, a := range m.Attributes {
// col = attrColumnIdent(a)
// if uniq[col] {
// continue
// }
//
// uniq[col] = true
// ii = append(ii, col)
// }
//
// return
//}
//
//func valueExpr(att *data.Attribute) (exp.Expression, error) {
// if att.Store == nil {
// return exp.NewIdentifierExpression("", "", att.Ident), nil
//
// }
// switch ss := att.Store.(type) {
// case data.StoreCodecJSON:
// return exp.NewIdentifierExpression("", "", ss.Ident), nil
// case data.StoreCodecAlias:
// return exp.NewIdentifierExpression("", "", ss.Ident).As(att.Ident), nil
//
// default:
// return nil, fmt.Errorf("unknown store strategy %T for attribute %q", ss, att.Ident)
// }
//}

85
store/adapters/rdbms/crs/driver_test.go Normal file
View File

@@ -0,0 +1,85 @@
package crs
import (
"testing"
"time"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/data"
"github.com/stretchr/testify/require"
)
func Test_sqlizers(t *testing.T) {
var (
m = &data.Model{
Ident: "test_tbl",
Attributes: data.AttributeSet{
&data.Attribute{Ident: sysID, Type: &data.TypeID{}, Store: &data.StoreCodecAlias{Ident: "id"}},
&data.Attribute{Ident: sysCreatedAt, Type: &data.TypeTimestamp{}, Store: &data.StoreCodecAlias{Ident: "created_at"}},
&data.Attribute{Ident: sysUpdatedAt, Type: &data.TypeTimestamp{}, Store: &data.StoreCodecAlias{Ident: "updated_at"}},
},
}
ms = CRS(m, nil)
req = require.New(t)
q sqlizer
sql string
args []any
err error
tenTenTen, _ = time.Parse("2006-01-02", "2010-10-10")
)
cases := []struct {
fn func() sqlizer
sql string
args []any
err error
}{
{
fn: func() sqlizer { return ms.lookupByIdSql(10) },
sql: `SELECT "test_tbl"."id", "test_tbl"."created_at", "test_tbl"."updated_at" FROM "test_tbl" WHERE ("test_tbl"."id" = 10) LIMIT 1`,
args: []any{},
err: nil,
},
{
fn: func() sqlizer {
return ms.updateSql(&types.Record{ID: 10, CreatedAt: tenTenTen, UpdatedAt: &tenTenTen})
},
sql: `UPDATE "test_tbl" SET "created_at"='2010-10-10T00:00:00Z',"updated_at"='2010-10-10T00:00:00Z' WHERE ("test_tbl"."id" = 10)`,
args: []any{},
err: nil,
},
{
fn: func() sqlizer { return ms.insertSql(&types.Record{ID: 10, CreatedAt: tenTenTen}) },
sql: `INSERT INTO "test_tbl" ("created_at", "id", "updated_at") VALUES ('2010-10-10T00:00:00Z', 10, NULL)`,
args: []any{},
err: nil,
},
{
fn: func() sqlizer { return ms.deleteByIdSql(12345) },
sql: `DELETE FROM "test_tbl" WHERE ("test_tbl"."id" = 12345)`,
args: []any{},
err: nil,
},
}
for _, c := range cases {
t.Run("", func(t *testing.T) {
q = c.fn()
sql, args, err = q.ToSQL()
if c.args == nil {
req.NoError(err)
} else {
req.ErrorIs(err, c.err)
}
req.Equal(c.sql, sql)
req.Equal(c.args, args)
})
}
}

120
store/adapters/rdbms/crs/encoder.go Normal file
View File

@@ -0,0 +1,120 @@
package crs
import (
"encoding/json"
"fmt"
"sort"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/data"
"github.com/cortezaproject/corteza-server/pkg/slice"
"github.com/spf13/cast"
)
// Encode takes a record and encodes it according to the given model
// encoding record values into form acceptable by underlying storage
//
// with omit slice you specify list of attribute idents to be omitted!
func encode(columns []*column, r *types.Record, omit ...string) (enc map[string]any, err error) {
enc = make(map[string]any)
for _, col := range columns {
if slice.ContainsAny(omit, col.attributes[0].Ident) {
continue
}
if enc[col.ident], err = col.encode(col.attributes, r); err != nil {
return nil, fmt.Errorf("could not encode value for attribute %q: %w", col.ident, err)
}
}
return
}
// @todo this could probably be generalized, nothing special for RDBMS
func encodeStdRecordValueJSON(attributes []*data.Attribute, r *types.Record) (any, error) {
var (
aux = make(map[string][]any)
rvs types.RecordValueSet
value any
place int
size int
)
for _, attr := range attributes {
place = 0
if isSystemField(attr.Ident) {
// for system fields just get the one
val, err := getSystemFieldValue(r, attr.Ident)
if err != nil {
return nil, err
}
aux[attr.Ident] = []any{val}
} else {
// for the record values, collect all we
// care about, sort and copy them to the auxiliary var.
rvs = r.Values.FilterByName(attr.Ident)
sort.Sort(rvs)
if size = len(rvs); size == 0 {
continue
}
aux[attr.Ident] = make([]any, size)
for _, v := range rvs {
aux[attr.Ident][place] = v.Value
}
}
// now, encode the value according to JSON format constraints
for place, value = range aux[attr.Ident] {
switch attr.Type.(type) {
case data.TypeBoolean:
aux[attr.Ident][place] = cast.ToBool(value)
default:
// by default: as string
aux[attr.Ident][place] = value
}
if !attr.MultiValue {
// model attribute supports storing of single values only.
break
}
}
}
return json.Marshal(aux)
}
// extracts value from a record with system fields as priority
func getSystemFieldValue(r *types.Record, name string) (any, error) {
switch name {
// handle system fields
case sysID:
return r.ID, nil
case sysNamespaceID:
return r.NamespaceID, nil
case sysModuleID:
return r.ModuleID, nil
case sysCreatedAt:
return r.CreatedAt, nil
case sysCreatedBy:
return r.CreatedBy, nil
case sysUpdatedAt:
return r.UpdatedAt, nil
case sysUpdatedBy:
return r.UpdatedBy, nil
case sysDeletedAt:
return r.DeletedAt, nil
case sysDeletedBy:
return r.DeletedBy, nil
case sysOwnedBy:
return r.OwnedBy, nil
}
return nil, fmt.Errorf("unknown compose record system field %q", name)
}

113
store/adapters/rdbms/crs/fetcher.go Normal file
View File

@@ -0,0 +1,113 @@
package crs
import (
"context"
"database/sql"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/data"
"github.com/cortezaproject/corteza-server/pkg/filter"
"github.com/google/uuid"
)
type (
iterator struct {
ms *crs
rows *sql.Rows
err error
next *types.Record
// @todo should filter also be here?
// buffer for scanned rows
scanBuf []any
}
)
func (i *iterator) Next(ctx context.Context) bool {
if i.err == nil && i.rows == nil {
i.rows, i.err = i.fetch(ctx)
}
if i.err != nil {
return false
}
return i.rows.Next()
}
func (i *iterator) fetch(ctx context.Context) (*sql.Rows, error) {
if i.scanBuf == nil {
// we're going to init scan buffer only once
// and rely on the sql.Rows.Scan function to
// fill it up with fresh values!
i.scanBuf = newScanBuffer(i.ms)
}
q := i.ms.selectSql()
// todo filter?
// todo cursor constraints?
sql, args, err := q.ToSQL()
if err != nil {
return nil, err
}
return i.ms.conn.QueryContext(ctx, sql, args...)
}
func (i *iterator) Scan(r *types.Record) (err error) {
if err = i.rows.Scan(i.scanBuf...); err != nil {
return
}
return decode(i.ms.columns, i.scanBuf, r)
}
func (i *iterator) Err() error {
return i.err
}
// Close iterator and cleanup
func (i *iterator) Close() error {
return i.rows.Close()
}
// prepare slice of (typed) placeholders we're going to
// push to be used by rows.Scan
func newScanBuffer(d *crs) []any {
var (
vv = make([]any, len(d.columns))
)
for i, c := range d.columns {
switch c.columnType.(type) {
case *data.TypeID, *data.TypeRef:
vv[i] = new(uint64)
case *data.TypeTimestamp, *data.TypeDate:
vv[i] = new(sql.NullTime)
case *data.TypeTime:
vv[i] = new(sql.NullString)
case *data.TypeNumber:
vv[i] = new(sql.NullString)
case *data.TypeText:
vv[i] = new(sql.NullString)
case *data.TypeBoolean:
vv[i] = new(sql.NullBool)
case *data.TypeEnum:
vv[i] = new(sql.NullString)
case *data.TypeJSON, *data.TypeGeometry, *data.TypeBlob:
vv[i] = new(sql.RawBytes)
case *data.TypeUUID:
vv[i] = new(uuid.UUID)
}
i++
}
return vv
}
func CursorBackward(r *types.Record, sort filter.SortExprSet) *filter.PagingCursor { return nil }
func CursorForward(r *types.Record, sort filter.SortExprSet) *filter.PagingCursor { return nil }
func Drain(*iterator) ([]*types.Record, error) { return nil, nil }

95
store/adapters/rdbms/crs/fetcher_test.go Normal file
View File

@@ -0,0 +1,95 @@
package crs
import (
"context"
"testing"
"time"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/data"
"github.com/cortezaproject/corteza-server/pkg/id"
"github.com/cortezaproject/corteza-server/pkg/logger"
"github.com/cortezaproject/corteza-server/store/adapters/rdbms"
"github.com/cortezaproject/corteza-server/store/adapters/rdbms/drivers/sqlite"
)
const (
repeatReads = 1000
totalRecords = 100
)
func TestIteratorNG(t *testing.T) {
ctx := context.Background()
ctx = logger.ContextWithValue(ctx, logger.MakeDebugLogger())
cfg, err := sqlite.NewConfig("sqlite3://file::memory:?cache=shared&mode=memory")
//cfg, err := postgres.NewConfig("postgres://darh@localhost:5432/corteza_2022_3?sslmode=disable&")
if err != nil {
t.Errorf("could not connect: %v", err)
}
db, err := rdbms.Connect(ctx, logger.Default(), cfg)
if err != nil {
t.Errorf("could not connect: %v", err)
}
_, err = db.ExecContext(ctx, `
CREATE TABLE test_tbl (
id NUMERIC NOT NULL,
created_at TIMESTAMP,
updated_at TIMESTAMP,
phy TEXT,
meta JSON
)
`)
if err != nil {
t.Errorf("could not create table: %v", err)
}
m := &data.Model{
Ident: "test_tbl",
Attributes: data.AttributeSet{
&data.Attribute{Ident: sysID, Type: &data.TypeID{}, Store: &data.StoreCodecAlias{Ident: "id"}},
&data.Attribute{Ident: sysCreatedAt, Type: &data.TypeTimestamp{}, Store: &data.StoreCodecAlias{Ident: "created_at"}},
&data.Attribute{Ident: sysUpdatedAt, Type: &data.TypeTimestamp{}, Store: &data.StoreCodecAlias{Ident: "updated_at"}},
&data.Attribute{Ident: "foo", Type: &data.TypeText{}, Store: &data.StoreCodecStdRecordValueJSON{Ident: "meta"}},
&data.Attribute{Ident: "bar", Type: &data.TypeNumber{}, Store: &data.StoreCodecStdRecordValueJSON{Ident: "meta"}},
&data.Attribute{Ident: "baz", Type: &data.TypeBoolean{}, Store: &data.StoreCodecStdRecordValueJSON{Ident: "meta"}},
&data.Attribute{Ident: "bbb", Type: &data.TypeUUID{}, Store: &data.StoreCodecStdRecordValueJSON{Ident: "meta"}},
&data.Attribute{Ident: "phy", Type: &data.TypeText{}, Store: &data.StoreCodecPlain{}},
},
}
d := CRS(m, db)
_, err = db.Exec("DELETE FROM test_tbl")
if err != nil {
t.Errorf("could not truncate: %v", err)
}
for i := 0; i < totalRecords; i++ {
r := &types.Record{ID: id.Next(), CreatedAt: time.Now()}
r.Values = r.Values.Set(&types.RecordValue{Name: "foo", Value: "FooVal1"})
r.Values = r.Values.Set(&types.RecordValue{Name: "bar", Value: "34234"})
r.Values = r.Values.Set(&types.RecordValue{Name: "baz", Value: "true"})
r.Values = r.Values.Set(&types.RecordValue{Name: "phy", Value: "lly"})
if err = d.Create(ctx, r); err != nil {
t.Errorf("could not create: %v", err)
}
}
t.Logf("records (%dx) created", totalRecords)
i := &iterator{
ms: d,
}
for rep := 0; rep < repeatReads; rep++ {
for i.Next(ctx) {
r := new(types.Record)
if err = i.Scan(r); err != nil {
t.Errorf("could not scan: %v", err)
}
}
}
}

34
store/adapters/rdbms/crs/record.go Normal file
View File

@@ -0,0 +1,34 @@
package crs
const (
sysID = "ID"
sysNamespaceID = "namespaceID"
sysModuleID = "moduleID"
sysCreatedAt = "createdAt"
sysCreatedBy = "createdBy"
sysUpdatedAt = "updatedAt"
sysUpdatedBy = "updatedBy"
sysDeletedAt = "deletedAt"
sysDeletedBy = "deletedBy"
sysOwnedBy = "ownedBy"
)
// is the
func isSystemField(f string) bool {
switch f {
// handle system fields
case sysID,
sysNamespaceID,
sysModuleID,
sysCreatedAt,
sysCreatedBy,
sysUpdatedAt,
sysUpdatedBy,
sysDeletedAt,
sysDeletedBy,
sysOwnedBy:
return true
default:
return false
}
}