[release-21.0] bugfix: INSERT IGNORE not inserting rows (vitessio#18151) (vitessio#18164)

Signed-off-by: Andres Taylor <[email protected]>
Co-authored-by: vitess-bot[bot] <108069721+vitess-bot[bot]@users.noreply.github.com>

Author: vitess-bot[bot]

go/test/endtoend/vtgate/queries/dml/dml_test.go

@@ -77,7 +77,6 @@ func TestUniqueLookupDuplicateEntries(t *testing.T) {
 
 // TestUniqueLookupDuplicateIgnore tests the insert ignore on lookup table.
 func TestUniqueLookupDuplicateIgnore(t *testing.T) {
-	t.Skip("this is fixed in https://github.com/vitessio/vitess/pull/18151, it will be backported to 22.0")
 	mcmp, closer := start(t)
 	defer closer()
 
@@ -94,21 +93,20 @@ func TestUniqueLookupDuplicateIgnore(t *testing.T) {
 
 	// insert duplicate row in multi-row insert - lookup single shard
 	// Current behavior does not work as expected—one of the rows should be inserted.
-	// The lookup table is updated, but the main table is not. This is a bug in Vitess.
 	// The issue occurs because the table has two vindex columns (`num` and `col`), both of which ignore nulls during vindex insertion.
 	// In the `INSERT IGNORE` case, after the vindex create API call, a verify call checks if the row exists in the lookup table.
 	// - If the row exists, it is inserted into the main table.
 	// - If the row does not exist, the main table insertion is skipped.
 	// Since the `col` column is null, the row is not inserted into the lookup table, causing the main table insertion to be ignored.
 	qr = utils.Exec(t, mcmp.VtConn, "insert ignore into s_tbl(id, num) values (3,20), (4,20)")
-	assert.EqualValues(t, 0, qr.RowsAffected)
-	utils.AssertMatches(t, mcmp.VtConn, "select id, num from s_tbl order by id", `[[INT64(1) INT64(10)]]`)
+	assert.EqualValues(t, 1, qr.RowsAffected)
+	utils.AssertMatches(t, mcmp.VtConn, "select id, num from s_tbl order by id", `[[INT64(1) INT64(10)] [INT64(3) INT64(20)]]`)
 	utils.AssertMatches(t, mcmp.VtConn, "select num, hex(keyspace_id) from num_vdx_tbl order by num", `[[INT64(10) VARCHAR("166B40B44ABA4BD6")] [INT64(20) VARCHAR("4EB190C9A2FA169C")]]`)
 
 	// insert duplicate row in multi-row insert - vindex values are not null
 	qr = utils.Exec(t, mcmp.VtConn, "insert ignore into s_tbl(id, num, col) values (3,20, 30), (4,20, 40)")
-	assert.EqualValues(t, 1, qr.RowsAffected)
-	utils.AssertMatches(t, mcmp.VtConn, "select id, num, col from s_tbl order by id", `[[INT64(1) INT64(10) NULL] [INT64(3) INT64(20) INT64(30)]]`)
+	assert.EqualValues(t, 0, qr.RowsAffected)
+	utils.AssertMatches(t, mcmp.VtConn, "select id, num, col from s_tbl order by id", `[[INT64(1) INT64(10) NULL] [INT64(3) INT64(20) NULL]]`)
 	utils.AssertMatches(t, mcmp.VtConn, "select num, hex(keyspace_id) from num_vdx_tbl order by num", `[[INT64(10) VARCHAR("166B40B44ABA4BD6")] [INT64(20) VARCHAR("4EB190C9A2FA169C")]]`)
 	utils.AssertMatches(t, mcmp.VtConn, "select col, hex(keyspace_id) from col_vdx_tbl order by col", `[[INT64(30) VARCHAR("4EB190C9A2FA169C")]]`)
 

go/test/endtoend/vtgate/queries/dml/insert_test.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/stretchr/testify/assert"
 
 	"vitess.io/vitess/go/test/endtoend/utils"
@@ -518,3 +520,28 @@ func TestInsertJson(t *testing.T) {
 
 	mcmp.Exec(`insert into j_tbl(id, jdoc) values (10, '{}'), (20, "[]")`)
 }
+
+func TestInsertIgnoreNullAndInsertNull(t *testing.T) {
+	mcmp, closer := start(t)
+	defer closer()
+
+	mcmp.Exec("insert ignore into s_tbl(id) values (1)") // the remaining columns are be set to NULL
+	mcmp.Exec("select id, num, col from s_tbl")
+	mcmp.Exec("insert ignore into s_tbl(id, num, col) values (2, 2, 2), (3, NULL, 3), (4, 4, NULL)")
+	mcmp.Exec("insert into s_tbl(id, num, col) values (5, 5, 5), (6, NULL, 6), (7, 7, NULL)")
+	mcmp.Exec("select id, num, col from s_tbl")
+	utils.AssertMatches(t, mcmp.VtConn, "select num, hex(keyspace_id) from num_vdx_tbl order by num",
+		`[[INT64(2) VARCHAR("06E7EA22CE92708F")] [INT64(4) VARCHAR("D2FD8867D50D2DFE")] [INT64(5) VARCHAR("70BB023C810CA87A")] [INT64(7) VARCHAR("FB8BAAAD918119B8")]]`)
+	utils.AssertMatches(t, mcmp.VtConn, "select col, id, hex(keyspace_id) from col_vdx_tbl order by col, id",
+		`[[INT64(2) INT64(2) VARCHAR("06E7EA22CE92708F")] [INT64(3) INT64(3) VARCHAR("4EB190C9A2FA169C")] [INT64(5) INT64(5) VARCHAR("70BB023C810CA87A")] [INT64(6) INT64(6) VARCHAR("F098480AC4C4BE71")]]`)
+
+	mcmp.Exec("insert ignore into name_tbl(id, name) values (1, 'foo'), (2, 'bar')")
+	mcmp.Exec("select id, name from name_tbl order by id")
+
+	// This should contain the one row that was inserted above
+	utils.AssertMatches(t, mcmp.VtConn, "select name, id, hex(keyspace_id) from name_vdx_tbl order by name, id",
+		`[[VARCHAR("bar") INT64(2) VARCHAR("06E7EA22CE92708F")] [VARCHAR("foo") INT64(1) VARCHAR("166B40B44ABA4BD6")]]`)
+
+	_, err := utils.ExecAllowError(t, mcmp.VtConn, "insert ignore into name_tbl(id, name) values (22, NULL)") // this should fail, since we have a vindex that does not ignore nulls
+	require.ErrorContains(t, err, "Column 'name,id' cannot be null")
+}

go/test/endtoend/vtgate/queries/dml/sharded_schema.sql

@@ -1,12 +1,17 @@
 create table s_tbl
 (
-    id  bigint,
-    num bigint,
-    col bigint,
+    id   bigint,
+    num  bigint,
+    col  bigint,
     unique key (num),
     primary key (id)
 ) Engine = InnoDB;
-
+create table name_tbl
+(
+    id   bigint,
+    name varchar(50),
+    primary key (id)
+) Engine = InnoDB;
 create table num_vdx_tbl
 (
     num         bigint,
@@ -20,6 +25,14 @@ create table col_vdx_tbl
     id          bigint,
     keyspace_id varbinary(20),
     primary key (col, id)
+) Engine = InnoDB
+;
+create table name_vdx_tbl
+(
+    name        varchar(50),
+    id          bigint,
+    keyspace_id varbinary(20),
+    primary key (name, id)
 ) Engine = InnoDB;
 
 create table user_tbl
@@ -100,7 +113,7 @@ create table lkp_mixed_idx
 
 create table j_tbl
 (
-    id  bigint,
+    id   bigint,
     jdoc json,
     primary key (id)
 ) Engine = InnoDB;

go/test/endtoend/vtgate/queries/dml/vschema.json

@@ -4,6 +4,9 @@
     "hash": {
       "type": "hash"
     },
+    "hash_varchar": {
+      "type": "unicode_loose_xxhash"
+    },
     "num_vdx": {
       "type": "consistent_lookup_unique",
       "params": {
@@ -24,6 +27,15 @@
       },
       "owner": "s_tbl"
     },
+    "name_vdx": {
+      "type": "consistent_lookup",
+      "params": {
+        "table": "name_vdx_tbl",
+        "from": "name,id",
+        "to": "keyspace_id"
+      },
+      "owner": "name_tbl"
+    },
     "oid_vdx": {
       "type": "consistent_lookup_unique",
       "params": {
@@ -76,11 +88,29 @@
           "name": "num_vdx"
         },
         {
-          "columns": ["col", "id"],
+          "columns": [
+            "col",
+            "id"
+          ],
           "name": "col_vdx"
         }
       ]
     },
+    "name_tbl": {
+      "column_vindexes": [
+        {
+          "column": "id",
+          "name": "hash"
+        },
+        {
+          "columns": [
+            "name",
+            "id"
+          ],
+          "name": "name_vdx"
+        }
+      ]
+    },
     "num_vdx_tbl": {
       "column_vindexes": [
         {
@@ -97,6 +127,14 @@
         }
       ]
     },
+    "name_vdx_tbl": {
+      "column_vindexes": [
+        {
+          "column": "name",
+          "name": "hash_varchar"
+        }
+      ]
+    },
     "user_tbl": {
       "auto_increment": {
         "column": "id",

go/vt/vtgate/engine/insert_common.go

@@ -230,14 +230,26 @@ func (ic *InsertCommon) processOwned(ctx context.Context, vcursor VCursor, vinde
 	var createIndexes []int
 	var createKeys []sqltypes.Row
 	var createKsids []ksID
+	var vindexNull []bool
 
 	for rowNum, rowColumnKeys := range vindexColumnsKeys {
 		if ksids[rowNum] == nil {
 			continue
 		}
+		keyContainsNull := false
+		for _, columnKey := range rowColumnKeys {
+			if columnKey.IsNull() {
+				// if any of the keys contains a null, we know the vindex will ignore this row,
+				// so it's safe to
+				keyContainsNull = true
+				break
+			}
+		}
+
 		createIndexes = append(createIndexes, rowNum)
 		createKeys = append(createKeys, rowColumnKeys)
 		createKsids = append(createKsids, ksids[rowNum])
+		vindexNull = append(vindexNull, keyContainsNull)
 	}
 	if createKeys == nil {
 		return nil
@@ -254,7 +266,7 @@ func (ic *InsertCommon) processOwned(ctx context.Context, vcursor VCursor, vinde
 		return err
 	}
 	for i, v := range verified {
-		if !v {
+		if !v && !vindexNull[i] {
 			ksids[createIndexes[i]] = nil
 		}
 	}