Reject binlog_format!=ROW from non-SQL threads in MyRocks (mysql#109)

Summary:
MyRocks doesn't support next-key locking so binlog_format!=ROW
may cause data inconsistency. On 5.6 slaves, binlog_format!=ROW is fine
because no concurrent access to the same table is guaranteed.

This diff changes as below.
- Rejecting updates (locking reads, including SELECT FOR UPDATE) on master
- No behavior changes on slave
- No behavior changes if binlog was disabled on the session
- Added a new session variable rocksdb_unsafe_for_binlog to optionally
      allow writing with statement based binary logging

Test Plan: mtr, new test case rpl_statement

Reviewers: spetrunia, hermanlee4, jkedgar, santoshb

Reviewed By: santoshb

Subscribers: webscalesql-eng

Differential Revision: https://reviews.facebook.net/D51231
Differential Revision: https://reviews.facebook.net/D51489
Differential Revision: https://reviews.facebook.net/D51555
Differential Revision: https://reviews.facebook.net/D51597

Author: yoshinorim

storage/rocksdb/ha_rocksdb.cc

@@ -667,6 +667,11 @@ static MYSQL_THDVAR_BOOL(skip_fill_cache,
   "Skip filling block cache on read requests",
   NULL, NULL, FALSE);
 
+static MYSQL_THDVAR_BOOL(unsafe_for_binlog,
+  PLUGIN_VAR_RQCMDARG,
+  "Allowing statement based binary logging which may break consistency",
+  nullptr, nullptr, FALSE);
+
 static MYSQL_THDVAR_UINT(records_in_range,
   PLUGIN_VAR_RQCMDARG,
   "Used to override the result of records_in_range(). Set to a positive number to override",
@@ -861,6 +866,7 @@ static struct st_mysql_sys_var* rocksdb_system_variables[]= {
   MYSQL_SYSVAR(write_ignore_missing_column_families),
 
   MYSQL_SYSVAR(skip_fill_cache),
+  MYSQL_SYSVAR(unsafe_for_binlog),
 
   MYSQL_SYSVAR(records_in_range),
   MYSQL_SYSVAR(debug_optimizer_n_rows),
@@ -5622,6 +5628,21 @@ int ha_rocksdb::external_lock(THD *thd, int lock_type)
 {
   int res= 0;
   DBUG_ENTER("ha_rocksdb::external_lock");
+
+  int binlog_format= thd_binlog_format(thd);
+  bool unsafe_for_binlog= THDVAR(ha_thd(), unsafe_for_binlog);
+  if (lock_type == F_WRLCK && !thd->rli_slave &&
+      !unsafe_for_binlog &&
+      binlog_format != BINLOG_FORMAT_ROW &&
+      binlog_format != BINLOG_FORMAT_UNSPEC &&
+      thd_binlog_filter_ok(thd))
+  {
+    my_printf_error(ER_UNKNOWN_ERROR,
+             "Can't execute updates on master with binlog_format != ROW.",
+             MYF(0));
+    DBUG_RETURN(HA_ERR_INTERNAL_ERROR);
+  }
+
   io_perf_start();
 
   if (lock_type == F_UNLCK)

storage/rocksdb/ha_rocksdb.h

@@ -67,19 +67,32 @@ inline bool looks_like_per_index_cf_typo(const char *name)
 extern PSI_stage_info stage_waiting_on_row_lock;
 
 extern "C"
+{
 void thd_enter_cond(MYSQL_THD thd, mysql_cond_t *cond, mysql_mutex_t *mutex,
                     const PSI_stage_info *stage, PSI_stage_info *old_stage);
-extern "C"
 void thd_exit_cond(MYSQL_THD thd, const PSI_stage_info *stage);
 
 /**
   Mark transaction to rollback and mark error as fatal to a sub-statement.
   @param  thd   Thread handle
   @param  all   TRUE <=> rollback main transaction.
 */
-extern "C"
 void thd_mark_transaction_to_rollback(MYSQL_THD thd, bool all);
 
+/**
+ *   Get the user thread's binary logging format
+ *   @param thd  user thread
+ *   @return Value to be used as index into the binlog_format_names array
+*/
+int thd_binlog_format(const MYSQL_THD thd);
+
+/**
+ *   Check if binary logging is filtered for thread's current db.
+ *   @param  thd   Thread handle
+ *   @retval 1 the query is not filtered, 0 otherwise.
+*/
+bool thd_binlog_filter_ok(const MYSQL_THD thd);
+}
 
 //#endif