Add RDBMS DAL model validation

This will prevent misconfigured models with atributes that read from the
same columns. This constraint could/should be removed in the
future.

pkg/dal/attribute_codec.go

@@ -11,6 +11,7 @@ type (
 	// source/destination of the value (table column, json document property)
 	Codec interface {
 		Type() AttributeCodecType
+		SingleValueOnly() bool
 	}
 
 	CodecPlain struct{}
@@ -56,3 +57,7 @@ func (*CodecRecordValueSetJSON) Type() AttributeCodecType {
 	return "corteza::dal:attribute-codec:record-value-set-json"
 }
 func (*CodecAlias) Type() AttributeCodecType { return "corteza::dal:attribute-codec:alias" }
+
+func (*CodecPlain) SingleValueOnly() bool              { return true }
+func (*CodecRecordValueSetJSON) SingleValueOnly() bool { return false }
+func (*CodecAlias) SingleValueOnly() bool              { return true }

store/adapters/rdbms/dal/connection.go

@@ -129,6 +129,12 @@ func (c *connection) Models(ctx context.Context) (dal.ModelSet, error) {
 //
 // @todo DDL operations
 func (c *connection) CreateModel(ctx context.Context, mm ...*dal.Model) (err error) {
+	for _, m := range mm {
+		if err = validate(m); err != nil {
+			return
+		}
+	}
+
 	c.mux.Lock()
 	defer c.mux.Unlock()
 	for _, m := range mm {
@@ -172,7 +178,11 @@ func (c *connection) DeleteModel(ctx context.Context, mm ...*dal.Model) (err err
 //
 // @todo DDL operations
 // @todo some tables should not be removed (like compose_record on primary connection)
-func (c *connection) UpdateModel(ctx context.Context, old *dal.Model, new *dal.Model) error {
+func (c *connection) UpdateModel(ctx context.Context, old *dal.Model, new *dal.Model) (err error) {
+	if err = validate(new); err != nil {
+		return
+	}
+
 	c.mux.Lock()
 	defer c.mux.Unlock()
 
@@ -183,7 +193,7 @@ func (c *connection) UpdateModel(ctx context.Context, old *dal.Model, new *dal.M
 
 	// update the cache
 	c.models[cacheKey(new)] = Model(new, c.db, c.dialect)
-	return nil
+	return
 }
 
 // UpdateModelAttribute alters column on a db table and runs data transformations

store/adapters/rdbms/dal/model.go

@@ -39,6 +39,34 @@ type (
 	}
 )
 
+func validate(m *dal.Model) error {
+	var (
+		c2c = make(map[string]*dal.Attribute, len(m.Attributes))
+	)
+
+	for _, a := range m.Attributes {
+		if a.StoreIdent() == "" {
+			return fmt.Errorf("attribute %q has no ident", a.Ident)
+		}
+
+		if a.Store == nil {
+			return fmt.Errorf("attribute %q has no store codec", a.Ident)
+		}
+
+		usedBy, has := c2c[a.StoreIdent()]
+		if has {
+			// column already in the map
+			if a.Store.SingleValueOnly() {
+				return fmt.Errorf("attribute %q has single value codec but column %q is already used by attribute %q", a.Ident, a.StoreIdent(), usedBy.Ident)
+			}
+		}
+
+		c2c[a.StoreIdent()] = a
+	}
+
+	return nil
+}
+
 // Model returns fully initialized model store
 //
 // It abstracts database table and its columns and provides unified interface

store/adapters/rdbms/ddl/dal_model.go

@@ -48,11 +48,11 @@ func UpdateModel(ctx context.Context, dd DataDefiner, m *dal.Model) (err error)
 		if col != nil {
 			// @todo check if column type matches
 			// @todo check if column is nullable
-			break
+			continue
 		}
 
 		// @todo add column to table
-		return fmt.Errorf("column %q on table %q not found; adding columns is not jet supported", attr.Ident, m.Ident)
+		return fmt.Errorf("column %q on table %q not found; adding columns is not yet supported", attr.Ident, m.Ident)
 	}
 
 	if err = EnsureIndexes(ctx, dd, m.Indexes...); err != nil {