Parse not null and primary key

database/config.go

@@ -9,17 +9,29 @@ import (
 
 // TableConfig holds the configuration of a table
 type TableConfig struct {
-	PrimaryKey       FieldConstraint
 	FieldConstraints []FieldConstraint
 
 	LastKey int64
 }
 
+// GetPrimaryKey returns the field constraint of the primary key.
+// Returns nil if there is no primary key.
+func (t TableConfig) GetPrimaryKey() *FieldConstraint {
+	for _, f := range t.FieldConstraints {
+		if f.IsPrimaryKey {
+			return &f
+		}
+	}
+
+	return nil
+}
+
 // FieldConstraint describes constraints on a particular field.
 type FieldConstraint struct {
-	Path    document.ValuePath
-	Type    document.ValueType
-	NotNull bool
+	Path         document.ValuePath
+	Type         document.ValueType
+	IsPrimaryKey bool
+	IsNotNull    bool
 }
 
 type tableConfigStore struct {

database/config_test.go

@@ -24,9 +24,8 @@ func TestTableConfigStore(t *testing.T) {
 	tcs := tableConfigStore{st}
 
 	cfg := TableConfig{
-		PrimaryKey: FieldConstraint{
-			Path: []string{"k"},
-			Type: document.Float64Value,
+		FieldConstraints: []FieldConstraint{
+			{Path: []string{"k"}, Type: document.Float64Value, IsPrimaryKey: true},
 		},
 		LastKey: 100,
 	}

database/table.go

@@ -77,10 +77,10 @@ func (t *Table) generateKey(d document.Document) ([]byte, error) {
 	}
 
 	var key []byte
-	if len(cfg.PrimaryKey.Path) != 0 {
-		v, err := cfg.PrimaryKey.Path.GetValue(d)
+	if pk := cfg.GetPrimaryKey(); pk != nil {
+		v, err := pk.Path.GetValue(d)
 		if err == document.ErrFieldNotFound {
-			return nil, fmt.Errorf("missing primary key at path %q", cfg.PrimaryKey.Path)
+			return nil, fmt.Errorf("missing primary key at path %q", pk.Path)
 		}
 		if err != nil {
 			return nil, err
@@ -129,7 +129,9 @@ func (t *Table) validateConstraints(d document.Document) (document.Document, err
 		return nil, err
 	}
 
-	if len(cfg.FieldConstraints) == 0 && len(cfg.PrimaryKey.Path) == 0 {
+	pk := cfg.GetPrimaryKey()
+
+	if len(cfg.FieldConstraints) == 0 && pk == nil {
 		return d, nil
 	}
 
@@ -142,15 +144,15 @@ func (t *Table) validateConstraints(d document.Document) (document.Document, err
 		return nil, err
 	}
 
-	if len(cfg.PrimaryKey.Path) != 0 {
-		err = validateConstraint(&fb, cfg.PrimaryKey)
+	if pk != nil {
+		err = validateConstraint(&fb, pk)
 		if err != nil {
 			return nil, err
 		}
 	}
 
 	for _, fc := range cfg.FieldConstraints {
-		err := validateConstraint(&fb, fc)
+		err := validateConstraint(&fb, &fc)
 		if err != nil {
 			return nil, err
 		}
@@ -159,7 +161,7 @@ func (t *Table) validateConstraints(d document.Document) (document.Document, err
 	return &fb, err
 }
 
-func validateConstraint(d document.Document, c FieldConstraint) error {
+func validateConstraint(d document.Document, c *FieldConstraint) error {
 	// get the parent buffer
 	parent, err := getParentValue(d, c.Path)
 	if err != nil {
@@ -178,7 +180,7 @@ func validateConstraint(d document.Document, c FieldConstraint) error {
 		// if the field is not found we make sure it is not required,
 		if err != nil {
 			if err == document.ErrFieldNotFound {
-				if c.NotNull {
+				if c.IsNotNull {
 					return fmt.Errorf("field %q is required and must be not null", c.Path)
 				}
 
@@ -189,6 +191,8 @@ func validateConstraint(d document.Document, c FieldConstraint) error {
 		}
 
 		// if not we convert it and replace it in the buffer
+
+		// if no type was provided, no need to convert though
 		if c.Type == 0 {
 			return nil
 		}
@@ -218,7 +222,7 @@ func validateConstraint(d document.Document, c FieldConstraint) error {
 		// if the value is not found we make sure it is not required,
 		if err != nil {
 			if err == document.ErrValueNotFound {
-				if c.NotNull {
+				if c.IsNotNull {
 					return fmt.Errorf("value %q is required and must be not null", c.Path)
 				}
 

database/table_test.go

@@ -127,9 +127,8 @@ func TestTableInsert(t *testing.T) {
 		defer cleanup()
 
 		err := tx.CreateTable("test", &database.TableConfig{
-			PrimaryKey: database.FieldConstraint{
-				Path: []string{"foo", "a", "1"},
-				Type: document.Int32Value,
+			FieldConstraints: []database.FieldConstraint{
+				{Path: []string{"foo", "a", "1"}, Type: document.Int32Value, IsPrimaryKey: true},
 			},
 		})
 		require.NoError(t, err)
@@ -189,9 +188,8 @@ func TestTableInsert(t *testing.T) {
 		defer cleanup()
 
 		err := tx.CreateTable("test", &database.TableConfig{
-			PrimaryKey: database.FieldConstraint{
-				Path: []string{"foo"},
-				Type: document.Int32Value,
+			FieldConstraints: []database.FieldConstraint{
+				{Path: []string{"foo"}, Type: document.Int32Value, IsPrimaryKey: true},
 			},
 		})
 		require.NoError(t, err)
@@ -268,8 +266,8 @@ func TestTableInsert(t *testing.T) {
 
 		err := tx.CreateTable("test", &database.TableConfig{
 			FieldConstraints: []database.FieldConstraint{
-				{[]string{"foo"}, document.Int32Value, false},
-				{[]string{"bar"}, document.Int8Value, false},
+				{[]string{"foo"}, document.Int32Value, false, false},
+				{[]string{"bar"}, document.Int8Value, false, false},
 			},
 		})
 		require.NoError(t, err)
@@ -305,7 +303,7 @@ func TestTableInsert(t *testing.T) {
 
 		err := tx.CreateTable("test1", &database.TableConfig{
 			FieldConstraints: []database.FieldConstraint{
-				{[]string{"foo"}, 0, true},
+				{[]string{"foo"}, 0, false, true},
 			},
 		})
 		require.NoError(t, err)
@@ -314,7 +312,7 @@ func TestTableInsert(t *testing.T) {
 
 		err = tx.CreateTable("test2", &database.TableConfig{
 			FieldConstraints: []database.FieldConstraint{
-				{[]string{"foo"}, document.Int32Value, true},
+				{[]string{"foo"}, document.Int32Value, false, true},
 			},
 		})
 		require.NoError(t, err)
@@ -343,7 +341,7 @@ func TestTableInsert(t *testing.T) {
 
 		err := tx.CreateTable("test1", &database.TableConfig{
 			FieldConstraints: []database.FieldConstraint{
-				{[]string{"foo", "1"}, 0, true},
+				{[]string{"foo", "1"}, 0, false, true},
 			},
 		})
 		require.NoError(t, err)

sql/parser/create.go

@@ -1,6 +1,8 @@
 package parser
 
 import (
+	"fmt"
+
 	"github.com/asdine/genji/database"
 	"github.com/asdine/genji/sql/query"
 	"github.com/asdine/genji/sql/scanner"
@@ -92,25 +94,14 @@ func (p *Parser) parseTableConfig(cfg *database.TableConfig) error {
 			break
 		}
 
-		fc.Type, err = p.parseType()
+		fc.Type = p.parseType()
+
+		err = p.parseFieldConstraint(&fc)
 		if err != nil {
 			return err
 		}
 
-		// Parse "PRIMARY"
-		if tok, _, _ := p.ScanIgnoreWhitespace(); tok == scanner.PRIMARY {
-			// Parse "KEY"
-			if tok, pos, lit := p.ScanIgnoreWhitespace(); tok != scanner.KEY {
-				return newParseError(scanner.Tokstr(tok, lit), []string{"KEY"}, pos)
-			}
-			if len(cfg.PrimaryKey.Path) != 0 {
-				return &ParseError{Message: "only one primary key is allowed"}
-			}
-			cfg.PrimaryKey = fc
-		} else {
-			p.Unscan()
-			cfg.FieldConstraints = append(cfg.FieldConstraints, fc)
-		}
+		cfg.FieldConstraints = append(cfg.FieldConstraints, fc)
 
 		if tok, _, _ := p.ScanIgnoreWhitespace(); tok != scanner.COMMA {
 			p.Unscan()
@@ -123,9 +114,55 @@ func (p *Parser) parseTableConfig(cfg *database.TableConfig) error {
 		return newParseError(scanner.Tokstr(tok, lit), []string{")"}, pos)
 	}
 
+	// ensure only one primary key
+	var pkCount int
+	for _, fc := range cfg.FieldConstraints {
+		if fc.IsPrimaryKey {
+			pkCount++
+		}
+	}
+	if pkCount > 1 {
+		return &ParseError{Message: fmt.Sprintf("only one primary key is allowed, got %d", pkCount)}
+	}
+
 	return nil
 }
 
+func (p *Parser) parseFieldConstraint(fc *database.FieldConstraint) error {
+	for {
+		tok, pos, lit := p.ScanIgnoreWhitespace()
+		switch tok {
+		case scanner.PRIMARY:
+			// Parse "KEY"
+			if tok, pos, lit := p.ScanIgnoreWhitespace(); tok != scanner.KEY {
+				return newParseError(scanner.Tokstr(tok, lit), []string{"KEY"}, pos)
+			}
+
+			// if it's already a primary key we return an error
+			if fc.IsPrimaryKey {
+				return newParseError(scanner.Tokstr(tok, lit), []string{"CONSTRAINT", ")"}, pos)
+			}
+
+			fc.IsPrimaryKey = true
+		case scanner.NOT:
+			// Parse "NULL"
+			if tok, pos, lit := p.ScanIgnoreWhitespace(); tok != scanner.NULL {
+				return newParseError(scanner.Tokstr(tok, lit), []string{"NULL"}, pos)
+			}
+
+			// if it's already not null we return an error
+			if fc.IsNotNull {
+				return newParseError(scanner.Tokstr(tok, lit), []string{"CONSTRAINT", ")"}, pos)
+			}
+
+			fc.IsNotNull = true
+		default:
+			p.Unscan()
+			return nil
+		}
+	}
+}
+
 // parseCreateIndexStatement parses a create index string and returns a Statement AST object.
 // This function assumes the CREATE INDEX or CREATE UNIQUE INDEX tokens have already been consumed.
 func (p *Parser) parseCreateIndexStatement(unique bool) (query.CreateIndexStmt, error) {

sql/parser/create_test.go

@@ -22,20 +22,73 @@ func TestParserCreateTable(t *testing.T) {
 			query.CreateTableStmt{
 				TableName: "test",
 				Config: database.TableConfig{
-					PrimaryKey: database.FieldConstraint{Path: []string{"foo"}, Type: document.Int64Value},
+					FieldConstraints: []database.FieldConstraint{
+						{Path: []string{"foo"}, Type: document.Int64Value, IsPrimaryKey: true},
+					},
+				},
+			}, false},
+		{"With primary key twice", "CREATE TABLE test(foo PRIMARY KEY PRIMARY KEY)",
+			query.CreateTableStmt{}, true},
+		{"With type", "CREATE TABLE test(foo INT)",
+			query.CreateTableStmt{
+				TableName: "test",
+				Config: database.TableConfig{
+					FieldConstraints: []database.FieldConstraint{
+						{Path: []string{"foo"}, Type: document.Int64Value},
+					},
+				},
+			}, false},
+		{"With not null", "CREATE TABLE test(foo NOT NULL)",
+			query.CreateTableStmt{
+				TableName: "test",
+				Config: database.TableConfig{
+					FieldConstraints: []database.FieldConstraint{
+						{Path: []string{"foo"}, IsNotNull: true},
+					},
+				},
+			}, false},
+		{"With not null twice", "CREATE TABLE test(foo NOT NULL NOT NULL)",
+			query.CreateTableStmt{}, true},
+		{"With type and not null", "CREATE TABLE test(foo INT NOT NULL)",
+			query.CreateTableStmt{
+				TableName: "test",
+				Config: database.TableConfig{
+					FieldConstraints: []database.FieldConstraint{
+						{Path: []string{"foo"}, Type: document.Int64Value, IsNotNull: true},
+					},
+				},
+			}, false},
+		{"With not null and primary key", "CREATE TABLE test(foo INT NOT NULL PRIMARY KEY)",
+			query.CreateTableStmt{
+				TableName: "test",
+				Config: database.TableConfig{
+					FieldConstraints: []database.FieldConstraint{
+						{Path: []string{"foo"}, Type: document.Int64Value, IsPrimaryKey: true, IsNotNull: true},
+					},
+				},
+			}, false},
+		{"With primary key and not null", "CREATE TABLE test(foo INT PRIMARY KEY NOT NULL)",
+			query.CreateTableStmt{
+				TableName: "test",
+				Config: database.TableConfig{
+					FieldConstraints: []database.FieldConstraint{
+						{Path: []string{"foo"}, Type: document.Int64Value, IsPrimaryKey: true, IsNotNull: true},
+					},
 				},
 			}, false},
-		{"With multiple constraints", "CREATE TABLE test(foo INT PRIMARY KEY, bar INT16, baz.4.1.bat STRING)",
+		{"With multiple constraints", "CREATE TABLE test(foo INT PRIMARY KEY, bar INT16 NOT NULL, baz.4.1.bat STRING)",
 			query.CreateTableStmt{
 				TableName: "test",
 				Config: database.TableConfig{
-					PrimaryKey: database.FieldConstraint{Path: []string{"foo"}, Type: document.Int64Value},
 					FieldConstraints: []database.FieldConstraint{
-						{Path: []string{"bar"}, Type: document.Int16Value},
+						{Path: []string{"foo"}, Type: document.Int64Value, IsPrimaryKey: true},
+						{Path: []string{"bar"}, Type: document.Int16Value, IsNotNull: true},
 						{Path: []string{"baz", "4", "1", "bat"}, Type: document.TextValue},
 					},
 				},
 			}, false},
+		{"With multiple primary keys", "CREATE TABLE test(foo PRIMARY KEY, bar PRIMARY KEY)",
+			query.CreateTableStmt{}, true},
 	}
 
 	for _, test := range tests {