Repository: genjidb/genji (go)

Issue:
Prevent creating a table if constraints are incoherent
```sql
CREATE TABLE foo (
  a INTEGER,
  a.b[0] TEXT
);
```

In this example, `a` is expected to be both an integer an a document, which is not possible. Detect these kind of incoherence and return an error.

Pull request:
Add constraint checking when creating table
Fixes #222.

Patch:
diff --git a/sql/query/constraint.go b/sql/query/constraint.go
new file mode 100644
index 00000000..04e1a683
--- /dev/null
+++ b/sql/query/constraint.go
@@ -0,0 +1,142 @@
+package query
+
+import (
+	"fmt"
+	"github.com/genjidb/genji/database"
+
+	"github.com/genjidb/genji/document"
+)
+
+// constraintNode is a tree node which stores a type of document field
+type constraintNode struct {
+	frag   document.ValuePathFragment
+	typ    document.ValueType
+	parent *constraintNode
+	sub    []*constraintNode
+}
+
+func createConstraintNode(parent *constraintNode, path document.ValuePath, typ document.ValueType) *constraintNode {
+	node := &constraintNode{
+		frag:   path[0],
+		parent: parent,
+	}
+
+	if len(path) == 1 {
+		node.typ = typ
+		return node
+	}
+
+	// Here we check next path fragment
+	// If next is index, current node is array
+	// Otherwise it is a document
+	if path[1].FieldName != "" {
+		node.typ = document.DocumentValue
+	} else {
+		node.typ = document.ArrayValue
+	}
+
+	node.sub = append(node.sub, createConstraintNode(node, path[1:], typ))
+	return node
+}
+
+func (n *constraintNode) getPath() document.ValuePath {
+	if n.parent == nil {
+		return document.ValuePath{n.frag}
+	}
+
+	return append(n.parent.getPath(), n.frag)
+}
+
+func (n *constraintNode) search(path document.ValuePath) *constraintNode {
+	if n.frag != path[0] {
+		return nil
+	}
+
+	if len(path) == 1 {
+		return n
+	}
+
+	for _, sub := range n.sub {
+		t := sub.search(path[1:])
+		if t != nil {
+			return t
+		}
+	}
+
+	return nil
+}
+
+func (n *constraintNode) insert(path document.ValuePath, typ document.ValueType) error {
+	switch {
+	case len(path) == 1:
+		return fmt.Errorf("%q already exists as type %s", n.getPath().String(), n.typ.String())
+	// when type is explicitly set and does not have array or document type
+	case n.typ != 0 && n.typ != document.ArrayValue && n.typ != document.DocumentValue:
+		p := append(n.getPath(), path[1]).String()
+		return fmt.Errorf("%q already exists as type %s, but trying add %q constraint", n.getPath().String(), n.typ.String(), p)
+	// when constraint tries to set document constraint for path, but there is already a array constraint
+	case path[1].FieldName != "" && n.typ == document.ArrayValue:
+		p := append(n.getPath(), path[1]).String()
+		return fmt.Errorf("%q already exists as array, but trying add %q constraint", n.getPath().String(), p)
+	// when constraint tries to set array constraint for path, but there is already a document constraint
+	case path[1].FieldName == "" && n.typ == document.DocumentValue:
+		p := append(n.getPath(), path[1]).String()
+		return fmt.Errorf("%q already exists as document, but trying add %q constraint", n.getPath().String(), p)
+	}
+
+	for _, sub := range n.sub {
+		if sub.frag == path[1] {
+			return sub.insert(path[1:], typ)
+		}
+	}
+
+	n.sub = append(n.sub, createConstraintNode(n, path[1:], typ))
+	return nil
+}
+
+// constraintTree is a tree of document field types.
+// Field type can be set explicitly or derived from path.
+// Example query:
+// 	CREATE TABLE foo(a.b TEXT, a.d[1])
+// as a tree
+// 	document(a)
+// 	├── text(b)
+// 	├── array(d)
+// 	│   ├──[1] any
+// 	│   └──...
+//
+type constraintTree struct {
+	roots []*constraintNode
+}
+
+func (tree *constraintTree) insert(path document.ValuePath, typ document.ValueType) error {
+	for _, sub := range tree.roots {
+		if sub.frag == path[0] {
+			return sub.insert(path, typ)
+		}
+	}
+
+	tree.roots = append(tree.roots, createConstraintNode(nil, path, typ))
+	return nil
+}
+
+func (tree *constraintTree) search(path document.ValuePath) *constraintNode {
+	for _, sub := range tree.roots {
+		if sub.frag == path[0] {
+			return sub.search(path)
+		}
+	}
+
+	return nil
+}
+
+func checkConstraints(constraints []database.FieldConstraint) error {
+	tree := constraintTree{}
+	for _, fc := range constraints {
+		if err := tree.insert(fc.Path, fc.Type); err != nil {
+			return fmt.Errorf("incoherent field constraint: %w", err)
+		}
+	}
+
+	return nil
+}
diff --git a/sql/query/create.go b/sql/query/create.go
index 764fae35..9110a605 100644
--- a/sql/query/create.go
+++ b/sql/query/create.go
@@ -30,7 +30,12 @@ func (stmt CreateTableStmt) Run(ctx context.Context, tx *database.Transaction, a
 		return res, errors.New("missing table name")
 	}
 
-	err := tx.CreateTable(stmt.TableName, &stmt.Info)
+	err := checkConstraints(stmt.Info.FieldConstraints)
+	if err != nil {
+		return res, err
+	}
+
+	err = tx.CreateTable(stmt.TableName, &stmt.Info)
 	if stmt.IfNotExists && err == database.ErrTableAlreadyExists {
 		err = nil
 	}


Test patch:
diff --git a/sql/query/create_test.go b/sql/query/create_test.go
index 95f465f1..c605b763 100644
--- a/sql/query/create_test.go
+++ b/sql/query/create_test.go
@@ -32,6 +32,16 @@ func TestCreateTable(t *testing.T) {
 		{"With primary key", "CREATE TABLE test(foo TEXT PRIMARY KEY)", false},
 		{"With field constraints", "CREATE TABLE test(foo.a[1][2] TEXT primary key, bar[4][0].bat INTEGER not null, baz not null)", false},
 		{"With no constraints", "CREATE TABLE test(a, b)", false},
+		{"With coherent constraint(common)", "CREATE TABLE test(a DOCUMENT, a.b ARRAY, a.b[0] TEXT);", false},
+		{"With coherent constraint(any)", "CREATE TABLE test(a, a.b[0] TEXT);", false},
+		{"With coherent constraint(document)", "CREATE TABLE test(a DOCUMENT, a.b TEXT);", false},
+		{"With coherent constraint(array)", "CREATE TABLE test(a ARRAY, a[0] TEXT);", false},
+		{"With incoherent constraint(common)", "CREATE TABLE test(a INTEGER, a.b[0] TEXT);", true},
+		{"With incoherent constraint(common)", "CREATE TABLE test(a DOCUMENT, a.b[0] TEXT, a.b.c TEXT);", true},
+		{"With incoherent constraint(common)", "CREATE TABLE test(a DOCUMENT, a.b.c TEXT, a.b[0] TEXT);", true},
+		{"With incoherent constraint(document)", "CREATE TABLE test(a INTEGER, a.b TEXT);", true},
+		{"With incoherent constraint(array)", "CREATE TABLE test(a INTEGER, a[0] TEXT);", true},
+		{"With duplicate constraints", "CREATE TABLE test(a INTEGER, a TEXT);", true},
 	}
 
 	for _, test := range tests {