Fix binary-column data corruption (#157)

This commit addresses a data corruption bug in the binlog streaming
phase, where the binlog writer incorrectly propagates values in
fixed-length BINARY columns that have trailing 0s in the original value.
These trailing 0s are removed from the binlog by the SQL master and
therefore do not show up in the WHERE clause for update/delete
statements executed by the binlog writer.

NOTE: This commit requires changes to one of the vendor'ed modules in

    github.com/siddontang/go-mysql

that we patch directly in the local repo. We are working on getting
these changes into the upstream module and will need to merge these
changes once we can use the latest upstream module version.

Change-Id: Ib9c1b7308e8198f1fd38439c37f444d9a8154e6a

dml_events.go

@@ -348,7 +348,13 @@ func appendEscapedValue(buffer []byte, value interface{}, column schema.TableCol
 
 	switch v := value.(type) {
 	case string:
-		return appendEscapedString(buffer, v)
+		var rightPadLengthForBinaryColumn int
+		// see appendEscapedString() for details why we need special
+		// handling of BINARY column types
+		if column.Type == schema.TYPE_BINARY {
+			rightPadLengthForBinaryColumn = int(column.FixedSize)
+		}
+		return appendEscapedString(buffer, v, rightPadLengthForBinaryColumn)
 	case []byte:
 		return appendEscapedBuffer(buffer, v, column.Type == schema.TYPE_JSON)
 	case bool:
@@ -362,7 +368,7 @@ func appendEscapedValue(buffer []byte, value interface{}, column schema.TableCol
 	case float32:
 		return strconv.AppendFloat(buffer, float64(v), 'g', -1, 64)
 	case decimal.Decimal:
-		return appendEscapedString(buffer, v.String())
+		return appendEscapedString(buffer, v.String(), 0)
 	default:
 		panic(fmt.Sprintf("unsupported type %t", value))
 	}
@@ -406,21 +412,48 @@ func Int64Value(value interface{}) (int64, bool) {
 //
 // ref: https://github.com/mysql/mysql-server/blob/mysql-5.7.5/mysys/charset.c#L963-L1038
 // ref: https://github.com/go-sql-driver/mysql/blob/9181e3a86a19bacd63e68d43ae8b7b36320d8092/utils.go#L717-L758
-func appendEscapedString(buffer []byte, value string) []byte {
+//
+// We need to support right-padding of the generated string using 0-bytes to
+// mimic what a MySQL server would do for BINARY columns (with fixed length).
+//
+// ref: https://github.com/Shopify/ghostferry/pull/159
+//
+// This is specifically mentioned in the the below link:
+//
+//    When BINARY values are stored, they are right-padded with the pad value
+//    to the specified length. The pad value is 0x00 (the zero byte). Values
+//    are right-padded with 0x00 for inserts, and no trailing bytes are removed
+//    for retrievals.
+//
+// ref: https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html
+func appendEscapedString(buffer []byte, value string, rightPadToLengthWithZeroBytes int) []byte {
 	buffer = append(buffer, '\'')
 
-	for i := 0; i < len(value); i++ {
+	var i int
+	for i = 0; i < len(value); i++ {
 		c := value[i]
 		if c == '\'' {
 			buffer = append(buffer, '\'', '\'')
 		} else {
 			buffer = append(buffer, c)
 		}
 	}
+	// continue 0-padding up to the desired length as provided by the
+	// caller
+	if i < rightPadToLengthWithZeroBytes {
+		buffer = rightPadBufferWithZeroBytes(buffer, rightPadToLengthWithZeroBytes-i)
+	}
 
 	return append(buffer, '\'')
 }
 
+func rightPadBufferWithZeroBytes(buffer []byte, padLength int) []byte {
+	for i := 0; i < padLength; i++ {
+		buffer = append(buffer, '\x00')
+	}
+	return buffer
+}
+
 func appendEscapedBuffer(buffer, value []byte, isJSON bool) []byte {
 	if isJSON {
 		// See https://bugs.mysql.com/bug.php?id=98496

test/integration/types_test.rb

@@ -268,6 +268,94 @@ def test_escaped_data
     end
   end
 
+  def test_decimal
+    # decimals are treated specially in binlog writing (they are inserted after
+    # conversion to string), so we add this test to make sure we don't corrupt
+    # data in the process
+    [source_db, target_db].each do |db|
+      db.query("CREATE DATABASE IF NOT EXISTS #{DEFAULT_DB}")
+      db.query("CREATE TABLE IF NOT EXISTS #{DEFAULT_FULL_TABLE_NAME} (id bigint(20) not null auto_increment, data decimal, primary key(id))")
+    end
+
+    source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (id, data) VALUES (1, 2)")
+
+    ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY)
+
+    row_copy_called = false
+    ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do
+      # this hook follows the design in the helper method
+      # execute_copy_data_in_fixed_size_binary_column below. See detailed
+      # comments there
+      res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}")
+      assert_equal 1, res.count
+      res.each do |row|
+        assert_equal 1, row["id"]
+        assert_equal 2, row["data"]
+      end
+
+      source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (id, data) VALUES (3, 4)")
+      source_db.query("UPDATE #{DEFAULT_FULL_TABLE_NAME} SET data = 5 WHERE id = 1")
+
+      row_copy_called = true
+    end
+
+    ghostferry.run
+
+    assert row_copy_called
+    assert_test_table_is_identical
+    res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}")
+    assert_equal 2, res.count
+    res.each do |row|
+      if row["id"] == 1
+        assert_equal 5, row["data"]
+      else
+        assert_equal 3, row["id"]
+        assert_equal 4, row["data"]
+      end
+    end
+  end
+
+  def test_copy_data_in_fixed_size_binary_column
+    # NOTE: We explicitly test with a value that is shorter than the max column
+    # size - MySQL will 0-pad the value up the full length of the BINARY column,
+    # but the MySQL replication binlogs will *not* contain these 0-bytes.
+    #
+    # As a result, the binlog writer must explicitly add then when building
+    # update/delete statements, as the WHERE clause would not match existing
+    # rows in the target DB
+    inserted_data = "ABC"
+    execute_copy_data_in_fixed_size_binary_column(
+        column_size: 4,
+        inserted_data: inserted_data,
+        expected_inserted_data: "#{inserted_data}\x00",
+        updated_data: "EFGH"
+    )
+  end
+
+  def test_copy_data_in_fixed_size_binary_column__value_completely_filled
+    # NOTE: This test is interesting (beyond what is covered above already),
+    # because it seems the server strips the trailing 0-bytes before sending
+    # them to the binlog.
+    inserted_data = "ABC\x00"
+    execute_copy_data_in_fixed_size_binary_column(
+        column_size: 4,
+        inserted_data: inserted_data,
+        expected_inserted_data: inserted_data,
+        updated_data: "EFGH"
+    )
+  end
+
+  def test_copy_data_in_fixed_size_binary_column__length1
+    # slight variation to cover the corner-case where there is no data in the
+    # column at all and the entire value is 0-padded (here, only 1 byte)
+    execute_copy_data_in_fixed_size_binary_column(
+        column_size: 1,
+        inserted_data: "",
+        expected_inserted_data: "\x00",
+        updated_data: "A"
+    )
+  end
+
   private
 
   def insert_json_on_source
@@ -280,4 +368,61 @@ def insert_json_on_source
     source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FALSE}')")
     source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_NUMBER}')")
   end
+
+  def execute_copy_data_in_fixed_size_binary_column(column_size:, inserted_data:, expected_inserted_data:, updated_data:)
+    # test for the BINARY columns needing 0-byte padding
+    #
+    # For details, see https://github.com/Shopify/ghostferry/pull/159
+
+    [source_db, target_db].each do |db|
+      db.query("CREATE DATABASE IF NOT EXISTS #{DEFAULT_DB}")
+      db.query("CREATE TABLE IF NOT EXISTS #{DEFAULT_FULL_TABLE_NAME} (id bigint(20) not null auto_increment, data BINARY(#{column_size}), primary key(id))")
+    end
+
+    source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (id, data) VALUES (1, _binary'#{inserted_data}')")
+
+    ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY)
+
+    row_copy_called = false
+    ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do
+      # select row from the target and then make sure the data with 0 padding
+      # is present. We do this to make sure there are no races in the test
+      res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}")
+      assert_equal 1, res.count
+      res.each do |row|
+        assert_equal 1, row["id"]
+        assert_equal expected_inserted_data, row["data"]
+      end
+
+      # now that the target is guaranteed to be in the same state as the
+      # source, trigger an update that will cause the binlog to stream an
+      # entry that needs the 0-byte padding
+      #
+      # NOTE: If we use BINLOG_STREAMING_STARTED as hook instead, we race
+      # with getting the update into the batch-copy instead of into the
+      # streaming
+      source_db.query("UPDATE #{DEFAULT_FULL_TABLE_NAME} SET data = _binary'#{updated_data}' WHERE id = 1")
+
+      # NOTE: We move this flag to the end of the callback to make sure that
+      # we don't confuse ourselves if the callback crashes before completing
+      row_copy_called = true
+    end
+
+    ghostferry.run
+
+    # make sure the test framework called the expected hooks above - otherwise
+    # the test doesn't make much sense
+    assert row_copy_called
+    assert_test_table_is_identical
+
+    # just being paranoid here: make sure the test outcome is as expected. It
+    # should be, since we made sure the tables have the same checksums, but it
+    # helps understand what the test code does
+    res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}")
+    assert_equal 1, res.count
+    res.each do |row|
+      assert_equal 1, row["id"]
+      assert_equal updated_data, row["data"]
+    end
+  end
 end