diff --git a/go/mysqlconn/replication/binlog_event_json.go b/go/mysqlconn/replication/binlog_event_json.go
new file mode 100644
index 00000000000..8273091e700
--- /dev/null
+++ b/go/mysqlconn/replication/binlog_event_json.go
@@ -0,0 +1,473 @@
+package replication
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"math"
+	"strconv"
+
+	"github.com/youtube/vitess/go/sqltypes"
+	querypb "github.com/youtube/vitess/go/vt/proto/query"
+)
+
+const (
+	jsonTypeSmallObject = 0
+	jsonTypeLargeObject = 1
+	jsonTypeSmallArray  = 2
+	jsonTypeLargeArray  = 3
+	jsonTypeLiteral     = 4
+	jsonTypeInt16       = 5
+	jsonTypeUint16      = 6
+	jsonTypeInt32       = 7
+	jsonTypeUint32      = 8
+	jsonTypeInt64       = 9
+	jsonTypeUint64      = 10
+	jsonTypeDouble      = 11
+	jsonTypeString      = 12
+	jsonTypeOpaque      = 15
+
+	jsonNullLiteral  = '\x00'
+	jsonTrueLiteral  = '\x01'
+	jsonFalseLiteral = '\x02'
+)
+
+// printJSONData parses the MySQL binary format for JSON data, and prints
+// the result as a string.
+func printJSONData(data []byte) ([]byte, error) {
+	result := &bytes.Buffer{}
+	typ := data[0]
+	if err := printJSONValue(typ, data[1:], true /* toplevel */, result); err != nil {
+		return nil, err
+	}
+	return result.Bytes(), nil
+}
+
+func printJSONValue(typ byte, data []byte, toplevel bool, result *bytes.Buffer) error {
+	switch typ {
+	case jsonTypeSmallObject:
+		return printJSONObject(data, false, result)
+	case jsonTypeLargeObject:
+		return printJSONObject(data, true, result)
+	case jsonTypeSmallArray:
+		return printJSONArray(data, false, result)
+	case jsonTypeLargeArray:
+		return printJSONArray(data, true, result)
+	case jsonTypeLiteral:
+		return printJSONLiteral(data[0], toplevel, result)
+	case jsonTypeInt16:
+		printJSONInt16(data[0:2], toplevel, result)
+	case jsonTypeUint16:
+		printJSONUint16(data[0:2], toplevel, result)
+	case jsonTypeInt32:
+		printJSONInt32(data[0:4], toplevel, result)
+	case jsonTypeUint32:
+		printJSONUint32(data[0:4], toplevel, result)
+	case jsonTypeInt64:
+		printJSONInt64(data[0:8], toplevel, result)
+	case jsonTypeUint64:
+		printJSONUint64(data[0:8], toplevel, result)
+	case jsonTypeDouble:
+		printJSONDouble(data[0:8], toplevel, result)
+	case jsonTypeString:
+		printJSONString(data, toplevel, result)
+	case jsonTypeOpaque:
+		return printJSONOpaque(data, toplevel, result)
+	default:
+		return fmt.Errorf("unknown object type in JSON: %v", typ)
+	}
+
+	return nil
+}
+
+func printJSONObject(data []byte, large bool, result *bytes.Buffer) error {
+	pos := 0
+	elementCount, pos := readOffsetOrSize(data, pos, large)
+	size, pos := readOffsetOrSize(data, pos, large)
+	if size > len(data) {
+		return fmt.Errorf("not enough data for object, have %v bytes need %v", len(data), size)
+	}
+
+	// Build an array for each key.
+	keys := make([]sqltypes.Value, elementCount)
+	for i := 0; i < elementCount; i++ {
+		var keyOffset, keyLength int
+		keyOffset, pos = readOffsetOrSize(data, pos, large)
+		keyLength, pos = readOffsetOrSize(data, pos, false) // always 16
+		keys[i] = sqltypes.MakeTrusted(sqltypes.VarBinary, data[keyOffset:keyOffset+keyLength])
+	}
+
+	// Now read each value, and output them.  The value entry is
+	// always one byte (the type), and then 2 or 4 bytes
+	// (depending on the large flag). If the value fits in the number of bytes,
+	// then it is inlined. This is always the case for Literal (one byte),
+	// and {,u}int16. For {u}int32, it depends if we're large or not.
+	result.WriteString("JSON_OBJECT(")
+	for i := 0; i < elementCount; i++ {
+		// First print the key value.
+		if i > 0 {
+			result.WriteByte(',')
+		}
+		keys[i].EncodeSQL(result)
+		result.WriteByte(',')
+
+		if err := printJSONValueEntry(data, pos, large, result); err != nil {
+			return err
+		}
+		if large {
+			pos += 5 // type byte + 4 bytes
+		} else {
+			pos += 3 // type byte + 2 bytes
+		}
+	}
+	result.WriteByte(')')
+	return nil
+}
+
+func printJSONArray(data []byte, large bool, result *bytes.Buffer) error {
+	pos := 0
+	elementCount, pos := readOffsetOrSize(data, pos, large)
+	size, pos := readOffsetOrSize(data, pos, large)
+	if size > len(data) {
+		return fmt.Errorf("not enough data for object, have %v bytes need %v", len(data), size)
+	}
+
+	// Now read each value, and output them.  The value entry is
+	// always one byte (the type), and then 2 or 4 bytes
+	// (depending on the large flag). If the value fits in the number of bytes,
+	// then it is inlined. This is always the case for Literal (one byte),
+	// and {,u}int16. For {u}int32, it depends if we're large or not.
+	result.WriteString("JSON_ARRAY(")
+	for i := 0; i < elementCount; i++ {
+		// Print the key value.
+		if i > 0 {
+			result.WriteByte(',')
+		}
+		if err := printJSONValueEntry(data, pos, large, result); err != nil {
+			return err
+		}
+		if large {
+			pos += 5 // type byte + 4 bytes
+		} else {
+			pos += 3 // type byte + 2 bytes
+		}
+	}
+	result.WriteByte(')')
+	return nil
+}
+
+// printJSONValueEntry prints an entry. The value entry is always one
+// byte (the type), and then 2 or 4 bytes (depending on the large
+// flag). If the value fits in the number of bytes, then it is
+// inlined. This is always the case for Literal (one byte), and
+// {,u}int16. For {u}int32, it depends if we're large or not.
+func printJSONValueEntry(data []byte, pos int, large bool, result *bytes.Buffer) error {
+	typ := data[pos]
+	pos++
+
+	switch {
+	case typ == jsonTypeLiteral:
+		// 3 possible literal values, always in-lined, as it is one byte.
+		if err := printJSONLiteral(data[pos], false /* toplevel */, result); err != nil {
+			return err
+		}
+	case typ == jsonTypeInt16:
+		// Value is always inlined in first 2 bytes.
+		printJSONInt16(data[pos:pos+2], false /* toplevel */, result)
+	case typ == jsonTypeUint16:
+		// Value is always inlined in first 2 bytes.
+		printJSONUint16(data[pos:pos+2], false /* toplevel */, result)
+	case typ == jsonTypeInt32 && large:
+		// Value is only inlined if large.
+		printJSONInt32(data[pos:pos+4], false /* toplevel */, result)
+	case typ == jsonTypeUint32 && large:
+		// Value is only inlined if large.
+		printJSONUint32(data[pos:pos+4], false /* toplevel */, result)
+	default:
+		// value is not inlined, we have its offset here.
+		// Note we don't have its length, so we just go to the end.
+		offset, _ := readOffsetOrSize(data, pos, large)
+		if err := printJSONValue(typ, data[offset:], false /* toplevel */, result); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func printJSONLiteral(b byte, toplevel bool, result *bytes.Buffer) error {
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	// Only three possible values.
+	switch b {
+	case jsonNullLiteral:
+		result.WriteString("null")
+	case jsonTrueLiteral:
+		result.WriteString("true")
+	case jsonFalseLiteral:
+		result.WriteString("false")
+	default:
+		return fmt.Errorf("unknown literal value %v", b)
+	}
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	return nil
+}
+
+func printJSONInt16(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := uint16(data[0]) +
+		uint16(data[1])<<8
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendInt(nil, int64(int16(val)), 10))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONUint16(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := uint16(data[0]) +
+		uint16(data[1])<<8
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendUint(nil, uint64(val), 10))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONInt32(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := uint32(data[0]) +
+		uint32(data[1])<<8 +
+		uint32(data[2])<<16 +
+		uint32(data[3])<<24
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendInt(nil, int64(int32(val)), 10))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONUint32(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := uint32(data[0]) +
+		uint32(data[1])<<8 +
+		uint32(data[2])<<16 +
+		uint32(data[3])<<24
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendUint(nil, uint64(val), 10))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONInt64(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := uint64(data[0]) +
+		uint64(data[1])<<8 +
+		uint64(data[2])<<16 +
+		uint64(data[3])<<24 +
+		uint64(data[4])<<32 +
+		uint64(data[5])<<40 +
+		uint64(data[6])<<48 +
+		uint64(data[7])<<56
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendInt(nil, int64(val), 10))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONUint64(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := binary.LittleEndian.Uint64(data[:8])
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendUint(nil, val, 10))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONDouble(data []byte, toplevel bool, result *bytes.Buffer) {
+	val := binary.LittleEndian.Uint64(data[:8])
+	fval := math.Float64frombits(val)
+	if toplevel {
+		result.WriteByte('\'')
+	}
+	result.Write(strconv.AppendFloat(nil, fval, 'E', -1, 64))
+	if toplevel {
+		result.WriteByte('\'')
+	}
+}
+
+func printJSONString(data []byte, toplevel bool, result *bytes.Buffer) {
+	size, pos := readVariableInt(data, 0)
+
+	// A toplevel JSON string is printed as a JSON-escaped
+	// string inside a string, as the value is parsed as JSON.
+	// So the value should be: '"value"'.
+	if toplevel {
+		result.WriteString("'\"")
+		// FIXME(alainjobart): escape reserved characters
+		result.Write(data[pos : pos+size])
+		result.WriteString("\"'")
+		return
+	}
+
+	// Inside a JSON_ARRAY() or JSON_OBJECT method, we just print the string
+	// as SQL string.
+	valStr := sqltypes.MakeTrusted(sqltypes.VarBinary, data[pos:pos+size])
+	valStr.EncodeSQL(result)
+}
+
+func printJSONOpaque(data []byte, toplevel bool, result *bytes.Buffer) error {
+	typ := data[0]
+	size, pos := readVariableInt(data, 1)
+
+	// A few types have special encoding.
+	switch typ {
+	case TypeDate:
+		return printJSONDate(data[pos:pos+size], toplevel, result)
+	case TypeTime:
+		return printJSONTime(data[pos:pos+size], toplevel, result)
+	case TypeDateTime:
+		return printJSONDateTime(data[pos:pos+size], toplevel, result)
+	case TypeNewDecimal:
+		return printJSONDecimal(data[pos:pos+size], toplevel, result)
+	}
+
+	// Other types are encoded in somewhat weird ways. Since we
+	// have no metadata, it seems some types first provide the
+	// metadata, and then the values. But even that metadata is
+	// not straightforward (for instance, a bit field seems to
+	// have one byte as metadata, not two as would be expected).
+	// To be on the safer side, we just reject these cases for now.
+	return fmt.Errorf("opaque type %v is not supported yet, with data %v", typ, data[1:])
+}
+
+func printJSONDate(data []byte, toplevel bool, result *bytes.Buffer) error {
+	raw := binary.LittleEndian.Uint64(data[:8])
+	value := raw >> 24
+	yearMonth := (value >> 22) & 0x01ffff // 17 bits starting at 22nd
+	year := yearMonth / 13
+	month := yearMonth % 13
+	day := (value >> 17) & 0x1f // 5 bits starting at 17th
+
+	if toplevel {
+		result.WriteString("CAST(")
+	}
+	fmt.Fprintf(result, "CAST('%04d-%02d-%02d' AS DATE)", year, month, day)
+	if toplevel {
+		result.WriteString(" AS JSON)")
+	}
+	return nil
+}
+
+func printJSONTime(data []byte, toplevel bool, result *bytes.Buffer) error {
+	raw := binary.LittleEndian.Uint64(data[:8])
+	value := raw >> 24
+	hour := (value >> 12) & 0x03ff // 10 bits starting at 12th
+	minute := (value >> 6) & 0x3f  // 6 bits starting at 6th
+	second := value & 0x3f         // 6 bits starting at 0th
+	microSeconds := raw & 0xffffff // 24 lower bits
+
+	if toplevel {
+		result.WriteString("CAST(")
+	}
+	result.WriteString("CAST('")
+	if value&0x8000000000 != 0 {
+		result.WriteByte('-')
+	}
+	fmt.Fprintf(result, "%02d:%02d:%02d", hour, minute, second)
+	if microSeconds != 0 {
+		fmt.Fprintf(result, ".%06d", microSeconds)
+	}
+	result.WriteString("' AS TIME(6))")
+	if toplevel {
+		result.WriteString(" AS JSON)")
+	}
+	return nil
+}
+
+func printJSONDateTime(data []byte, toplevel bool, result *bytes.Buffer) error {
+	raw := binary.LittleEndian.Uint64(data[:8])
+	value := raw >> 24
+	yearMonth := (value >> 22) & 0x01ffff // 17 bits starting at 22nd
+	year := yearMonth / 13
+	month := yearMonth % 13
+	day := (value >> 17) & 0x1f    // 5 bits starting at 17th
+	hour := (value >> 12) & 0x1f   // 5 bits starting at 12th
+	minute := (value >> 6) & 0x3f  // 6 bits starting at 6th
+	second := value & 0x3f         // 6 bits starting at 0th
+	microSeconds := raw & 0xffffff // 24 lower bits
+
+	if toplevel {
+		result.WriteString("CAST(")
+	}
+	fmt.Fprintf(result, "CAST('%04d-%02d-%02d %02d:%02d:%02d", year, month, day, hour, minute, second)
+	if microSeconds != 0 {
+		fmt.Fprintf(result, ".%06d", microSeconds)
+	}
+	result.WriteString("' AS DATETIME(6))")
+	if toplevel {
+		result.WriteString(" AS JSON)")
+	}
+	return nil
+}
+
+func printJSONDecimal(data []byte, toplevel bool, result *bytes.Buffer) error {
+	// Precision and scale are first (as there is no metadata)
+	// then we use the same decoding.
+	precision := data[0]
+	scale := data[1]
+	metadata := (uint16(precision) << 8) + uint16(scale)
+	val, _, err := CellValue(data, 2, TypeNewDecimal, metadata, querypb.Type_DECIMAL)
+	if err != nil {
+		return err
+	}
+	if toplevel {
+		result.WriteString("CAST(")
+	}
+	result.WriteString("CAST('")
+	result.Write(val.Raw())
+	fmt.Fprintf(result, "' AS DECIMAL(%d,%d))", precision, scale)
+	if toplevel {
+		result.WriteString(" AS JSON)")
+	}
+	return nil
+}
+
+func readOffsetOrSize(data []byte, pos int, large bool) (int, int) {
+	if large {
+		return int(data[pos]) +
+				int(data[pos+1])<<8 +
+				int(data[pos+2])<<16 +
+				int(data[pos+3])<<24,
+			pos + 4
+	}
+	return int(data[pos]) +
+		int(data[pos+1])<<8, pos + 2
+}
+
+func readVariableInt(data []byte, pos int) (int, int) {
+	var b byte
+	var result int
+	for {
+		b = data[pos]
+		pos++
+		result = (result << 7) + int(b&0x7f)
+		if b >= 0 {
+			break
+		}
+	}
+	return result, pos
+}
diff --git a/go/mysqlconn/replication/binlog_event_json_test.go b/go/mysqlconn/replication/binlog_event_json_test.go
new file mode 100644
index 00000000000..71bdfe8fe04
--- /dev/null
+++ b/go/mysqlconn/replication/binlog_event_json_test.go
@@ -0,0 +1,122 @@
+package replication
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestJSON(t *testing.T) {
+	testcases := []struct {
+		data     []byte
+		expected string
+	}{{
+		data:     []byte{0, 1, 0, 14, 0, 11, 0, 1, 0, 12, 12, 0, 97, 1, 98},
+		expected: `JSON_OBJECT('a','b')`,
+	}, {
+		data:     []byte{0, 1, 0, 12, 0, 11, 0, 1, 0, 5, 2, 0, 97},
+		expected: `JSON_OBJECT('a',2)`,
+	}, {
+		data:     []byte{2, 2, 0, 10, 0, 5, 1, 0, 5, 2, 0},
+		expected: `JSON_ARRAY(1,2)`,
+	}, {
+		data:     []byte{0, 4, 0, 60, 0, 32, 0, 1, 0, 33, 0, 1, 0, 34, 0, 2, 0, 36, 0, 2, 0, 12, 38, 0, 12, 40, 0, 12, 42, 0, 2, 46, 0, 97, 99, 97, 98, 98, 99, 1, 98, 1, 100, 3, 97, 98, 99, 2, 0, 14, 0, 12, 10, 0, 12, 12, 0, 1, 120, 1, 121},
+		expected: `JSON_OBJECT('a','b','c','d','ab','abc','bc',JSON_ARRAY('x','y'))`,
+	}, {
+		data:     []byte{2, 3, 0, 37, 0, 12, 13, 0, 2, 18, 0, 12, 33, 0, 4, 104, 101, 114, 101, 2, 0, 15, 0, 12, 10, 0, 12, 12, 0, 1, 73, 2, 97, 109, 3, 33, 33, 33},
+		expected: `JSON_ARRAY('here',JSON_ARRAY('I','am'),'!!!')`,
+	}, {
+		data:     []byte{12, 13, 115, 99, 97, 108, 97, 114, 32, 115, 116, 114, 105, 110, 103},
+		expected: `'"scalar string"'`,
+	}, {
+		data:     []byte{4, 1},
+		expected: `'true'`,
+	}, {
+		data:     []byte{4, 2},
+		expected: `'false'`,
+	}, {
+		data:     []byte{4, 0},
+		expected: `'null'`,
+	}, {
+		data:     []byte{5, 255, 255},
+		expected: `'-1'`,
+	}, {
+		data:     []byte{6, 1, 0},
+		expected: `'1'`,
+	}, {
+		data:     []byte{5, 255, 127},
+		expected: `'32767'`,
+	}, {
+		data:     []byte{7, 0, 128, 0, 0},
+		expected: `'32768'`,
+	}, {
+		data:     []byte{5, 0, 128},
+		expected: `'-32768'`,
+	}, {
+		data:     []byte{7, 255, 127, 255, 255},
+		expected: `'-32769'`,
+	}, {
+		data:     []byte{7, 255, 255, 255, 127},
+		expected: `'2147483647'`,
+	}, {
+		data:     []byte{9, 0, 0, 0, 128, 0, 0, 0, 0},
+		expected: `'2147483648'`,
+	}, {
+		data:     []byte{7, 0, 0, 0, 128},
+		expected: `'-2147483648'`,
+	}, {
+		data:     []byte{9, 255, 255, 255, 127, 255, 255, 255, 255},
+		expected: `'-2147483649'`,
+	}, {
+		data:     []byte{10, 255, 255, 255, 255, 255, 255, 255, 255},
+		expected: `'18446744073709551615'`,
+	}, {
+		data:     []byte{9, 0, 0, 0, 0, 0, 0, 0, 128},
+		expected: `'-9223372036854775808'`,
+	}, {
+		data:     []byte{11, 110, 134, 27, 240, 249, 33, 9, 64},
+		expected: `'3.14159E+00'`,
+	}, {
+		data:     []byte{0, 0, 0, 4, 0},
+		expected: `JSON_OBJECT()`,
+	}, {
+		data:     []byte{2, 0, 0, 4, 0},
+		expected: `JSON_ARRAY()`,
+	}, {
+		// opaque, datetime
+		data:     []byte{15, 12, 8, 0, 0, 0, 25, 118, 31, 149, 25},
+		expected: `CAST(CAST('2015-01-15 23:24:25' AS DATETIME(6)) AS JSON)`,
+	}, {
+		// opaque, time
+		data:     []byte{15, 11, 8, 0, 0, 0, 25, 118, 1, 0, 0},
+		expected: `CAST(CAST('23:24:25' AS TIME(6)) AS JSON)`,
+	}, {
+		// opaque, time
+		data:     []byte{15, 11, 8, 192, 212, 1, 25, 118, 1, 0, 0},
+		expected: `CAST(CAST('23:24:25.120000' AS TIME(6)) AS JSON)`,
+	}, {
+		// opaque, date
+		data:     []byte{15, 10, 8, 0, 0, 0, 0, 0, 30, 149, 25},
+		expected: `CAST(CAST('2015-01-15' AS DATE) AS JSON)`,
+	}, {
+		// opaque, decimal
+		data:     []byte{15, 246, 8, 13, 4, 135, 91, 205, 21, 4, 210},
+		expected: `CAST(CAST('123456789.1234' AS DECIMAL(13,4)) AS JSON)`,
+	}, {
+		// opaque, bit field. Not yet implemented.
+		data:     []byte{15, 16, 2, 202, 254},
+		expected: `ERROR: opaque type 16 is not supported yet, with data [2 202 254]`,
+	}}
+
+	for _, tcase := range testcases {
+		r, err := printJSONData(tcase.data)
+		got := ""
+		if err != nil {
+			got = fmt.Sprintf("ERROR: %v", err)
+		} else {
+			got = string(r)
+		}
+		if got != tcase.expected {
+			t.Errorf("unexpected output for %v: got %v expected %v", tcase.data, got, tcase.expected)
+		}
+	}
+}
diff --git a/go/mysqlconn/replication/binlog_event_rbr.go b/go/mysqlconn/replication/binlog_event_rbr.go
index d23f7f765c6..03993dd54b2 100644
--- a/go/mysqlconn/replication/binlog_event_rbr.go
+++ b/go/mysqlconn/replication/binlog_event_rbr.go
@@ -223,14 +223,6 @@ func cellLength(data []byte, pos int, typ byte, metadata uint16) (int, error) {
 		// metadata has number of decimals. One byte encodes
 		// two decimals.
 		return 3 + (int(metadata)+1)/2, nil
-	case TypeJSON:
-		// length in encoded in 'meta' bytes, but at least 2,
-		// and the value cannot be > 64k, so just read 2 bytes.
-		// (meta also should have '2' as value).
-		// (this weird logic is what event printing does).
-		l := int(uint64(data[pos]) |
-			uint64(data[pos+1])<<8)
-		return l + int(metadata), nil
 	case TypeNewDecimal:
 		precision := int(metadata >> 8)
 		scale := int(metadata & 0xff)
@@ -259,8 +251,8 @@ func cellLength(data []byte, pos int, typ byte, metadata uint16) (int, error) {
 		return intg0*4 + dig2bytes[intg0x] + frac0*4 + dig2bytes[frac0x], nil
 	case TypeEnum, TypeSet:
 		return int(metadata & 0xff), nil
-	case TypeTinyBlob, TypeMediumBlob, TypeLongBlob, TypeBlob, TypeGeometry:
-		// of the Blobs, only TypeBlob is used in binary logs,
+	case TypeJSON, TypeTinyBlob, TypeMediumBlob, TypeLongBlob, TypeBlob, TypeGeometry:
+		// Of the Blobs, only TypeBlob is used in binary logs,
 		// but supports others just in case.
 		switch metadata {
 		case 1:
@@ -625,20 +617,6 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
 		return sqltypes.MakeTrusted(querypb.Type_TIME,
 			[]byte(fmt.Sprintf("%v%02d:%02d:%02d%v", sign, hour, minute, second, fracStr))), 3 + (int(metadata)+1)/2, nil
 
-	case TypeJSON:
-		l := int(uint64(data[pos]) |
-			uint64(data[pos+1])<<8)
-		// length in encoded in 'meta' bytes, but at least 2,
-		// and the value cannot be > 64k, so just read 2 bytes.
-		// (meta also should have '2' as value).
-		// (this weird logic is what event printing does).
-
-		// TODO(alainjobart) the binary data for JSON should
-		// be parsed, and re-printed as JSON. This is a large
-		// project, as the binary version of the data is
-		// somewhat complex. For now, just return NULL.
-		return sqltypes.NULL, l + int(metadata), nil
-
 	case TypeNewDecimal:
 		precision := int(metadata >> 8) // total digits number
 		scale := int(metadata & 0xff)   // number of fractional digits
@@ -786,7 +764,7 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
 		return sqltypes.MakeTrusted(querypb.Type_SET,
 			data[pos:pos+l]), l, nil
 
-	case TypeTinyBlob, TypeMediumBlob, TypeLongBlob, TypeBlob:
+	case TypeJSON, TypeTinyBlob, TypeMediumBlob, TypeLongBlob, TypeBlob:
 		// Only TypeBlob is used in binary logs,
 		// but supports others just in case.
 		l := 0
@@ -809,6 +787,17 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
 			return sqltypes.NULL, 0, fmt.Errorf("unsupported blob metadata value %v (data: %v pos: %v)", metadata, data, pos)
 		}
 		pos += int(metadata)
+
+		// For JSON, we parse the data, and emit SQL.
+		if typ == TypeJSON {
+			d, err := printJSONData(data[pos : pos+l])
+			if err != nil {
+				return sqltypes.NULL, 0, fmt.Errorf("error parsing JSON data %v: %v", data[pos:pos+l], err)
+			}
+			return sqltypes.MakeTrusted(sqltypes.TypeSQL,
+				d), l + int(metadata), nil
+		}
+
 		return sqltypes.MakeTrusted(querypb.Type_VARBINARY,
 			data[pos:pos+l]), l + int(metadata), nil
 
diff --git a/go/mysqlconn/replication/binlog_event_rbr_test.go b/go/mysqlconn/replication/binlog_event_rbr_test.go
index 8d3b1972eb9..ceb55ed16c2 100644
--- a/go/mysqlconn/replication/binlog_event_rbr_test.go
+++ b/go/mysqlconn/replication/binlog_event_rbr_test.go
@@ -387,8 +387,17 @@ func TestCellLengthAndData(t *testing.T) {
 	}, {
 		typ:      TypeJSON,
 		metadata: 2,
-		data:     []byte{0x03, 0x00, 'a', 'b', 'c'},
-		out:      sqltypes.NULL,
+		data: []byte{0x0f, 0x00,
+			0, 1, 0, 14, 0, 11, 0, 1, 0, 12, 12, 0, 97, 1, 98},
+		out: sqltypes.MakeTrusted(sqltypes.TypeSQL,
+			[]byte(`JSON_OBJECT('a','b')`)),
+	}, {
+		typ:      TypeJSON,
+		metadata: 4,
+		data: []byte{0x0f, 0x00, 0x00, 0x00,
+			0, 1, 0, 14, 0, 11, 0, 1, 0, 12, 12, 0, 97, 1, 98},
+		out: sqltypes.MakeTrusted(sqltypes.TypeSQL,
+			[]byte(`JSON_OBJECT('a','b')`)),
 	}, {
 		typ:      TypeEnum,
 		metadata: 1,
diff --git a/go/mysqlconn/replication_test.go b/go/mysqlconn/replication_test.go
index 0b9b802e874..05d4070e4bf 100644
--- a/go/mysqlconn/replication_test.go
+++ b/go/mysqlconn/replication_test.go
@@ -1021,17 +1021,139 @@ func testRowReplicationTypesWithRealDatabase(t *testing.T, params *sqldb.ConnPar
 	// JSON is only supported by MySQL 5.7+
 	// However the binary format is not just the text version.
 	// So it doesn't work as expected.
-	if false && strings.HasPrefix(conn.ServerVersion, "5.7") {
-		testcases = append(testcases, struct {
+	if strings.HasPrefix(conn.ServerVersion, "5.7") {
+		testcases = append(testcases, []struct {
 			name        string
 			createType  string
 			createValue string
-		}{
-			// JSON
+		}{{
 			name:        "json1",
 			createType:  "JSON",
-			createValue: "'{\"a\":\"b\"}'",
-		})
+			createValue: "'{\"a\": 2}'",
+		}, {
+			name:        "json2",
+			createType:  "JSON",
+			createValue: "'[1,2]'",
+		}, {
+			name:        "json3",
+			createType:  "JSON",
+			createValue: "'{\"a\":\"b\", \"c\":\"d\",\"ab\":\"abc\", \"bc\": [\"x\", \"y\"]}'",
+		}, {
+			name:        "json4",
+			createType:  "JSON",
+			createValue: "'[\"here\", [\"I\", \"am\"], \"!!!\"]'",
+		}, {
+			name:        "json5",
+			createType:  "JSON",
+			createValue: "'\"scalar string\"'",
+		}, {
+			name:        "json6",
+			createType:  "JSON",
+			createValue: "'true'",
+		}, {
+			name:        "json7",
+			createType:  "JSON",
+			createValue: "'false'",
+		}, {
+			name:        "json8",
+			createType:  "JSON",
+			createValue: "'null'",
+		}, {
+			name:        "json9",
+			createType:  "JSON",
+			createValue: "'-1'",
+		}, {
+			name:        "json10",
+			createType:  "JSON",
+			createValue: "CAST(CAST(1 AS UNSIGNED) AS JSON)",
+		}, {
+			name:        "json11",
+			createType:  "JSON",
+			createValue: "'32767'",
+		}, {
+			name:        "json12",
+			createType:  "JSON",
+			createValue: "'32768'",
+		}, {
+			name:        "json13",
+			createType:  "JSON",
+			createValue: "'-32768'",
+		}, {
+			name:        "json14",
+			createType:  "JSON",
+			createValue: "'-32769'",
+		}, {
+			name:        "json15",
+			createType:  "JSON",
+			createValue: "'2147483647'",
+		}, {
+			name:        "json16",
+			createType:  "JSON",
+			createValue: "'2147483648'",
+		}, {
+			name:        "json17",
+			createType:  "JSON",
+			createValue: "'-2147483648'",
+		}, {
+			name:        "json18",
+			createType:  "JSON",
+			createValue: "'-2147483649'",
+		}, {
+			name:        "json19",
+			createType:  "JSON",
+			createValue: "'18446744073709551615'",
+		}, {
+			name:        "json20",
+			createType:  "JSON",
+			createValue: "'18446744073709551616'",
+		}, {
+			name:        "json21",
+			createType:  "JSON",
+			createValue: "'3.14159'",
+		}, {
+			name:        "json22",
+			createType:  "JSON",
+			createValue: "'{}'",
+		}, {
+			name:        "json23",
+			createType:  "JSON",
+			createValue: "'[]'",
+		}, {
+			name:        "json24",
+			createType:  "JSON",
+			createValue: "CAST(CAST('2015-01-15 23:24:25' AS DATETIME) AS JSON)",
+		}, {
+			name:        "json25",
+			createType:  "JSON",
+			createValue: "CAST(CAST('23:24:25' AS TIME) AS JSON)",
+		}, {
+			name:        "json26",
+			createType:  "JSON",
+			createValue: "CAST(CAST('23:24:25.12' AS TIME(3)) AS JSON)",
+		}, {
+			name:        "json27",
+			createType:  "JSON",
+			createValue: "CAST(CAST('2015-01-15' AS DATE) AS JSON)",
+		}, {
+			name:        "json28",
+			createType:  "JSON",
+			createValue: "CAST(TIMESTAMP'2015-01-15 23:24:25' AS JSON)",
+		}, {
+			name:        "json29",
+			createType:  "JSON",
+			createValue: "CAST(ST_GeomFromText('POINT(1 1)') AS JSON)",
+		}, {
+			// Decimal has special treatment.
+			name:        "json30",
+			createType:  "JSON",
+			createValue: "CAST(CAST('123456789.1234' AS DECIMAL(13,4)) AS JSON)",
+			// FIXME(alainjobart) opaque types are complicated.
+			//		}, {
+			// This is a bit field. Opaque type in JSON.
+			//			name:        "json31",
+			//			createType:  "JSON",
+			//			createValue: "CAST(x'cafe' AS JSON)",
+		}}...)
 	}
 
 	ctx := context.Background()
@@ -1185,7 +1307,7 @@ func testRowReplicationTypesWithRealDatabase(t *testing.T, params *sqldb.ConnPar
 	}
 	for i, tcase := range testcases {
 		if !reflect.DeepEqual(result.Rows[0][i+1], result.Rows[1][i+1]) {
-			t.Errorf("Field %v is not the same, got %v(%v) and %v(%v)", tcase.name, result.Rows[0][i+1], result.Rows[0][i+1].Type, result.Rows[1][i+1], result.Rows[1][i+1].Type)
+			t.Errorf("Field %v is not the same, got %v(%v) and %v(%v)", tcase.name, result.Rows[0][i+1], result.Rows[0][i+1].Type(), result.Rows[1][i+1], result.Rows[1][i+1].Type())
 		}
 	}
 
diff --git a/go/sqltypes/type.go b/go/sqltypes/type.go
index aaa4c546e9d..8433f80cd7f 100644
--- a/go/sqltypes/type.go
+++ b/go/sqltypes/type.go
@@ -96,6 +96,10 @@ const (
 	Tuple     = querypb.Type_TUPLE
 	Geometry  = querypb.Type_GEOMETRY
 	TypeJSON  = querypb.Type_JSON
+
+	// TypeSQL is exposed here in the code, but not in the proto file.
+	// This is an internal type used for binlogs only.
+	TypeSQL = querypb.Type(245)
 )
 
 // bit-shift the mysql flags by two byte so we
diff --git a/proto/query.proto b/proto/query.proto
index 6d7b69e72cf..e92f523e96c 100644
--- a/proto/query.proto
+++ b/proto/query.proto
@@ -182,6 +182,11 @@ enum Type {
   // JSON specified a JSON type.
   // Properties: 30, IsQuoted.
   JSON = 2078;
+
+  // SQL specifies a SQL expression. This is not exposed, as it is only
+  // used in binlog parsing. There is no way to reserve a value in enums,
+  // so just using a comment here.
+  // Properties: 245, None.
 }
 
 // Value represents a typed value.