enforce int min/max constraints on insert (fb-1772) (#2325)

* moved the debug code to the right spot

* enforce int min/max constraints on inserts

* add a check for decimal min and max

* fixed borked tests

* fix the decimal to int conversion in constraint check

Co-authored-by: Travis Turner <travis@molecula.com>

etcd/embed.go

@@ -228,10 +228,10 @@ func NewEtcd(opt Options, logger logger.Logger, replicas int, version string) *E
 
 	if e.options.HeartbeatTTL == 0 {
 		e.options.HeartbeatTTL = 5 // seconds
-		// !! DEBUG
-		// e.options.HeartbeatTTL = 3600 // seconds
-		// !! DEBUG
 	}
+	// !! DEBUG
+	// e.options.HeartbeatTTL = 3600 // seconds
+	// !! DEBUG
 	return e
 }
 

sql3/errors.go

@@ -89,6 +89,10 @@ const (
 	ErrParameterTypeMistmatch               errors.Code = "ErrParameterTypeMistmatch"
 	ErrCallParameterValueInvalid            errors.Code = "ErrCallParameterValueInvalid"
 
+	// insert errors
+
+	ErrInsertValueOutOfRange errors.Code = "ErrInsertValueOutOfRange"
+
 	// bulk insert errors
 
 	ErrReadingDatasource       errors.Code = "ErrReadingDatasource"
@@ -576,6 +580,15 @@ func NewErrCallParameterValueInvalid(line, col int, badParameterValue string, pa
 	)
 }
 
+// insert
+
+func NewErrInsertValueOutOfRange(line, col int, columnName string, rowNumber int, badValue interface{}) error {
+	return errors.New(
+		ErrInsertValueOutOfRange,
+		fmt.Sprintf("[%d:%d] inserting value into column '%s', row %d, value '%v' out of range", line, col, columnName, rowNumber, badValue),
+	)
+}
+
 // bulk insert
 
 func NewErrReadingDatasource(line, col int, dataSource string, errorText string) error {

sql3/planner/opinsert.go

@@ -10,6 +10,7 @@ import (
 
 	pilosa "github.com/molecula/featurebase/v3"
 	fbbatch "github.com/molecula/featurebase/v3/batch"
+	"github.com/molecula/featurebase/v3/pql"
 	"github.com/molecula/featurebase/v3/sql3"
 	"github.com/molecula/featurebase/v3/sql3/planner/types"
 	"github.com/pkg/errors"
@@ -184,7 +185,7 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
 	// record ID ("_id") since that's stored in row.ID.
 	row.Values = make([]interface{}, len(i.targetColumns)-1)
 
-	for _, tuple := range i.insertValues {
+	for rowNumber, tuple := range i.insertValues {
 		// Evaluate and set the record ID.
 		if eval, err := tuple[posID].Evaluate(nil); err != nil {
 			return nil, errors.Wrapf(err, "evaluating record id: %v", tuple[posID])
@@ -219,6 +220,8 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
 				return nil, errors.Wrapf(err, "evaluating tuple value: %v", iv)
 			}
 
+			columnName := idxInfo.Fields[posVals[idx]].Name
+
 			// batch.Add does not typically look at field type to determine how
 			// to handle a particular value in a row. Instead, it uses value
 			// type. As an example, if the value type is int64, then batch.Add
@@ -255,6 +258,32 @@ func (i *insertRowIter) Next(ctx context.Context) (types.Row, error) {
 					row.Values[posVals[idx]] = eval
 				}
 
+			case pilosa.FieldTypeInt:
+				if eval != nil {
+					v, ok := eval.(int64)
+					if !ok {
+						return nil, sql3.NewErrInternalf("unexpected type %v", eval)
+					}
+					// check the min and max constraints here
+					if v < opts.Min.ToInt64(0) || v > opts.Max.ToInt64(0) {
+						return nil, sql3.NewErrInsertValueOutOfRange(0, 0, columnName, rowNumber+1, v)
+					}
+				}
+				row.Values[posVals[idx]] = eval
+
+			case pilosa.FieldTypeDecimal:
+				if eval != nil {
+					v, ok := eval.(pql.Decimal)
+					if !ok {
+						return nil, sql3.NewErrInternalf("unexpected type %v", eval)
+					}
+					// check the min and max constraints here
+					if v.LessThan(opts.Min) || v.GreaterThan(opts.Max) {
+						return nil, sql3.NewErrInsertValueOutOfRange(0, 0, columnName, rowNumber+1, v)
+					}
+				}
+				row.Values[posVals[idx]] = eval
+
 			case pilosa.FieldTypeTimestamp:
 				switch v := eval.(type) {
 

sql3/test/defs/defs_inserts.go

@@ -9,7 +9,7 @@ var insertTest = TableTest{
 			srcHdr("b", fldTypeInt, "min 0", "max 1000"),
 			srcHdr("s", fldTypeString),
 			srcHdr("bl", fldTypeBool),
-			srcHdr("d", fldTypeDecimal2),
+			srcHdr("d", fldTypeDecimal2, "min 0", "max 1000"),
 			srcHdr("event", fldTypeStringSet),
 			srcHdr("ievent", fldTypeIDSet),
 		),
@@ -124,5 +124,33 @@ var insertTest = TableTest{
 			),
 			ExpErr: "an expression of type 'stringset' cannot be assigned to type 'idset'",
 		},
+		{
+			name: "min constraint",
+			SQLs: sqls(
+				"insert into testinsert (_id, a) values (400, -1)",
+			),
+			ExpErr: "inserting value into column 'a', row 1, value '-1' out of range",
+		},
+		{
+			name: "max constraint",
+			SQLs: sqls(
+				"insert into testinsert (_id, a) values (400, 1001)",
+			),
+			ExpErr: "inserting value into column 'a', row 1, value '1001' out of range",
+		},
+		{
+			name: "min constraint decimal",
+			SQLs: sqls(
+				"insert into testinsert (_id, d) values (400, -1.00)",
+			),
+			ExpErr: "inserting value into column 'd', row 1, value '-1' out of range",
+		},
+		{
+			name: "max constraint decimal",
+			SQLs: sqls(
+				"insert into testinsert (_id, d) values (400, 1001.00)",
+			),
+			ExpErr: "inserting value into column 'd', row 1, value '1001' out of range",
+		},
 	},
 }

sql3/test/defs/types.go

@@ -274,6 +274,7 @@ func (s source) insertInto(t *testing.T, rowSets []int) string {
 	for _, rowSet := range rowSets {
 		ii += sourceRows(s.rows[rowSet]).insertTuples(t)
 	}
+	log.Printf("INSERT: %s", ii)
 	return ii
 }