Fix bug with NULL JSON columns

We still needed to increment the json col index for those.

Signed-off-by: Matt Lord <mattalord@gmail.com>

go/mysql/binlog_event_rbr.go

@@ -435,6 +435,9 @@ func (rs *Rows) StringValuesForTests(tm *TableMap, rowIndex int) ([]string, erro
 			// This column is represented, but its value is NULL.
 			result = append(result, "NULL")
 			valueIndex++
+			if tm.Types[c] == binlog.TypeJSON {
+				jsonIndex++
+			}
 			continue
 		}
 

go/test/endtoend/vreplication/vreplication_test.go

@@ -721,15 +721,8 @@ func shardCustomer(t *testing.T, testReverse bool, cells []*Cell, sourceCellOrAl
 		// Confirm that the 0 scale decimal field, dec80, is replicated correctly
 		execVtgateQuery(t, vtgateConn, sourceKs, "update customer set dec80 = 0")
 		execVtgateQuery(t, vtgateConn, sourceKs, "update customer set blb = \"new blob data\" where cid=3")
-		// TODO: file a MySQL bug for this. The following query results in the quoted string literal "null"
-		// stored in the j3 column. But with and without the literal quotes, the value in the PARTIAL_JSON
-		// diff is the unquoted literal null which is a JSON null type. This leads to a vdiff mismatch.
-		// I'm not sure if the bug is that it allows the quoted value "null" to be inserted or that it
-		// doesn't reflect this in the partial diff value, but the combination of the two is certainly a bug.
-		//execVtgateQuery(t, vtgateConn, sourceKs, "update json_tbl set j1 = null, j2 = 'null', j3 = '\"null\"'")
-		//execVtgateQuery(t, vtgateConn, sourceKs, "insert into json_tbl(id, j1, j2, j3) values (7, null, 'null', '\"null\"')")
-		execVtgateQuery(t, vtgateConn, sourceKs, "update json_tbl set j1 = null, j2 = 'null', j3 = 'null'")
-		execVtgateQuery(t, vtgateConn, sourceKs, "insert into json_tbl(id, j1, j2, j3) values (7, null, 'null', 'null')")
+		execVtgateQuery(t, vtgateConn, sourceKs, "update json_tbl set j1 = null, j2 = 'null', j3 = '\"null\"'")
+		execVtgateQuery(t, vtgateConn, sourceKs, "insert into json_tbl(id, j1, j2, j3) values (7, null, 'null', '\"null\"')")
 		waitForNoWorkflowLag(t, vc, targetKs, workflow)
 		dec80Replicated := false
 		for _, tablet := range []*cluster.VttabletProcess{customerTab1, customerTab2} {

go/vt/vttablet/tabletserver/vstreamer/vstreamer.go

@@ -1111,6 +1111,9 @@ func (vs *vstreamer) extractRowAndFilter(plan *streamerPlan, data []byte, dataCo
 		}
 		if nullColumns.Bit(valueIndex) {
 			valueIndex++
+			if plan.Table.Fields[colNum].Type == querypb.Type_JSON {
+				jsonIndex++
+			}
 			continue
 		}
 		partialJSON := false