replication: support geometry (go-mysql-org#110)

Author: siddontang

replication/replication_test.go

@@ -211,6 +211,23 @@ func (t *testSyncerSuite) testSync(c *C, s *BinlogStreamer) {
 for _, query := range tbls {
 t.testExecute(c, query)
 }
+
+// If MySQL supports JSON, it must supports GEOMETRY.
+t.testExecute(c, "DROP TABLE IF EXISTS test_geo")
+
+str = `CREATE TABLE test_geo (g GEOMETRY)`
+_, err = t.c.Execute(str)
+c.Assert(err, IsNil)
+
+tbls = []string{
+`INSERT INTO test_geo VALUES (POINT(1, 1))`,
+`INSERT INTO test_geo VALUES (LINESTRING(POINT(0,0), POINT(1,1), POINT(2,2)))`,
+// TODO: add more geometry tests
+}
+
+for _, query := range tbls {
+t.testExecute(c, query)
+}
 }
 
 t.wg.Wait()

replication/row_event.go

@@ -456,26 +456,7 @@ func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16) (v interface{
 
 v, err = decodeBit(data, nbits, n)
 case MYSQL_TYPE_BLOB:
-switch meta {
-case 1:
-length = int(data[0])
-v = data[1 : 1+length]
-n = length + 1
-case 2:
-length = int(binary.LittleEndian.Uint16(data))
-v = data[2 : 2+length]
-n = length + 2
-case 3:
-length = int(FixedLengthInt(data[0:3]))
-v = data[3 : 3+length]
-n = length + 3
-case 4:
-length = int(binary.LittleEndian.Uint32(data))
-v = data[4 : 4+length]
-n = length + 4
-default:
-err = fmt.Errorf("invalid blob packlen = %d", meta)
-}
+v, n, err = decodeBlob(data, meta)
 case MYSQL_TYPE_VARCHAR,
 MYSQL_TYPE_VAR_STRING:
 length = int(meta)
@@ -487,6 +468,14 @@ func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16) (v interface{
 length = int(binary.LittleEndian.Uint32(data[0:]))
 n = length + int(meta)
 v, err = decodeJsonBinary(data[meta:n])
+case MYSQL_TYPE_GEOMETRY:
+// MySQL saves Geometry as Blob in binlog
+// Seem that the binary format is SRID (4 bytes) + WKB, outer can use
+// MySQL GeoFromWKB or others to create the geometry data.
+// Refer https://dev.mysql.com/doc/refman/5.7/en/gis-wkb-functions.html
+// I also find some go libs to handle WKB if possible
+// see https://github.com/twpayne/go-geom or https://github.com/paulmach/go.geo
+v, n, err = decodeBlob(data, meta)
 default:
 err = fmt.Errorf("unsupport type %d in binlog and don't know how to handle", tp)
 }
@@ -766,6 +755,32 @@ func decodeTime2(data []byte, dec uint16) (string, int, error) {
 return fmt.Sprintf("%s%02d:%02d:%02d", sign, hour, minute, second), n, nil
 }
 
+func decodeBlob(data []byte, meta uint16) (v []byte, n int, err error) {
+var length int
+switch meta {
+case 1:
+length = int(data[0])
+v = data[1 : 1+length]
+n = length + 1
+case 2:
+length = int(binary.LittleEndian.Uint16(data))
+v = data[2 : 2+length]
+n = length + 2
+case 3:
+length = int(FixedLengthInt(data[0:3]))
+v = data[3 : 3+length]
+n = length + 3
+case 4:
+length = int(binary.LittleEndian.Uint32(data))
+v = data[4 : 4+length]
+n = length + 4
+default:
+err = fmt.Errorf("invalid blob packlen = %d", meta)
+}
+
+return
+}
+
 func (e *RowsEvent) Dump(w io.Writer) {
 fmt.Fprintf(w, "TableID: %d\n", e.TableID)
 fmt.Fprintf(w, "Flags: %d\n", e.Flags)