Repository: genjidb/genji (go)

Issue:
where condition do not match on indexed array field (equality comparison) 
<!-- Please answer these questions before submitting your issue. Thanks! -->

### What version of Genji are you using?

Latest commit on the main branch

<pre>
$ genji version
Genji v0.11.0
Genji CLI (devel)
</pre>

### Does this issue reproduce with the latest release?

Yes

### What did you do?

```
~› genji
Opened an in-memory database.
Enter ".help" for usage hints.
genji> CREATE TABLE foo;
genji> CREATE INDEX idx_foo_a ON foo(a);
genji> INSERT INTO foo (a) VALUES ([1,2,3]);
genji> INSERT INTO foo (a) VALUES ([4,5,6]);
genji> SELECT * from foo;
{
  "a": [
    1,
    2,
    3
  ]
}
{
  "a": [
    4,
    5,
    6
  ]
}
genji> EXPLAIN SELECT * FROM foo WHERE a = [1,2,3];
{
  "plan": "indexScan(\"idx_foo_a\", [1, 2, 3])"
}
genji> SELECT * FROM foo WHERE a = [1,2,3];
genji> 
```

### What did you expect to see?

```
genji> SELECT * FROM foo WHERE a = [1,2,3];
{
  "a": [
    1,
    2,
    3
  ]
}
genji> 
```

### What did you see instead?

genji> SELECT * FROM foo WHERE a = [1,2,3];
genji> 

### What Go version and environment are you using?

<pre>
$ go version
go version go1.15.8 darwin/amd64
</pre>

<details><summary><code>go env</code> Output</summary><br><pre>
$ go env
GO111MODULE="on"
GOARCH="amd64"
GOBIN="/Users/tech/code/bin"
GOCACHE="/Users/tech/Library/Caches/go-build"
GOENV="/Users/tech/Library/Application Support/go/env"
GOEXE=""
GOFLAGS=""
GOHOSTARCH="amd64"
GOHOSTOS="darwin"
GOINSECURE=""
GOMODCACHE="/Users/tech/code/pkg/mod"
GONOPROXY=""
GONOSUMDB=""
GOOS="darwin"
GOPATH="/Users/tech/code"
GOPRIVATE=""
GOPROXY="https://proxy.golang.org,direct"
GOROOT="/Users/tech/.asdf/installs/golang/1.15.8/go"
GOSUMDB="sum.golang.org"
GOTMPDIR=""
GOTOOLDIR="/Users/tech/.asdf/installs/golang/1.15.8/go/pkg/tool/darwin_amd64"
GCCGO="gccgo"
AR="ar"
CC="clang"
CXX="clang++"
CGO_ENABLED="1"
GOMOD="/dev/null"
CGO_CFLAGS="-g -O2"
CGO_CPPFLAGS=""
CGO_CXXFLAGS="-g -O2"
CGO_FFLAGS="-g -O2"
CGO_LDFLAGS="-g -O2"
PKG_CONFIG="pkg-config"
GOGCCFLAGS="-fPIC -m64 -pthread -fno-caret-diagnostics -Qunused-arguments -fmessage-length=0 -fdebug-prefix-map=/var/folders/xl/dfsx6nzs4cb2y40m555mlfjh0000gn/T/go-build752488097=/tmp/go-build -gno-record-gcc-switches -fno-common"

</pre></details>


Pull request:
Number conversion
The planner now does a better job at selecting nodes that match an index by converting operands recursively using field constraints.

Fixes #359 #361 

Patch:
diff --git a/database/catalog.go b/database/catalog.go
index 0ac12b15..cc500428 100644
--- a/database/catalog.go
+++ b/database/catalog.go
@@ -42,6 +42,7 @@ func (c *Catalog) Load(tx *Transaction) error {
 						FieldName: "table_name",
 					},
 				},
+				Type:         document.TextValue,
 				IsPrimaryKey: true,
 			},
 		},
@@ -58,6 +59,7 @@ func (c *Catalog) Load(tx *Transaction) error {
 						FieldName: "index_name",
 					},
 				},
+				Type:         document.TextValue,
 				IsPrimaryKey: true,
 			},
 		},
diff --git a/database/config.go b/database/config.go
index 44794903..e6d81370 100644
--- a/database/config.go
+++ b/database/config.go
@@ -3,7 +3,6 @@ package database
 import (
 	"bytes"
 	"encoding/binary"
-	"strings"
 
 	"github.com/genjidb/genji/document"
 	"github.com/genjidb/genji/engine"
@@ -12,389 +11,6 @@ import (
 
 const storePrefix = 't'
 
-// FieldConstraint describes constraints on a particular field.
-type FieldConstraint struct {
-	Path         document.Path
-	Type         document.ValueType
-	IsPrimaryKey bool
-	IsNotNull    bool
-	DefaultValue document.Value
-	IsInferred   bool
-	InferredBy   []document.Path
-}
-
-// IsEqual compares f with other member by member.
-// Inference is not compared.
-func (f *FieldConstraint) IsEqual(other *FieldConstraint) (bool, error) {
-	if !f.Path.IsEqual(other.Path) {
-		return false, nil
-	}
-
-	if f.Type != other.Type {
-		return false, nil
-	}
-
-	if f.IsPrimaryKey != other.IsPrimaryKey {
-		return false, nil
-	}
-
-	if f.IsNotNull != other.IsNotNull {
-		return false, nil
-	}
-
-	if f.HasDefaultValue() != other.HasDefaultValue() {
-		return false, nil
-	}
-
-	if f.HasDefaultValue() {
-		if ok, err := f.DefaultValue.IsEqual(other.DefaultValue); !ok || err != nil {
-			return ok, err
-		}
-	}
-
-	return true, nil
-}
-
-func (f *FieldConstraint) String() string {
-	var s strings.Builder
-
-	s.WriteString(f.Path.String())
-	s.WriteString(" ")
-	s.WriteString(f.Type.String())
-	if f.IsNotNull {
-		s.WriteString(" NOT NULL")
-	}
-	if f.IsPrimaryKey {
-		s.WriteString(" PRIMARY KEY")
-	}
-
-	if f.HasDefaultValue() {
-		s.WriteString(" DEFAULT ")
-		s.WriteString(f.DefaultValue.String())
-	}
-
-	return s.String()
-}
-
-// MergeInferred adds the other.InferredBy to f.InferredBy and ensures there are no duplicates.
-func (f *FieldConstraint) MergeInferred(other *FieldConstraint) {
-	for _, by := range other.InferredBy {
-		duplicate := false
-		for _, fby := range f.InferredBy {
-			if fby.IsEqual(by) {
-				duplicate = true
-				break
-			}
-		}
-
-		if !duplicate {
-			f.InferredBy = append(f.InferredBy, by)
-		}
-	}
-}
-
-// HasDefaultValue returns this field contains a default value constraint.
-func (f *FieldConstraint) HasDefaultValue() bool {
-	return f.DefaultValue.Type != 0
-}
-
-// ToDocument returns a document from f.
-func (f *FieldConstraint) ToDocument() document.Document {
-	buf := document.NewFieldBuffer()
-
-	buf.Add("path", document.NewArrayValue(pathToArray(f.Path)))
-	buf.Add("type", document.NewIntegerValue(int64(f.Type)))
-	buf.Add("is_primary_key", document.NewBoolValue(f.IsPrimaryKey))
-	buf.Add("is_not_null", document.NewBoolValue(f.IsNotNull))
-	if f.HasDefaultValue() {
-		buf.Add("default_value", f.DefaultValue)
-	}
-	buf.Add("is_inferred", document.NewBoolValue(f.IsInferred))
-	if f.IsInferred {
-		vb := document.NewValueBuffer()
-		for _, by := range f.InferredBy {
-			vb = vb.Append(document.NewArrayValue(pathToArray(by)))
-		}
-		buf.Add("inferred_by", document.NewArrayValue(vb))
-	}
-	return buf
-}
-
-// ScanDocument implements the document.Scanner interface.
-func (f *FieldConstraint) ScanDocument(d document.Document) error {
-	v, err := d.GetByField("path")
-	if err != nil {
-		return err
-	}
-	f.Path, err = arrayToPath(v.V.(document.Array))
-	if err != nil {
-		return err
-	}
-
-	v, err = d.GetByField("type")
-	if err != nil {
-		return err
-	}
-	tp := v.V.(int64)
-	f.Type = document.ValueType(tp)
-
-	v, err = d.GetByField("is_primary_key")
-	if err != nil {
-		return err
-	}
-	f.IsPrimaryKey = v.V.(bool)
-
-	v, err = d.GetByField("is_not_null")
-	if err != nil {
-		return err
-	}
-	f.IsNotNull = v.V.(bool)
-
-	v, err = d.GetByField("default_value")
-	if err != nil && err != document.ErrFieldNotFound {
-		return err
-	}
-	if err == nil {
-		f.DefaultValue = v
-	}
-
-	v, err = d.GetByField("is_inferred")
-	if err != nil && err != document.ErrFieldNotFound {
-		return err
-	}
-	if err == nil {
-		f.IsInferred = v.V.(bool)
-	}
-
-	v, err = d.GetByField("inferred_by")
-	if err != nil && err != document.ErrFieldNotFound {
-		return err
-	}
-	if err == nil {
-		v.V.(document.Array).Iterate(func(i int, value document.Value) error {
-			by, err := arrayToPath(value.V.(document.Array))
-			if err != nil {
-				return err
-			}
-			f.InferredBy = append(f.InferredBy, by)
-			return nil
-		})
-	}
-
-	return nil
-}
-
-// FieldConstraints is a list of field constraints.
-type FieldConstraints []*FieldConstraint
-
-// NewFieldConstraints takes user-defined field constraints, validates them, infers additional
-// constraints if needed, and returns a valid FieldConstraints type that can be assigned to a table.
-func NewFieldConstraints(userConstraints []*FieldConstraint) (FieldConstraints, error) {
-	return FieldConstraints(userConstraints).Infer()
-}
-
-// Infer additional constraints based on user defined ones.
-// For example, given the following table:
-//   CREATE TABLE foo (a.b[0] TEXT)
-// this function will return a TableInfo that behaves as if the table
-// had been created like this:
-//   CREATE TABLE foo(
-//      a DOCUMENT
-//      a.b ARRAY
-//      a.b[0] TEXT
-//   )
-func (f FieldConstraints) Infer() (FieldConstraints, error) {
-	newConstraints := make(FieldConstraints, 0, len(f))
-
-	for _, fc := range f {
-		// loop over all the path fragments and
-		// create intermediary inferred constraints.
-		if len(fc.Path) > 1 {
-			for i := range fc.Path {
-				// stop before reaching the last fragment
-				// which will be added outside of this loop
-				if i+1 == len(fc.Path) {
-					break
-				}
-
-				newFc := FieldConstraint{
-					Path:       fc.Path[:i+1],
-					IsInferred: true,
-					InferredBy: []document.Path{fc.Path},
-				}
-				if fc.Path[i+1].FieldName != "" {
-					newFc.Type = document.DocumentValue
-				} else {
-					newFc.Type = document.ArrayValue
-				}
-
-				err := newConstraints.Add(&newFc)
-				if err != nil {
-					return nil, err
-				}
-			}
-		}
-
-		// add the non inferred path to the list
-		// and ensure there are no conflicts with
-		// existing ones.
-		err := newConstraints.Add(fc)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	return newConstraints, nil
-}
-
-// Add a field constraint to the list. If another constraint exists for the same path
-// and they are equal, newFc will be ignored. Otherwise an error will be returned.
-// If newFc has been inferred by another constraint and another constraint exists with the same
-// path, their InferredBy member will be merged.
-func (f *FieldConstraints) Add(newFc *FieldConstraint) error {
-	for i, c := range *f {
-		if c.Path.IsEqual(newFc.Path) {
-			ok, err := c.IsEqual(newFc)
-			if err != nil {
-				return err
-			}
-
-			// if constraints are different
-			if !ok {
-				return stringutil.Errorf("conflicting constraints: %q and %q", c.String(), newFc.String())
-			}
-
-			// if both non inferred, they are duplicate
-			if !newFc.IsInferred && !c.IsInferred {
-				return stringutil.Errorf("conflicting constraints: %q and %q", c.String(), newFc.String())
-			}
-
-			// if both inferred, merge the InferredBy member
-			if newFc.IsInferred && c.IsInferred {
-				c.MergeInferred(newFc)
-				return nil
-			}
-
-			// if existing one is not inferred, ignore newFc
-			if newFc.IsInferred && !c.IsInferred {
-				return nil
-			}
-
-			// if existing one is inferred, and newFc is not,
-			// replace it
-			(*f)[i] = newFc
-			return nil
-		}
-
-		// ensure we don't have duplicate primary keys
-		if c.IsPrimaryKey && newFc.IsPrimaryKey {
-			return stringutil.Errorf(
-				"multiple primary keys are not allowed (%q is primary key)",
-				c.Path.String(),
-			)
-		}
-	}
-
-	// convert default values to the right types
-	targetType := newFc.Type
-
-	// if there is no type constraint, numbers must be converted to double
-	if newFc.DefaultValue.Type == document.IntegerValue && newFc.Type == 0 {
-		targetType = document.DoubleValue
-	}
-	if newFc.DefaultValue.Type != 0 && targetType != 0 {
-		var err error
-		newFc.DefaultValue, err = newFc.DefaultValue.CastAs(targetType)
-		if err != nil {
-			return err
-		}
-	}
-
-	*f = append(*f, newFc)
-	return nil
-}
-
-// ValidateDocument calls Convert then ensures the document validates against the field constraints.
-func (f FieldConstraints) ValidateDocument(d document.Document) (*document.FieldBuffer, error) {
-	fb, err := f.ConvertDocument(d)
-	if err != nil {
-		return nil, err
-	}
-
-	// ensure no field is missing
-	for _, fc := range f {
-		v, err := fc.Path.GetValueFromDocument(fb)
-		if err == nil {
-			// if field is found, it has already been converted
-			// to the right type above.
-			// check if it is required but null.
-			if v.Type == document.NullValue && fc.IsNotNull {
-				return nil, stringutil.Errorf("field %q is required and must be not null", fc.Path)
-			}
-			continue
-		}
-
-		if err != document.ErrFieldNotFound {
-			return nil, err
-		}
-
-		// if field is not found
-		// check if there is a default value
-		if fc.DefaultValue.Type != 0 {
-			err = fb.Set(fc.Path, fc.DefaultValue)
-			if err != nil {
-				return nil, err
-			}
-			// if there is no default value
-			// check if field is required
-		} else if fc.IsNotNull {
-			return nil, stringutil.Errorf("field %q is required and must be not null", fc.Path)
-		}
-	}
-
-	return fb, nil
-}
-
-// ConvertDocument the document using the field constraints.
-// It converts any path that has a field constraint on it into the specified type.
-// If there is no constraint on an integer field or value, it converts it into a double.
-// Default values on missing fields are not applied.
-func (f FieldConstraints) ConvertDocument(d document.Document) (*document.FieldBuffer, error) {
-	fb := document.NewFieldBuffer()
-	err := fb.Copy(d)
-	if err != nil {
-		return nil, err
-	}
-
-	// convert the document using field constraints type information.
-	// if there is a type constraint on a path, apply it.
-	// if a value is an integer and has no constraint, convert it to double.
-	err = fb.Apply(func(p document.Path, v document.Value) (document.Value, error) {
-		for _, fc := range f {
-			if !fc.Path.IsEqual(p) {
-				continue
-			}
-
-			// check if the constraint enforce a particular type
-			// and if so convert the value to the new type.
-			if fc.Type != 0 {
-				return v.CastAs(fc.Type)
-			}
-			break
-		}
-
-		// no constraint have been found for this path.
-		// check if this is an integer and convert it to double.
-		if v.Type == document.IntegerValue {
-			return v.CastAsDouble()
-		}
-
-		return v, nil
-	})
-
-	return fb, err
-}
-
 // TableInfo contains information about a table.
 type TableInfo struct {
 	// name of the table.
diff --git a/database/constraint.go b/database/constraint.go
new file mode 100644
index 00000000..a45aea10
--- /dev/null
+++ b/database/constraint.go
@@ -0,0 +1,476 @@
+package database
+
+import (
+	"strings"
+
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/stringutil"
+)
+
+// FieldConstraint describes constraints on a particular field.
+type FieldConstraint struct {
+	Path         document.Path
+	Type         document.ValueType
+	IsPrimaryKey bool
+	IsNotNull    bool
+	DefaultValue document.Value
+	IsInferred   bool
+	InferredBy   []document.Path
+}
+
+// IsEqual compares f with other member by member.
+// Inference is not compared.
+func (f *FieldConstraint) IsEqual(other *FieldConstraint) (bool, error) {
+	if !f.Path.IsEqual(other.Path) {
+		return false, nil
+	}
+
+	if f.Type != other.Type {
+		return false, nil
+	}
+
+	if f.IsPrimaryKey != other.IsPrimaryKey {
+		return false, nil
+	}
+
+	if f.IsNotNull != other.IsNotNull {
+		return false, nil
+	}
+
+	if f.HasDefaultValue() != other.HasDefaultValue() {
+		return false, nil
+	}
+
+	if f.HasDefaultValue() {
+		if ok, err := f.DefaultValue.IsEqual(other.DefaultValue); !ok || err != nil {
+			return ok, err
+		}
+	}
+
+	return true, nil
+}
+
+func (f *FieldConstraint) String() string {
+	var s strings.Builder
+
+	s.WriteString(f.Path.String())
+	s.WriteString(" ")
+	s.WriteString(f.Type.String())
+	if f.IsNotNull {
+		s.WriteString(" NOT NULL")
+	}
+	if f.IsPrimaryKey {
+		s.WriteString(" PRIMARY KEY")
+	}
+
+	if f.HasDefaultValue() {
+		s.WriteString(" DEFAULT ")
+		s.WriteString(f.DefaultValue.String())
+	}
+
+	return s.String()
+}
+
+// MergeInferred adds the other.InferredBy to f.InferredBy and ensures there are no duplicates.
+func (f *FieldConstraint) MergeInferred(other *FieldConstraint) {
+	for _, by := range other.InferredBy {
+		duplicate := false
+		for _, fby := range f.InferredBy {
+			if fby.IsEqual(by) {
+				duplicate = true
+				break
+			}
+		}
+
+		if !duplicate {
+			f.InferredBy = append(f.InferredBy, by)
+		}
+	}
+}
+
+// HasDefaultValue returns this field contains a default value constraint.
+func (f *FieldConstraint) HasDefaultValue() bool {
+	return f.DefaultValue.Type != 0
+}
+
+// ToDocument returns a document from f.
+func (f *FieldConstraint) ToDocument() document.Document {
+	buf := document.NewFieldBuffer()
+
+	buf.Add("path", document.NewArrayValue(pathToArray(f.Path)))
+	buf.Add("type", document.NewIntegerValue(int64(f.Type)))
+	buf.Add("is_primary_key", document.NewBoolValue(f.IsPrimaryKey))
+	buf.Add("is_not_null", document.NewBoolValue(f.IsNotNull))
+	if f.HasDefaultValue() {
+		buf.Add("default_value", f.DefaultValue)
+	}
+	buf.Add("is_inferred", document.NewBoolValue(f.IsInferred))
+	if f.IsInferred {
+		vb := document.NewValueBuffer()
+		for _, by := range f.InferredBy {
+			vb = vb.Append(document.NewArrayValue(pathToArray(by)))
+		}
+		buf.Add("inferred_by", document.NewArrayValue(vb))
+	}
+	return buf
+}
+
+// ScanDocument implements the document.Scanner interface.
+func (f *FieldConstraint) ScanDocument(d document.Document) error {
+	v, err := d.GetByField("path")
+	if err != nil {
+		return err
+	}
+	f.Path, err = arrayToPath(v.V.(document.Array))
+	if err != nil {
+		return err
+	}
+
+	v, err = d.GetByField("type")
+	if err != nil {
+		return err
+	}
+	tp := v.V.(int64)
+	f.Type = document.ValueType(tp)
+
+	v, err = d.GetByField("is_primary_key")
+	if err != nil {
+		return err
+	}
+	f.IsPrimaryKey = v.V.(bool)
+
+	v, err = d.GetByField("is_not_null")
+	if err != nil {
+		return err
+	}
+	f.IsNotNull = v.V.(bool)
+
+	v, err = d.GetByField("default_value")
+	if err != nil && err != document.ErrFieldNotFound {
+		return err
+	}
+	if err == nil {
+		f.DefaultValue = v
+	}
+
+	v, err = d.GetByField("is_inferred")
+	if err != nil && err != document.ErrFieldNotFound {
+		return err
+	}
+	if err == nil {
+		f.IsInferred = v.V.(bool)
+	}
+
+	v, err = d.GetByField("inferred_by")
+	if err != nil && err != document.ErrFieldNotFound {
+		return err
+	}
+	if err == nil {
+		v.V.(document.Array).Iterate(func(i int, value document.Value) error {
+			by, err := arrayToPath(value.V.(document.Array))
+			if err != nil {
+				return err
+			}
+			f.InferredBy = append(f.InferredBy, by)
+			return nil
+		})
+	}
+
+	return nil
+}
+
+// FieldConstraints is a list of field constraints.
+type FieldConstraints []*FieldConstraint
+
+// NewFieldConstraints takes user-defined field constraints, validates them, infers additional
+// constraints if needed, and returns a valid FieldConstraints type that can be assigned to a table.
+func NewFieldConstraints(userConstraints []*FieldConstraint) (FieldConstraints, error) {
+	return FieldConstraints(userConstraints).Infer()
+}
+
+// Infer additional constraints based on user defined ones.
+// For example, given the following table:
+//   CREATE TABLE foo (a.b[0] TEXT)
+// this function will return a TableInfo that behaves as if the table
+// had been created like this:
+//   CREATE TABLE foo(
+//      a DOCUMENT
+//      a.b ARRAY
+//      a.b[0] TEXT
+//   )
+func (f FieldConstraints) Infer() (FieldConstraints, error) {
+	newConstraints := make(FieldConstraints, 0, len(f))
+
+	for _, fc := range f {
+		// loop over all the path fragments and
+		// create intermediary inferred constraints.
+		if len(fc.Path) > 1 {
+			for i := range fc.Path {
+				// stop before reaching the last fragment
+				// which will be added outside of this loop
+				if i+1 == len(fc.Path) {
+					break
+				}
+
+				newFc := FieldConstraint{
+					Path:       fc.Path[:i+1],
+					IsInferred: true,
+					InferredBy: []document.Path{fc.Path},
+				}
+				if fc.Path[i+1].FieldName != "" {
+					newFc.Type = document.DocumentValue
+				} else {
+					newFc.Type = document.ArrayValue
+				}
+
+				err := newConstraints.Add(&newFc)
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+
+		// add the non inferred path to the list
+		// and ensure there are no conflicts with
+		// existing ones.
+		err := newConstraints.Add(fc)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return newConstraints, nil
+}
+
+// Add a field constraint to the list. If another constraint exists for the same path
+// and they are equal, newFc will be ignored. Otherwise an error will be returned.
+// If newFc has been inferred by another constraint and another constraint exists with the same
+// path, their InferredBy member will be merged.
+func (f *FieldConstraints) Add(newFc *FieldConstraint) error {
+	for i, c := range *f {
+		if c.Path.IsEqual(newFc.Path) {
+			// if both non inferred, they are duplicate
+			if !newFc.IsInferred && !c.IsInferred {
+				return stringutil.Errorf("conflicting constraints: %q and %q", c.String(), newFc.String())
+			}
+
+			// determine which one is inferred
+			inferredFc, nonInferredFc := c, newFc
+			if newFc.IsInferred {
+				inferredFc, nonInferredFc = nonInferredFc, inferredFc
+			}
+
+			// the inferred one may have less constraints that the user-defined one
+			inferredFc.DefaultValue = nonInferredFc.DefaultValue
+			inferredFc.IsNotNull = nonInferredFc.IsNotNull
+			inferredFc.IsPrimaryKey = nonInferredFc.IsPrimaryKey
+
+			// safe-guard in case we add more fields to the struct
+			ok, err := c.IsEqual(newFc)
+			if err != nil {
+				return err
+			}
+
+			// if constraints are different
+			if !ok {
+				return stringutil.Errorf("conflicting constraints: %q and %q", c.String(), newFc.String())
+			}
+
+			// if both inferred, merge the InferredBy member
+			if newFc.IsInferred && c.IsInferred {
+				c.MergeInferred(newFc)
+				return nil
+			}
+
+			// if existing one is not inferred, ignore newFc
+			if newFc.IsInferred && !c.IsInferred {
+				return nil
+			}
+
+			// if existing one is inferred, and newFc is not,
+			// replace it
+			(*f)[i] = newFc
+			return nil
+		}
+
+		// ensure we don't have duplicate primary keys
+		if c.IsPrimaryKey && newFc.IsPrimaryKey {
+			return stringutil.Errorf(
+				"multiple primary keys are not allowed (%q is primary key)",
+				c.Path.String(),
+			)
+		}
+	}
+
+	// convert default values to the right types
+	targetType := newFc.Type
+
+	// if there is no type constraint, numbers must be converted to double
+	if newFc.DefaultValue.Type == document.IntegerValue && newFc.Type == 0 {
+		targetType = document.DoubleValue
+	}
+	if newFc.DefaultValue.Type != 0 && targetType != 0 {
+		var err error
+		newFc.DefaultValue, err = newFc.DefaultValue.CastAs(targetType)
+		if err != nil {
+			return err
+		}
+	}
+
+	*f = append(*f, newFc)
+	return nil
+}
+
+// ValidateDocument calls Convert then ensures the document validates against the field constraints.
+func (f FieldConstraints) ValidateDocument(d document.Document) (*document.FieldBuffer, error) {
+	fb, err := f.ConvertDocument(d)
+	if err != nil {
+		return nil, err
+	}
+
+	// ensure no field is missing
+	for _, fc := range f {
+		v, err := fc.Path.GetValueFromDocument(fb)
+		if err == nil {
+			// if field is found, it has already been converted
+			// to the right type above.
+			// check if it is required but null.
+			if v.Type == document.NullValue && fc.IsNotNull {
+				return nil, stringutil.Errorf("field %q is required and must be not null", fc.Path)
+			}
+
+			continue
+		}
+
+		if err != document.ErrFieldNotFound {
+			return nil, err
+		}
+
+		// if field is not found
+		// check if there is a default value
+		if fc.DefaultValue.Type != 0 {
+			err = fb.Set(fc.Path, fc.DefaultValue)
+			if err != nil {
+				return nil, err
+			}
+			// if there is no default value
+			// check if field is required
+		} else if fc.IsNotNull {
+			return nil, stringutil.Errorf("field %q is required and must be not null", fc.Path)
+		}
+	}
+
+	return fb, nil
+}
+
+// ConvertDocument the document using the field constraints.
+// It converts any path that has a field constraint on it into the specified type using CAST.
+// If there is no constraint on an integer field or value, it converts it into a double.
+// Default values on missing fields are not applied.
+func (f FieldConstraints) ConvertDocument(d document.Document) (*document.FieldBuffer, error) {
+	return f.convertDocumentAtPath(nil, d, CastConversion)
+}
+
+// ConversionFunc is called when the type of a value is different than the expected type
+// and the value needs to be converted.
+type ConversionFunc func(v document.Value, path document.Path, targetType document.ValueType) (document.Value, error)
+
+// CastConversion is a ConversionFunc that casts the value to the target type.
+func CastConversion(v document.Value, path document.Path, targetType document.ValueType) (document.Value, error) {
+	newV, err := v.CastAs(targetType)
+	if err != nil {
+		return v, stringutil.Errorf("field %q must be of type %q, got %q", path, targetType, v.Type)
+	}
+
+	return newV, nil
+}
+
+// LosslessConversion is a ConversionFunc that only converts numbers if there is no precision loss in the process.
+func LosslessNumbersConversion(v document.Value, path document.Path, targetType document.ValueType) (document.Value, error) {
+	if v.Type == document.IntegerValue && targetType == document.DoubleValue {
+		return v.CastAsDouble()
+	}
+
+	if v.Type == document.DoubleValue && targetType == document.IntegerValue {
+		f := v.V.(float64)
+		if float64(int64(f)) == f {
+			return v.CastAsInteger()
+		}
+	}
+
+	return v, nil
+}
+
+// ConvertValueAtPath converts the value using the field constraints that are applicable
+// at the given path.
+func (f FieldConstraints) ConvertValueAtPath(path document.Path, v document.Value, conversionFn ConversionFunc) (document.Value, error) {
+	switch v.Type {
+	case document.ArrayValue:
+		vb, err := f.convertArrayAtPath(path, v.V.(document.Array), conversionFn)
+		return document.NewArrayValue(vb), err
+	case document.DocumentValue:
+		fb, err := f.convertDocumentAtPath(path, v.V.(document.Document), conversionFn)
+		return document.NewDocumentValue(fb), err
+	}
+	return f.convertScalarAtPath(path, v, conversionFn)
+}
+
+// convert the value using field constraints type information.
+// if there is a type constraint on a path, apply it.
+// if a value is an integer and has no constraint, convert it to double.
+func (f FieldConstraints) convertScalarAtPath(path document.Path, v document.Value, conversionFn ConversionFunc) (document.Value, error) {
+	for _, fc := range f {
+		if !fc.Path.IsEqual(path) {
+			continue
+		}
+
+		// check if the constraint enforce a particular type
+		// and if so convert the value to the new type.
+		if fc.Type != 0 {
+			newV, err := conversionFn(v, fc.Path, fc.Type)
+			if err != nil {
+				return v, err
+			}
+
+			return newV, nil
+		}
+		break
+	}
+
+	// no constraint have been found for this path.
+	// check if this is an integer and convert it to double.
+	if v.Type == document.IntegerValue {
+		newV, _ := v.CastAsDouble()
+		return newV, nil
+	}
+
+	return v, nil
+}
+
+func (f FieldConstraints) convertDocumentAtPath(path document.Path, d document.Document, conversionFn ConversionFunc) (*document.FieldBuffer, error) {
+	fb := document.NewFieldBuffer()
+	err := fb.Copy(d)
+	if err != nil {
+		return nil, err
+	}
+
+	err = fb.Apply(func(p document.Path, v document.Value) (document.Value, error) {
+		return f.convertScalarAtPath(append(path, p...), v, conversionFn)
+	})
+
+	return fb, err
+}
+
+func (f FieldConstraints) convertArrayAtPath(path document.Path, a document.Array, conversionFn ConversionFunc) (*document.ValueBuffer, error) {
+	vb := document.NewValueBuffer()
+	err := vb.Copy(a)
+	if err != nil {
+		return nil, err
+	}
+
+	err = vb.Apply(func(p document.Path, v document.Value) (document.Value, error) {
+		return f.convertScalarAtPath(append(path, p...), v, conversionFn)
+	})
+
+	return vb, err
+}
diff --git a/database/index.go b/database/index.go
index 40070fcf..1f0292aa 100644
--- a/database/index.go
+++ b/database/index.go
@@ -207,17 +207,6 @@ func (idx *Index) EncodeValue(v document.Value) ([]byte, error) {
 	}
 
 	var err error
-	if v.Type == document.IntegerValue {
-		if v.V == nil {
-			v.Type = document.DoubleValue
-		} else {
-			v, err = v.CastAsDouble()
-			if err != nil {
-				return nil, err
-			}
-		}
-	}
-
 	var buf bytes.Buffer
 	err = document.NewValueEncoder(&buf).Encode(v)
 	if err != nil {
@@ -248,17 +237,6 @@ func (idx *Index) iterate(st engine.Store, pivot document.Value, reverse bool, f
 	var seek []byte
 	var err error
 
-	if idx.Info.Type == 0 && pivot.Type == document.IntegerValue {
-		if pivot.V == nil {
-			pivot.Type = document.DoubleValue
-		} else {
-			pivot, err = pivot.CastAsDouble()
-			if err != nil {
-				return err
-			}
-		}
-	}
-
 	if pivot.V != nil {
 		seek, err = idx.EncodeValue(pivot)
 		if err != nil {
diff --git a/document/compare.go b/document/compare.go
index 3447059d..f28eebd6 100644
--- a/document/compare.go
+++ b/document/compare.go
@@ -224,7 +224,7 @@ func compareArrays(op operator, l Array, r Array) (bool, error) {
 		if lerr != nil || rerr != nil {
 			break
 		}
-		if lv.Type == rv.Type {
+		if lv.Type == rv.Type || (lv.Type.IsNumber() && rv.Type.IsNumber()) {
 			isEq, err := compare(operatorEq, lv, rv)
 			if err != nil {
 				return false, err
@@ -341,7 +341,7 @@ func compareDocuments(op operator, l, r Document) (bool, error) {
 		if lerr != nil || rerr != nil {
 			break
 		}
-		if lv.Type == rv.Type {
+		if lv.Type == rv.Type || (lv.Type.IsNumber() && rv.Type.IsNumber()) {
 			isEq, err := compare(operatorEq, lv, rv)
 			if err != nil {
 				return false, err
diff --git a/planner/optimizer.go b/planner/optimizer.go
index d41ef56d..0fcc2aab 100644
--- a/planner/optimizer.go
+++ b/planner/optimizer.go
@@ -506,7 +506,7 @@ func getCandidateFromfilterNode(f *stream.FilterOperator, tableName string, info
 	}
 
 	// determine if the operator can benefit from an index
-	ok, path, e := opCanUseIndex(op)
+	ok, path, e := operatorCanUseIndex(op)
 	if !ok {
 		return nil, nil
 	}
@@ -525,9 +525,10 @@ func getCandidateFromfilterNode(f *stream.FilterOperator, tableName string, info
 
 	// we'll start with checking if the path is the primary key of the table
 	if pk := info.GetPrimaryKey(); pk != nil && pk.Path.IsEqual(path) {
-		// if both types are different, don't select this scanner
-		if pk.Type != v.Type {
-			return nil, nil
+		// check if the operand can be used and convert it when possible
+		v, ok, err := operandCanUseIndex(pk.Type, pk.Path, info.FieldConstraints, v)
+		if err != nil || !ok {
+			return nil, err
 		}
 
 		cd.isPk = true
@@ -545,9 +546,10 @@ func getCandidateFromfilterNode(f *stream.FilterOperator, tableName string, info
 
 	// if not, check if an index exists for that path
 	if idx := indexes.GetIndexByPath(document.Path(path)); idx != nil {
-		// if both types are different, don't select this scanner
-		if !idx.Info.Type.IsZero() && idx.Info.Type != v.Type {
-			return nil, nil
+		// check if the operand can be used and convert it when possible
+		v, ok, err := operandCanUseIndex(idx.Info.Type, idx.Info.Path, info.FieldConstraints, v)
+		if err != nil || !ok {
+			return nil, err
 		}
 
 		cd.isIndex = true
@@ -571,7 +573,7 @@ func getCandidateFromfilterNode(f *stream.FilterOperator, tableName string, info
 	return nil, nil
 }
 
-func opCanUseIndex(op expr.Operator) (bool, document.Path, expr.Expr) {
+func operatorCanUseIndex(op expr.Operator) (bool, document.Path, expr.Expr) {
 	lf, leftIsField := op.LeftHand().(expr.Path)
 	rf, rightIsField := op.RightHand().(expr.Path)
 
@@ -608,6 +610,24 @@ func opCanUseIndex(op expr.Operator) (bool, document.Path, expr.Expr) {
 	return false, nil, nil
 }
 
+func operandCanUseIndex(indexType document.ValueType, path document.Path, fc database.FieldConstraints, v document.Value) (document.Value, bool, error) {
+	// ensure the operand satisfies all the constraints, index can work only on exact types.
+	// if a number is encountered, try to convert it to the right type if and only if the conversion
+	// is lossless.
+	converted, err := fc.ConvertValueAtPath(path, v, database.LosslessNumbersConversion)
+	if err != nil {
+		return v, false, err
+	}
+
+	// if the index is not typed, any operand can work
+	if indexType.IsZero() {
+		return converted, true, nil
+	}
+
+	// if the index is typed, it must be of the same type as the converted value
+	return converted, indexType == converted.Type, nil
+}
+
 func getRangesFromOp(op expr.Operator, v document.Value) (stream.Ranges, error) {
 	var ranges stream.Ranges
 


Test patch:
diff --git a/database/catalog_test.go b/database/catalog_test.go
index df00db82..6bf45379 100644
--- a/database/catalog_test.go
+++ b/database/catalog_test.go
@@ -5,8 +5,10 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/genjidb/genji"
 	"github.com/genjidb/genji/database"
 	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/testutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -662,3 +664,19 @@ func TestReIndexAll(t *testing.T) {
 		require.Equal(t, clone, catalog)
 	})
 }
+
+func TestReadOnlyTables(t *testing.T) {
+	db, err := genji.Open(":memory:")
+	require.NoError(t, err)
+	defer db.Close()
+
+	doc, err := db.QueryDocument(`CREATE TABLE foo; SELECT * FROM __genji_tables`)
+	require.NoError(t, err)
+
+	testutil.RequireDocJSONEq(t, doc, `{"field_constraints": [], "read_only":false, "store_name":"dAE=", "table_name":"foo"}`)
+
+	doc, err = db.QueryDocument(`CREATE INDEX idx_foo_a ON foo(a); SELECT * FROM __genji_indexes`)
+	require.NoError(t, err)
+
+	testutil.RequireDocJSONEq(t, doc, `{"index_name":"idx_foo_a", "path":["a"], "table_name":"foo", "unique":false}`)
+}
diff --git a/database/config_test.go b/database/config_test.go
index 1b9486dd..fc86cb17 100644
--- a/database/config_test.go
+++ b/database/config_test.go
@@ -218,213 +218,3 @@ func TestIndexStore(t *testing.T) {
 		require.Len(t, list, len(idxcfgs)-1)
 	})
 }
-
-func TestFieldConstraintsInfer(t *testing.T) {
-	tests := []struct {
-		name      string
-		got, want FieldConstraints
-		fails     bool
-	}{
-		{
-			"No change",
-			[]*FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
-			[]*FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
-			false,
-		},
-		{
-			"Array",
-			[]*FieldConstraint{{Path: document.NewPath("a", "0"), Type: document.IntegerValue}},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.ArrayValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "0")}},
-				{Path: document.NewPath("a", "0"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Document",
-			[]*FieldConstraint{{Path: document.NewPath("a", "b"), Type: document.IntegerValue}},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b")}},
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Complex path",
-			[]*FieldConstraint{{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.IntegerValue}},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
-				{Path: document.NewPath("a", "b"), Type: document.ArrayValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
-				{Path: document.NewPath("a", "b", "3"), Type: document.ArrayValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
-				{Path: document.NewPath("a", "b", "3", "1"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
-				{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Overlaping constraints",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-				{Path: document.NewPath("a", "c"), Type: document.IntegerValue},
-			},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b"), document.NewPath("a", "c")}},
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-				{Path: document.NewPath("a", "c"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Same path inferred and non inferred: inferred first",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-				{Path: document.NewPath("a"), Type: document.DocumentValue},
-			},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue},
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Same path inferred and non inferred: inferred last",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue},
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-			},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue},
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Complex case",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.DocumentValue},
-				{Path: document.NewPath("a", "b", "3", "1", "c", "d"), Type: document.IntegerValue},
-				{Path: document.NewPath("a", "b", "2"), Type: document.IntegerValue},
-			},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true,
-					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d"), document.NewPath("a", "b", "2")}},
-				{Path: document.NewPath("a", "b"), Type: document.ArrayValue, IsInferred: true,
-					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d"), document.NewPath("a", "b", "2")}},
-				{Path: document.NewPath("a", "b", "3"), Type: document.ArrayValue, IsInferred: true,
-					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d")}},
-				{Path: document.NewPath("a", "b", "3", "1"), Type: document.DocumentValue, IsInferred: true,
-					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d")}},
-				{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.DocumentValue},
-				{Path: document.NewPath("a", "b", "3", "1", "c", "d"), Type: document.IntegerValue},
-				{Path: document.NewPath("a", "b", "2"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Same path, different constraint",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-				{Path: document.NewPath("a", "b"), Type: document.DoubleValue},
-			},
-			nil,
-			true,
-		},
-		{
-			"Inferred constraint first, conflict with non inferred",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-				{Path: document.NewPath("a"), Type: document.IntegerValue},
-			},
-			nil,
-			true,
-		},
-		{
-			"Non inferred constraint first, conflict with inferred",
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.IntegerValue},
-				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
-			},
-			nil,
-			true,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			got, err := test.got.Infer()
-			if test.fails {
-				require.Error(t, err)
-			} else {
-				require.NoError(t, err)
-				require.Equal(t, test.want, got)
-			}
-		})
-	}
-}
-
-func TestFieldConstraintsAdd(t *testing.T) {
-	tests := []struct {
-		name  string
-		got   FieldConstraints
-		add   FieldConstraint
-		want  FieldConstraints
-		fails bool
-	}{
-		{
-			"Same path",
-			[]*FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
-			FieldConstraint{Path: document.NewPath("a"), Type: document.IntegerValue},
-			nil,
-			true,
-		},
-		{
-			"Duplicate primary key",
-			[]*FieldConstraint{{Path: document.NewPath("a"), IsPrimaryKey: true, Type: document.IntegerValue}},
-			FieldConstraint{Path: document.NewPath("b"), IsPrimaryKey: true, Type: document.IntegerValue},
-			nil,
-			true,
-		},
-		{
-			"Different path",
-			[]*FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
-			FieldConstraint{Path: document.NewPath("b"), Type: document.IntegerValue},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.IntegerValue},
-				{Path: document.NewPath("b"), Type: document.IntegerValue},
-			},
-			false,
-		},
-		{
-			"Default value conversion, typed constraint",
-			[]*FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
-			FieldConstraint{Path: document.NewPath("b"), Type: document.IntegerValue, DefaultValue: document.NewDoubleValue(5)},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.IntegerValue},
-				{Path: document.NewPath("b"), Type: document.IntegerValue, DefaultValue: document.NewIntegerValue(5)},
-			},
-			false,
-		},
-		{
-			"Default value conversion, untyped constraint",
-			[]*FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
-			FieldConstraint{Path: document.NewPath("b"), DefaultValue: document.NewIntegerValue(5)},
-			[]*FieldConstraint{
-				{Path: document.NewPath("a"), Type: document.IntegerValue},
-				{Path: document.NewPath("b"), DefaultValue: document.NewDoubleValue(5)},
-			},
-			false,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			err := test.got.Add(&test.add)
-			if test.fails {
-				require.Error(t, err)
-			} else {
-				require.NoError(t, err)
-				require.Equal(t, test.want, test.got)
-			}
-		})
-	}
-}
diff --git a/database/constraint_test.go b/database/constraint_test.go
new file mode 100644
index 00000000..22eb5361
--- /dev/null
+++ b/database/constraint_test.go
@@ -0,0 +1,304 @@
+package database_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/genjidb/genji/database"
+	"github.com/genjidb/genji/document"
+	"github.com/genjidb/genji/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFieldConstraintsInfer(t *testing.T) {
+	tests := []struct {
+		name      string
+		got, want database.FieldConstraints
+		fails     bool
+	}{
+		{
+			"No change",
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			false,
+		},
+		{
+			"Array",
+			[]*database.FieldConstraint{{Path: document.NewPath("a", "0"), Type: document.IntegerValue}},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.ArrayValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "0")}},
+				{Path: document.NewPath("a", "0"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Document",
+			[]*database.FieldConstraint{{Path: document.NewPath("a", "b"), Type: document.IntegerValue}},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b")}},
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Primary key",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.ArrayValue, IsPrimaryKey: true},
+				{Path: document.NewPath("a", "0"), Type: document.IntegerValue},
+			},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.ArrayValue, IsPrimaryKey: true},
+				{Path: document.NewPath("a", "0"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Complex path",
+			[]*database.FieldConstraint{{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.IntegerValue}},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
+				{Path: document.NewPath("a", "b"), Type: document.ArrayValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
+				{Path: document.NewPath("a", "b", "3"), Type: document.ArrayValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
+				{Path: document.NewPath("a", "b", "3", "1"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c")}},
+				{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Overlaping constraints",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+				{Path: document.NewPath("a", "c"), Type: document.IntegerValue},
+			},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true, InferredBy: []document.Path{document.NewPath("a", "b"), document.NewPath("a", "c")}},
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+				{Path: document.NewPath("a", "c"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Same path inferred and non inferred: inferred first",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+				{Path: document.NewPath("a"), Type: document.DocumentValue},
+			},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue},
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Same path inferred and non inferred: inferred last",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue},
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+			},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue},
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Complex case",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.DocumentValue},
+				{Path: document.NewPath("a", "b", "3", "1", "c", "d"), Type: document.IntegerValue},
+				{Path: document.NewPath("a", "b", "2"), Type: document.IntegerValue},
+			},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.DocumentValue, IsInferred: true,
+					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d"), document.NewPath("a", "b", "2")}},
+				{Path: document.NewPath("a", "b"), Type: document.ArrayValue, IsInferred: true,
+					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d"), document.NewPath("a", "b", "2")}},
+				{Path: document.NewPath("a", "b", "3"), Type: document.ArrayValue, IsInferred: true,
+					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d")}},
+				{Path: document.NewPath("a", "b", "3", "1"), Type: document.DocumentValue, IsInferred: true,
+					InferredBy: []document.Path{document.NewPath("a", "b", "3", "1", "c"), document.NewPath("a", "b", "3", "1", "c", "d")}},
+				{Path: document.NewPath("a", "b", "3", "1", "c"), Type: document.DocumentValue},
+				{Path: document.NewPath("a", "b", "3", "1", "c", "d"), Type: document.IntegerValue},
+				{Path: document.NewPath("a", "b", "2"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Same path, different constraint",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+				{Path: document.NewPath("a", "b"), Type: document.DoubleValue},
+			},
+			nil,
+			true,
+		},
+		{
+			"Inferred constraint first, conflict with non inferred",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+				{Path: document.NewPath("a"), Type: document.IntegerValue},
+			},
+			nil,
+			true,
+		},
+		{
+			"Non inferred constraint first, conflict with inferred",
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.IntegerValue},
+				{Path: document.NewPath("a", "b"), Type: document.IntegerValue},
+			},
+			nil,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, err := test.got.Infer()
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.want, got)
+			}
+		})
+	}
+}
+
+func TestFieldConstraintsAdd(t *testing.T) {
+	tests := []struct {
+		name  string
+		got   database.FieldConstraints
+		add   database.FieldConstraint
+		want  database.FieldConstraints
+		fails bool
+	}{
+		{
+			"Same path",
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			database.FieldConstraint{Path: document.NewPath("a"), Type: document.IntegerValue},
+			nil,
+			true,
+		},
+		{
+			"Duplicate primary key",
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), IsPrimaryKey: true, Type: document.IntegerValue}},
+			database.FieldConstraint{Path: document.NewPath("b"), IsPrimaryKey: true, Type: document.IntegerValue},
+			nil,
+			true,
+		},
+		{
+			"Different path",
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			database.FieldConstraint{Path: document.NewPath("b"), Type: document.IntegerValue},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.IntegerValue},
+				{Path: document.NewPath("b"), Type: document.IntegerValue},
+			},
+			false,
+		},
+		{
+			"Default value conversion, typed constraint",
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			database.FieldConstraint{Path: document.NewPath("b"), Type: document.IntegerValue, DefaultValue: document.NewDoubleValue(5)},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.IntegerValue},
+				{Path: document.NewPath("b"), Type: document.IntegerValue, DefaultValue: document.NewIntegerValue(5)},
+			},
+			false,
+		},
+		{
+			"Default value conversion, untyped constraint",
+			[]*database.FieldConstraint{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			database.FieldConstraint{Path: document.NewPath("b"), DefaultValue: document.NewIntegerValue(5)},
+			[]*database.FieldConstraint{
+				{Path: document.NewPath("a"), Type: document.IntegerValue},
+				{Path: document.NewPath("b"), DefaultValue: document.NewDoubleValue(5)},
+			},
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := test.got.Add(&test.add)
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.want, test.got)
+			}
+		})
+	}
+}
+
+func TestFieldConstraintsConvert(t *testing.T) {
+	tests := []struct {
+		constraints database.FieldConstraints
+		path        document.Path
+		in, want    document.Value
+		fails       bool
+	}{
+		{
+			nil,
+			document.NewPath("a"),
+			document.NewIntegerValue(10),
+			document.NewDoubleValue(10),
+			false,
+		},
+		{
+			database.FieldConstraints{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			document.NewPath("a"),
+			document.NewIntegerValue(10),
+			document.NewIntegerValue(10),
+			false,
+		},
+		{
+			database.FieldConstraints{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			document.NewPath("a"),
+			document.NewDoubleValue(10.5),
+			document.NewIntegerValue(10),
+			false,
+		},
+		{
+			database.FieldConstraints{{Path: document.NewPath("a", "0"), Type: document.IntegerValue}},
+			document.NewPath("a"),
+			document.NewArrayValue(testutil.MakeArray(t, `[10.5, 10.5]`)),
+			document.NewArrayValue(testutil.MakeArray(t, `[10, 10.5]`)),
+			false,
+		},
+		{
+			database.FieldConstraints{{Path: document.NewPath("a", "b"), Type: document.IntegerValue}},
+			document.NewPath("a"),
+			document.NewDocumentValue(testutil.MakeDocument(t, `{"b": 10.5, "c": 10.5}`)),
+			document.NewDocumentValue(testutil.MakeDocument(t, `{"b": 10, "c": 10.5}`)),
+			false,
+		},
+		{
+			database.FieldConstraints{{Path: document.NewPath("a"), Type: document.IntegerValue}},
+			document.NewPath("a"),
+			document.NewTextValue("foo"),
+			document.NewTextValue("foo"),
+			true,
+		},
+		{
+			database.FieldConstraints{{Path: document.NewPath("a"), DefaultValue: document.NewIntegerValue(10)}},
+			document.NewPath("a"),
+			document.NewTextValue("foo"),
+			document.NewTextValue("foo"),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%s / %v to %v", test.path, test.in, test.want), func(t *testing.T) {
+			got, err := test.constraints.ConvertValueAtPath(test.path, test.in, database.CastConversion)
+			if test.fails {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.want, got)
+			}
+		})
+	}
+}
diff --git a/database/index_test.go b/database/index_test.go
index 8a2b89dd..d3eed616 100644
--- a/database/index_test.go
+++ b/database/index_test.go
@@ -82,10 +82,10 @@ func TestIndexDelete(t *testing.T) {
 		i := 0
 		err := idx.AscendGreaterOrEqual(pivot, func(v, k []byte) error {
 			if i == 0 {
-				requireEqualEncoded(t, document.NewDoubleValue(10), v)
+				requireEqualEncoded(t, document.NewIntegerValue(10), v)
 				require.Equal(t, "other-key", string(k))
 			} else if i == 1 {
-				requireEqualEncoded(t, document.NewDoubleValue(11), v)
+				requireEqualEncoded(t, document.NewIntegerValue(11), v)
 				require.Equal(t, "yet-another-key", string(k))
 			} else {
 				return errors.New("should not reach this point")
@@ -111,10 +111,10 @@ func TestIndexDelete(t *testing.T) {
 		err := idx.AscendGreaterOrEqual(document.Value{Type: document.IntegerValue}, func(v, k []byte) error {
 			switch i {
 			case 0:
-				requireEqualEncoded(t, document.NewDoubleValue(10), v)
+				requireEqualEncoded(t, document.NewIntegerValue(10), v)
 				require.Equal(t, "key1", string(k))
 			case 1:
-				requireEqualEncoded(t, document.NewDoubleValue(12), v)
+				requireEqualEncoded(t, document.NewIntegerValue(12), v)
 				require.Equal(t, "key3", string(k))
 			default:
 				return errors.New("should not reach this point")
@@ -191,7 +191,7 @@ func TestIndexAscendGreaterThan(t *testing.T) {
 					err := idx.AscendGreaterOrEqual(test.pivot, func(val, rid []byte) error {
 						switch test.t {
 						case document.IntegerValue:
-							requireEqualEncoded(t, document.NewDoubleValue(float64(i)), val)
+							requireEqualEncoded(t, document.NewIntegerValue(int64(i)), val)
 						case document.DoubleValue:
 							requireEqualEncoded(t, document.NewDoubleValue(float64(i)+float64(i)/2), val)
 						case document.TextValue:
@@ -299,7 +299,7 @@ func TestIndexAscendGreaterThan(t *testing.T) {
 		var doubles, texts int
 		i := int64(0)
 		err := idx.AscendGreaterOrEqual(document.Value{Type: document.IntegerValue}, func(val, rid []byte) error {
-			requireEqualEncoded(t, document.NewDoubleValue(1), val)
+			requireEqualEncoded(t, document.NewIntegerValue(1), val)
 			require.Equal(t, binarysort.AppendInt64(nil, i), rid)
 			i++
 			doubles++
@@ -349,7 +349,7 @@ func TestIndexDescendLessOrEqual(t *testing.T) {
 			var i uint8 = 8
 			var count int
 			err := idx.DescendLessOrEqual(document.Value{Type: document.IntegerValue}, func(val, key []byte) error {
-				requireEqualEncoded(t, document.NewDoubleValue(float64(i)), val)
+				requireEqualEncoded(t, document.NewIntegerValue(int64(i)), val)
 				require.Equal(t, []byte{'a' + i}, key)
 
 				i -= 2
@@ -400,7 +400,7 @@ func TestIndexDescendLessOrEqual(t *testing.T) {
 					require.Equal(t, []byte{'s', 'a' + byte(texts)}, rid)
 					texts--
 				} else {
-					requireEqualEncoded(t, document.NewDoubleValue(float64(ints)), val)
+					requireEqualEncoded(t, document.NewIntegerValue(int64(ints)), val)
 					require.Equal(t, []byte{'i', 'a' + byte(ints)}, rid)
 					ints--
 				}
diff --git a/document/compare_test.go b/document/compare_test.go
index 491bd5d1..91a04dfd 100644
--- a/document/compare_test.go
+++ b/document/compare_test.go
@@ -158,7 +158,7 @@ func TestCompare(t *testing.T) {
 		{"=", `[]`, `[]`, true, jsonToArray},
 		{"=", `[1]`, `[1]`, true, jsonToArray},
 		{"=", `[1]`, `[]`, false, jsonToArray},
-		{"=", `[1.0, 2]`, `[1, 2]`, false, jsonToArray},
+		{"=", `[1.0, 2]`, `[1, 2]`, true, jsonToArray},
 		{"=", `[1,2,3]`, `[1,2,3]`, true, jsonToArray},
 		{"!=", `[1]`, `[5]`, true, jsonToArray},
 		{"!=", `[1]`, `[1, 1]`, true, jsonToArray},
@@ -210,6 +210,7 @@ func TestCompare(t *testing.T) {
 		// document
 		{"=", `{}`, `{}`, true, jsonToDocument},
 		{"=", `{"a": 1}`, `{"a": 1}`, true, jsonToDocument},
+		{"=", `{"a": 1.0}`, `{"a": 1}`, true, jsonToDocument},
 		{"=", `{"a": 1, "b": 2}`, `{"b": 2, "a": 1}`, true, jsonToDocument},
 		{"=", `{"a": 1, "b": {"a": 1}}`, `{"b": {"a": 1}, "a": 1}`, true, jsonToDocument},
 		{">", `{"a": 2}`, `{"a": 1}`, true, jsonToDocument},
@@ -226,7 +227,7 @@ func TestCompare(t *testing.T) {
 
 	for _, test := range tests {
 		a, b := test.converter(t, test.a), test.converter(t, test.b)
-		t.Run(fmt.Sprintf("%s/%v%v%v", a.Type.String(), a, test.op, b), func(t *testing.T) {
+		t.Run(fmt.Sprintf("%s/%v%v%v", a.Type.String(), test.a, test.op, test.b), func(t *testing.T) {
 			var ok bool
 			var err error
 
diff --git a/planner/optimizer_test.go b/planner/optimizer_test.go
index eb854323..2dacce5a 100644
--- a/planner/optimizer_test.go
+++ b/planner/optimizer_test.go
@@ -9,6 +9,7 @@ import (
 	"github.com/genjidb/genji/planner"
 	"github.com/genjidb/genji/sql/parser"
 	st "github.com/genjidb/genji/stream"
+	"github.com/genjidb/genji/testutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -426,6 +427,11 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			st.New(st.IndexScan("idx_foo_a", st.Range{Min: document.NewIntegerValue(1), Exact: true})).
 				Pipe(st.Filter(parser.MustParseExpr("k = 'hello'"))),
 		},
+		{ // c is an INT, 1.1 cannot be converted to int without precision loss, don't use the index
+			"FROM foo WHERE c < 1.1",
+			st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("c < 1.1"))),
+			st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("c < 1.1"))),
+		},
 	}
 
 	for _, test := range tests {
@@ -455,4 +461,72 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			require.Equal(t, test.expected.String(), res.String())
 		})
 	}
+
+	t.Run("array indexes", func(t *testing.T) {
+		tests := []struct {
+			name           string
+			root, expected *st.Stream
+		}{
+			{
+				"non-indexed path",
+				st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("b = [1, 1]"))),
+				st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("b = [1, 1]"))),
+			},
+			{
+				"FROM foo WHERE k = [1, 1]",
+				st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("k = [1, 1]"))),
+				st.New(st.PkScan("foo", st.Range{Min: document.NewArrayValue(testutil.MakeArray(t, `[1, 1]`)), Exact: true})),
+			},
+			{ // constraint on k[0] INT should not modify the operand
+				"FROM foo WHERE k = [1.5, 1.5]",
+				st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("k = [1.5, 1.5]"))),
+				st.New(st.PkScan("foo", st.Range{Min: document.NewArrayValue(testutil.MakeArray(t, `[1.5, 1.5]`)), Exact: true})),
+			},
+			{
+				"FROM foo WHERE a = [1, 1]",
+				st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("a = [1, 1]"))),
+				st.New(st.IndexScan("idx_foo_a", st.Range{Min: document.NewArrayValue(testutil.MakeArray(t, `[1, 1]`)), Exact: true})),
+			},
+			{ // constraint on a[0] DOUBLE should modify the operand because it's lossless
+				"FROM foo WHERE a = [1, 1.5]",
+				st.New(st.SeqScan("foo")).Pipe(st.Filter(parser.MustParseExpr("a = [1, 1.5]"))),
+				st.New(st.IndexScan("idx_foo_a", st.Range{Min: document.NewArrayValue(testutil.MakeArray(t, `[1.0, 1.5]`)), Exact: true})),
+			},
+		}
+
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				db, err := genji.Open(":memory:")
+				require.NoError(t, err)
+				defer db.Close()
+
+				tx, err := db.Begin(true)
+				require.NoError(t, err)
+				defer tx.Rollback()
+
+				err = tx.Exec(`
+					CREATE TABLE foo (
+						k ARRAY PRIMARY KEY,
+						k[0] INT,
+						a ARRAY,
+						a[0] DOUBLE
+					);
+					CREATE INDEX idx_foo_a ON foo(a);
+					CREATE INDEX idx_foo_a0 ON foo(a[0]);
+					INSERT INTO foo (k, a, b) VALUES
+						([1, 1], [1, 1], [1, 1]),
+						([2, 2], [2, 2], [2, 2]),
+						([3, 3], [3, 3], [3, 3])
+				`)
+				require.NoError(t, err)
+
+				res, err := planner.PrecalculateExprRule(test.root, tx.Transaction, nil)
+				require.NoError(t, err)
+
+				res, err = planner.UseIndexBasedOnFilterNodeRule(res, tx.Transaction, nil)
+				require.NoError(t, err)
+				require.Equal(t, test.expected.String(), res.String())
+			})
+		}
+	})
 }
diff --git a/query/select_test.go b/query/select_test.go
index 6fda89bb..6408b302 100644
--- a/query/select_test.go
+++ b/query/select_test.go
@@ -265,6 +265,37 @@ func TestSelectStmt(t *testing.T) {
 
 		require.JSONEq(t, `{"MAX(a)": null, "MIN(b)": null, "COUNT(*)": 0, "SUM(id)": null}`, string(enc))
 	})
+
+	t.Run("array number comparison with no constraints", func(t *testing.T) {
+		db, err := genji.Open(":memory:")
+		require.NoError(t, err)
+		defer db.Close()
+
+		err = db.Exec(`
+			CREATE TABLE test;
+			INSERT INTO test (a) VALUES ([1,2,3]), ([4, 5, 6]);
+		`)
+		require.NoError(t, err)
+
+		check := func() {
+			t.Helper()
+
+			d, err := db.QueryDocument("SELECT * FROM test WHERE a = [1,2,3];")
+			require.NoError(t, err)
+
+			enc, err := json.Marshal(d)
+			require.NoError(t, err)
+
+			require.JSONEq(t, `{"a": [1, 2, 3]}`, string(enc))
+		}
+
+		check()
+
+		err = db.Exec("CREATE INDEX idx_test_a ON test(a);")
+		require.NoError(t, err)
+
+		check()
+	})
 }
 
 func TestDistinct(t *testing.T) {
diff --git a/testutil/document.go b/testutil/document.go
index fe181273..0c6bbb6c 100644
--- a/testutil/document.go
+++ b/testutil/document.go
@@ -39,6 +39,18 @@ func MakeDocuments(t testing.TB, jsonDocs ...string) (docs Docs) {
 	return
 }
 
+// MakeArray creates an array from a json string.
+func MakeArray(t testing.TB, jsonArray string) document.Array {
+	t.Helper()
+
+	var vb document.ValueBuffer
+
+	err := vb.UnmarshalJSON([]byte(jsonArray))
+	require.NoError(t, err)
+
+	return &vb
+}
+
 type Docs []document.Document
 
 func (docs Docs) RequireEqual(t testing.TB, others Docs) {
@@ -66,3 +78,11 @@ func Dump(t testing.TB, v interface{}) {
 	err := enc.Encode(v)
 	require.NoError(t, err)
 }
+
+func RequireDocJSONEq(t testing.TB, d document.Document, expected string) {
+	t.Helper()
+
+	data, err := json.Marshal(d)
+	require.NoError(t, err)
+	require.JSONEq(t, expected, string(data))
+}