From 2d06cf1cf92a6a7710f0c5be017da539cc538a2a Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Sat, 23 Apr 2022 16:21:55 +0200 Subject: [PATCH] Initial CRS implementation for RDBMS --- compose/crs/driver.go | 7 + compose/crs/model.go | 2 +- pkg/data/model.go | 22 +- pkg/data/store_codec.go | 53 +- pkg/slice/strings.go | 26 + store/adapters/rdbms/crs/decoder.go | 203 +++++++ store/adapters/rdbms/crs/driver.go | 701 +++++++++++++++++++++++ store/adapters/rdbms/crs/driver_test.go | 85 +++ store/adapters/rdbms/crs/encoder.go | 120 ++++ store/adapters/rdbms/crs/fetcher.go | 113 ++++ store/adapters/rdbms/crs/fetcher_test.go | 95 +++ store/adapters/rdbms/crs/record.go | 34 ++ 12 files changed, 1448 insertions(+), 13 deletions(-) create mode 100644 store/adapters/rdbms/crs/decoder.go create mode 100644 store/adapters/rdbms/crs/driver.go create mode 100644 store/adapters/rdbms/crs/driver_test.go create mode 100644 store/adapters/rdbms/crs/encoder.go create mode 100644 store/adapters/rdbms/crs/fetcher.go create mode 100644 store/adapters/rdbms/crs/fetcher_test.go create mode 100644 store/adapters/rdbms/crs/record.go diff --git a/compose/crs/driver.go b/compose/crs/driver.go index 4035ea227..a3628dbcc 100644 --- a/compose/crs/driver.go +++ b/compose/crs/driver.go @@ -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 } diff --git a/compose/crs/model.go b/compose/crs/model.go index 626e85871..6f5a9b9d8 100644 --- a/compose/crs/model.go +++ b/compose/crs/model.go @@ -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, diff --git a/pkg/data/model.go b/pkg/data/model.go index eb89fa013..b27878cbc 100644 --- a/pkg/data/model.go +++ b/pkg/data/model.go @@ -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 { diff --git a/pkg/data/store_codec.go b/pkg/data/store_codec.go index 3ddc09a5b..15802bc69 100644 --- a/pkg/data/store_codec.go +++ b/pkg/data/store_codec.go @@ -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 } + // + // 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 } diff --git a/pkg/slice/strings.go b/pkg/slice/strings.go index e625f0201..2977a5a8e 100644 --- a/pkg/slice/strings.go +++ b/pkg/slice/strings.go @@ -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 { diff --git a/store/adapters/rdbms/crs/decoder.go b/store/adapters/rdbms/crs/decoder.go new file mode 100644 index 000000000..000bbd542 --- /dev/null +++ b/store/adapters/rdbms/crs/decoder.go @@ -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 +} diff --git a/store/adapters/rdbms/crs/driver.go b/store/adapters/rdbms/crs/driver.go new file mode 100644 index 000000000..bcdc415cc --- /dev/null +++ b/store/adapters/rdbms/crs/driver.go @@ -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) +// } +//} diff --git a/store/adapters/rdbms/crs/driver_test.go b/store/adapters/rdbms/crs/driver_test.go new file mode 100644 index 000000000..6b3104ef8 --- /dev/null +++ b/store/adapters/rdbms/crs/driver_test.go @@ -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) + + }) + } + +} diff --git a/store/adapters/rdbms/crs/encoder.go b/store/adapters/rdbms/crs/encoder.go new file mode 100644 index 000000000..45f3d9879 --- /dev/null +++ b/store/adapters/rdbms/crs/encoder.go @@ -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) +} diff --git a/store/adapters/rdbms/crs/fetcher.go b/store/adapters/rdbms/crs/fetcher.go new file mode 100644 index 000000000..bbbe8f4ac --- /dev/null +++ b/store/adapters/rdbms/crs/fetcher.go @@ -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 } diff --git a/store/adapters/rdbms/crs/fetcher_test.go b/store/adapters/rdbms/crs/fetcher_test.go new file mode 100644 index 000000000..371059c02 --- /dev/null +++ b/store/adapters/rdbms/crs/fetcher_test.go @@ -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) + } + } + } +} diff --git a/store/adapters/rdbms/crs/record.go b/store/adapters/rdbms/crs/record.go new file mode 100644 index 000000000..da8bbfdbe --- /dev/null +++ b/store/adapters/rdbms/crs/record.go @@ -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 + } +}