allow before insert trigger to specify missing column

Author: James Cor

enginetest/queries/trigger_queries.go

@@ -506,6 +506,32 @@ var TriggerTests = []ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "insert trigger with missing column default value",
+		SetUpScript: []string{
+			"CREATE TABLE t (i INT PRIMARY KEY, j INT NOT NULL);",
+			`
+CREATE TRIGGER trig BEFORE INSERT ON t 
+FOR EACH ROW
+BEGIN
+    SET new.j = 10;
+END;`,
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query: "INSERT INTO t (i) VALUES (1);",
+				Expected: []sql.Row{
+					{types.OkResult{RowsAffected: 1}},
+				},
+			},
+			{
+				Query: "SELECT * FROM t;",
+				Expected: []sql.Row{
+					{1, 10},
+				},
+			},
+		},
+	},
 
 	// UPDATE triggers
 	{

sql/analyzer/inserts.go

@@ -90,6 +90,65 @@ func resolveInsertRows(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.Sc
 	})
 }
 
+func validateInsertRows(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.Scope, sel RuleSelector, qFlags *sql.QueryFlags) (sql.Node, transform.TreeIdentity, error) {
+	if _, ok := n.(*plan.TriggerExecutor); ok {
+		return n, transform.SameTree, nil
+	} else if _, ok := n.(*plan.CreateProcedure); ok {
+		return n, transform.SameTree, nil
+	}
+	// We capture all INSERTs along the tree, such as those inside of block statements.
+	var err error
+	transform.Inspect(n, func(n sql.Node) bool {
+		insert, ok := n.(*plan.InsertInto)
+		if !ok {
+			return true
+		}
+
+		var insertable sql.InsertableTable
+		table := getResolvedTable(insert.Destination)
+		insertable, err = plan.GetInsertable(table)
+		if err != nil {
+			return false
+		}
+
+		source := insert.Source
+		dstSchema := insertable.Schema()
+
+		// normalize the column name
+		columnNames := make([]string, len(insert.ColumnNames))
+		for i, name := range insert.ColumnNames {
+			columnNames[i] = strings.ToLower(name)
+		}
+
+		// If no columns are given and value tuples are not all empty, use the full schema
+		if len(columnNames) == 0 && existsNonZeroValueCount(source) {
+			columnNames = make([]string, len(dstSchema))
+			for i, f := range dstSchema {
+				columnNames[i] = f.Name
+			}
+		}
+
+		for _, col := range dstSchema {
+			colIdx := findColIdx(col.Name, columnNames)
+			if colIdx != -1 {
+				continue
+			}
+
+			if _, isTrigExec := source.(*plan.TriggerExecutor); !isTrigExec && !col.AutoIncrement && !col.Nullable && col.Default == nil && col.Generated == nil {
+				err = sql.ErrInsertIntoNonNullableDefaultNullColumn.New(col.Name)
+				return false
+			}
+		}
+
+		return true
+	})
+
+	if err != nil {
+		return nil, transform.SameTree, err
+	}
+	return n, transform.SameTree, nil
+}
+
 // Ensures that the number of elements in each Value tuple is empty
 func existsNonZeroValueCount(values sql.Node) bool {
 	switch node := values.(type) {
@@ -123,15 +182,11 @@ func wrapRowSource(ctx *sql.Context, insertSource sql.Node, destTbl sql.Table, s
 
 	for i, col := range schema {
 		colIdx := findColIdx(col.Name, columnNames)
-		// if column was not explicitly specified, try to substitute with default or generated value
 		if colIdx == -1 {
 			defaultExpr := col.Default
 			if defaultExpr == nil {
 				defaultExpr = col.Generated
 			}
-			if !col.Nullable && defaultExpr == nil && !col.AutoIncrement {
-				return nil, -1, sql.ErrInsertIntoNonNullableDefaultNullColumn.New(col.Name)
-			}
 
 			var err error
 			colNameToIdx := make(map[string]int)

sql/analyzer/rule_ids.go

@@ -61,6 +61,7 @@ const (
 	applyHashInId                // applyHashIn
 	resolveInsertRowsId          // resolveInsertRows
 	applyTriggersId              // applyTriggers
+	validateInsertRowsId         // validateInsertRows
 	applyProceduresId            // applyProcedures
 	assignRoutinesId             // assignRoutines
 	modifyUpdateExprsForJoinId   // modifyUpdateExprsForJoin

sql/analyzer/ruleid_string.go

@@ -20,6 +20,7 @@ func init() {
 		// resolveInsertRows inserts a projection wrapping values that cannot be seen by fixup
 		{resolveInsertRowsId, resolveInsertRows},
 		{applyTriggersId, applyTriggers},
+		{validateInsertRowsId, validateInsertRows},
 		{applyProceduresId, applyProcedures},
 		{inlineSubqueryAliasRefsId, inlineSubqueryAliasRefs},
 		{cacheSubqueryAliasesInJoinsId, cacheSubqueryAliasesInJoins},