Fixes for binary logging --read-only mode

- Any temporary tables created under read-only mode will never be logged
  to binary log.  Any usage of these tables to update normal tables, even
  after read-only has been disabled, will use row base logging (as the
  temporary table will not be on the slave).
- Analyze, check and repair table will not be logged in read-only mode.

Other things:
- Removed not used varaibles in
  MYSQL_BIN_LOG::flush_and_set_pending_rows_event.
- Set table_share->table_creation_was_logged for all normal tables.
- THD::binlog_query() now returns -1 if statement was not logged., This
  is used to update table_share->table_creation_was_logged.
- Don't log admin statements in opt_readonly is set.
- Table's that doesn't have table_creation_was_logged will set binlog format to row
  logging.
- Removed not needed/wrong setting of table->s->table_creation_was_logged
  in create_table_from_items()

18 files changed, 425 insertions, 62 deletions

diff --git a/mysql-test/suite/binlog/r/read_only.result b/mysql-test/suite/binlog/r/read_only.result
new file mode 100644
index 00000000000..aa1893b56a3
--- /dev/null
+++ b/mysql-test/suite/binlog/r/read_only.result
@@ -0,0 +1,109 @@
+#
+# MDEV-17863 DROP TEMPORARY TABLE creates a transaction in
+# binary log on read only server
+# MDEV-19074 Improved read_only mode for slaves with
+# gtid_strict_mode enabled
+#
+create user test@localhost;
+grant CREATE, DROP, INSERT, SELECT on *.* to test@localhost;
+create table t1 (a int) engine=myisam;
+insert into t1 values (1),(2);
+reset master;
+set global read_only=1;
+# Ensure that optimize and analyze doesn't log to binary log
+connect  con1,localhost,test,,test;
+insert into t1 values(3);
+ERROR HY000: The MariaDB server is running with the --read-only option so it cannot execute this statement
+analyze table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	OK
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+repair table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	Error	The MariaDB server is running with the --read-only option so it cannot execute this statement
+test.t1	repair	error	Corrupt
+optimize table t1;
+ERROR HY000: The MariaDB server is running with the --read-only option so it cannot execute this statement
+# Ensure that using temporary tables is not logged
+create temporary table tmp1 (a int) engine=myisam;
+insert into tmp1 values (1),(2);
+update tmp1 set a=10 where a=2;
+delete from tmp1 where a=1;
+create temporary table tmp2 select * from t1;
+select * from tmp1;
+a
+10
+select * from tmp2;
+a
+1
+2
+create temporary table tmp3 like t1;
+create or replace temporary table tmp3 like t1;
+alter table tmp2 add column (b int);
+select * from tmp2;
+a	b
+1	NULL
+2	NULL
+drop table tmp1,tmp2,tmp3;
+# Clean up test connection
+disconnect con1;
+connection default;
+# Execute some commands as root that should not be logged
+optimize table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	status	Table is already up to date
+repair table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	status	OK
+# Changes to temporary tables created under readonly should not
+# be logged
+create temporary table tmp4 (a int) engine=myisam;
+insert into tmp4 values (1),(2);
+create temporary table tmp5 (a int) engine=myisam;
+insert into tmp5 select * from tmp4;
+alter table tmp5 add column (b int);
+set global read_only=0;
+insert into tmp4 values (3),(4);
+insert into tmp5 values (10,3),(11,4);
+select * from tmp4;
+a
+1
+2
+3
+4
+select * from tmp5;
+a	b
+1	NULL
+2	NULL
+10	3
+11	4
+update tmp4 set a=10 where a=2;
+delete from tmp4 where a=1;
+create table t2 select * from tmp4;
+alter table tmp5 add column (c int);
+insert into tmp5 values (20,5,1),(21,5,2);
+select * from tmp5;
+a	b	c
+1	NULL	NULL
+2	NULL	NULL
+10	3	NULL
+11	4	NULL
+20	5	1
+21	5	2
+drop table tmp4,tmp5;
+# Check what is logged. Only the last create select should be row-logged
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+)
+master-bin.000001	#	Annotate_rows	#	#	create table t2 select * from tmp4
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+# Clean up
+drop user test@localhost;
+drop table t1,t2;
diff --git a/mysql-test/suite/binlog/r/read_only_statement.result b/mysql-test/suite/binlog/r/read_only_statement.result
new file mode 100644
index 00000000000..aa1893b56a3
--- /dev/null
+++ b/mysql-test/suite/binlog/r/read_only_statement.result
@@ -0,0 +1,109 @@
+#
+# MDEV-17863 DROP TEMPORARY TABLE creates a transaction in
+# binary log on read only server
+# MDEV-19074 Improved read_only mode for slaves with
+# gtid_strict_mode enabled
+#
+create user test@localhost;
+grant CREATE, DROP, INSERT, SELECT on *.* to test@localhost;
+create table t1 (a int) engine=myisam;
+insert into t1 values (1),(2);
+reset master;
+set global read_only=1;
+# Ensure that optimize and analyze doesn't log to binary log
+connect  con1,localhost,test,,test;
+insert into t1 values(3);
+ERROR HY000: The MariaDB server is running with the --read-only option so it cannot execute this statement
+analyze table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	OK
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+repair table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	Error	The MariaDB server is running with the --read-only option so it cannot execute this statement
+test.t1	repair	error	Corrupt
+optimize table t1;
+ERROR HY000: The MariaDB server is running with the --read-only option so it cannot execute this statement
+# Ensure that using temporary tables is not logged
+create temporary table tmp1 (a int) engine=myisam;
+insert into tmp1 values (1),(2);
+update tmp1 set a=10 where a=2;
+delete from tmp1 where a=1;
+create temporary table tmp2 select * from t1;
+select * from tmp1;
+a
+10
+select * from tmp2;
+a
+1
+2
+create temporary table tmp3 like t1;
+create or replace temporary table tmp3 like t1;
+alter table tmp2 add column (b int);
+select * from tmp2;
+a	b
+1	NULL
+2	NULL
+drop table tmp1,tmp2,tmp3;
+# Clean up test connection
+disconnect con1;
+connection default;
+# Execute some commands as root that should not be logged
+optimize table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	status	Table is already up to date
+repair table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	status	OK
+# Changes to temporary tables created under readonly should not
+# be logged
+create temporary table tmp4 (a int) engine=myisam;
+insert into tmp4 values (1),(2);
+create temporary table tmp5 (a int) engine=myisam;
+insert into tmp5 select * from tmp4;
+alter table tmp5 add column (b int);
+set global read_only=0;
+insert into tmp4 values (3),(4);
+insert into tmp5 values (10,3),(11,4);
+select * from tmp4;
+a
+1
+2
+3
+4
+select * from tmp5;
+a	b
+1	NULL
+2	NULL
+10	3
+11	4
+update tmp4 set a=10 where a=2;
+delete from tmp4 where a=1;
+create table t2 select * from tmp4;
+alter table tmp5 add column (c int);
+insert into tmp5 values (20,5,1),(21,5,2);
+select * from tmp5;
+a	b	c
+1	NULL	NULL
+2	NULL	NULL
+10	3	NULL
+11	4	NULL
+20	5	1
+21	5	2
+drop table tmp4,tmp5;
+# Check what is logged. Only the last create select should be row-logged
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+)
+master-bin.000001	#	Annotate_rows	#	#	create table t2 select * from tmp4
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+# Clean up
+drop user test@localhost;
+drop table t1,t2;
diff --git a/mysql-test/suite/binlog/t/read_only.inc b/mysql-test/suite/binlog/t/read_only.inc
new file mode 100644
index 00000000000..75cd4a5dfb3
--- /dev/null
+++ b/mysql-test/suite/binlog/t/read_only.inc
@@ -0,0 +1,74 @@
+--echo #
+--echo # MDEV-17863 DROP TEMPORARY TABLE creates a transaction in
+--echo # binary log on read only server
+--echo # MDEV-19074 Improved read_only mode for slaves with
+--echo # gtid_strict_mode enabled
+--echo #
+
+create user test@localhost;
+grant CREATE, DROP, INSERT, SELECT on *.* to test@localhost;
+create table t1 (a int) engine=myisam;
+insert into t1 values (1),(2);
+reset master;
+
+set global read_only=1;
+--echo # Ensure that optimize and analyze doesn't log to binary log
+connect (con1,localhost,test,,test);
+--error ER_OPTION_PREVENTS_STATEMENT
+insert into t1 values(3);
+analyze table t1;
+check table t1;
+repair table t1;
+--error ER_OPTION_PREVENTS_STATEMENT
+optimize table t1;
+
+--echo # Ensure that using temporary tables is not logged
+create temporary table tmp1 (a int) engine=myisam;
+insert into tmp1 values (1),(2);
+update tmp1 set a=10 where a=2;
+delete from tmp1 where a=1;
+create temporary table tmp2 select * from t1;
+select * from tmp1;
+select * from tmp2;
+create temporary table tmp3 like t1;
+create or replace temporary table tmp3 like t1;
+alter table tmp2 add column (b int);
+select * from tmp2;
+drop table tmp1,tmp2,tmp3;
+
+--echo # Clean up test connection
+disconnect con1;
+connection default;
+
+--echo # Execute some commands as root that should not be logged
+optimize table t1;
+repair table t1;
+
+--echo # Changes to temporary tables created under readonly should not
+--echo # be logged
+create temporary table tmp4 (a int) engine=myisam;
+insert into tmp4 values (1),(2);
+create temporary table tmp5 (a int) engine=myisam;
+insert into tmp5 select * from tmp4;
+alter table tmp5 add column (b int);
+
+set global read_only=0;
+
+insert into tmp4 values (3),(4);
+insert into tmp5 values (10,3),(11,4);
+select * from tmp4;
+select * from tmp5;
+update tmp4 set a=10 where a=2;
+delete from tmp4 where a=1;
+create table t2 select * from tmp4;
+alter table tmp5 add column (c int);
+insert into tmp5 values (20,5,1),(21,5,2);
+select * from tmp5;
+drop table tmp4,tmp5;
+
+--echo # Check what is logged. Only the last create select should be row-logged
+source include/show_binlog_events.inc;
+
+--echo # Clean up
+drop user test@localhost;
+drop table t1,t2;
diff --git a/mysql-test/suite/binlog/t/read_only.test b/mysql-test/suite/binlog/t/read_only.test
new file mode 100644
index 00000000000..14a4650b36f
--- /dev/null
+++ b/mysql-test/suite/binlog/t/read_only.test
@@ -0,0 +1,2 @@
+--source include/have_binlog_format_mixed_or_row.inc
+--source read_only.inc
diff --git a/mysql-test/suite/binlog/t/read_only_statement.test b/mysql-test/suite/binlog/t/read_only_statement.test
new file mode 100644
index 00000000000..a976854fbb2
--- /dev/null
+++ b/mysql-test/suite/binlog/t/read_only_statement.test
@@ -0,0 +1,2 @@
+--source include/have_binlog_format_statement.inc
+--source read_only.inc
diff --git a/sql/log.cc b/sql/log.cc
index 63c781b3be7..fe8ec38bacd 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5909,7 +5909,6 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
   DBUG_ASSERT(WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open());
   DBUG_PRINT("enter", ("event: %p", event));
 
-  int error= 0;
   binlog_cache_mngr *const cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
 
@@ -5947,7 +5946,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
 
   thd->binlog_set_pending_rows_event(event, is_transactional);
 
-  DBUG_RETURN(error);
+  DBUG_RETURN(0);
 }
 
 
diff --git a/sql/sp.cc b/sql/sp.cc
index 78efd6de95b..5661c6910cb 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -1466,7 +1466,7 @@ log:
     /* Such a statement can always go directly to binlog, no trans cache */
     if (thd->binlog_query(THD::STMT_QUERY_TYPE,
                           log_query.ptr(), log_query.length(),
-                          FALSE, FALSE, FALSE, 0))
+                          FALSE, FALSE, FALSE, 0) > 0)
     {
       my_error(ER_ERROR_ON_WRITE, MYF(MY_WME), "binary log", -1);
       goto done;
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index ad35f872a46..c92dc3f9f71 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3354,7 +3354,7 @@ bool change_password(THD *thd, LEX_USER *user)
     DBUG_ASSERT(query_length);
     thd->clear_error();
     result= thd->binlog_query(THD::STMT_QUERY_TYPE, buff, query_length,
-                              FALSE, FALSE, FALSE, 0);
+                              FALSE, FALSE, FALSE, 0) > 0;
   }
 end:
   close_mysql_tables(thd);
@@ -3516,7 +3516,7 @@ int acl_set_default_role(THD *thd, const char *host, const char *user,
       DBUG_ASSERT(query_length);
       thd->clear_error();
       result= thd->binlog_query(THD::STMT_QUERY_TYPE, buff, query_length,
-                                FALSE, FALSE, FALSE, 0);
+                                FALSE, FALSE, FALSE, 0) > 0;
     }
   end:
     close_mysql_tables(thd);
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 1c1ab82e570..297733a00a7 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -1314,7 +1314,7 @@ bool Sql_cmd_analyze_table::execute(THD *thd)
                          "analyze", lock_type, 1, 0, 0, 0,
                          &handler::ha_analyze, 0);
   /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog)
+  if (!res && !m_lex->no_write_to_binlog && !opt_readonly)
   {
     /*
       Presumably, ANALYZE and binlog writing doesn't require synchronization
@@ -1372,7 +1372,7 @@ bool Sql_cmd_optimize_table::execute(THD *thd)
                       "optimize", TL_WRITE, 1, 0, 0, 0,
                       &handler::ha_optimize, 0);
   /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog)
+  if (!res && !m_lex->no_write_to_binlog && !opt_readonly)
   {
     /*
       Presumably, OPTIMIZE and binlog writing doesn't require synchronization
@@ -1406,7 +1406,7 @@ bool Sql_cmd_repair_table::execute(THD *thd)
                          &handler::ha_repair, &view_repair);
 
   /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog)
+  if (!res && !m_lex->no_write_to_binlog && !opt_readonly)
   {
     /*
       Presumably, REPAIR and binlog writing doesn't require synchronization
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 3d42d6bedb0..0fc74eff19f 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -6026,7 +6026,16 @@ int THD::decide_logging_format(TABLE_LIST *tables)
       if (table->placeholder())
         continue;
 
-      handler::Table_flags const flags= table->table->file->ha_table_flags();
+      handler::Table_flags flags= table->table->file->ha_table_flags();
+      if (!table->table->s->table_creation_was_logged)
+      {
+        /*
+          This is a temporary table which was not logged in the binary log.
+          Disable statement logging to enforce row level logging.
+        */
+        DBUG_ASSERT(table->table->s->tmp_table);
+        flags&= ~HA_BINLOG_STMT_CAPABLE;
+      }
 
       DBUG_PRINT("info", ("table: %s; ha_table_flags: 0x%llx",
                           table->table_name.str, flags));
@@ -6248,9 +6257,16 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         {
           /*
             5. Error: Cannot modify table that uses a storage engine
-               limited to row-logging when binlog_format = STATEMENT
+               limited to row-logging when binlog_format = STATEMENT, except
+               if all tables that are updated are temporary tables
           */
-	  if (IF_WSREP((!WSREP(this) || wsrep_exec_mode == LOCAL_STATE),1))
+          if (!lex->stmt_writes_to_non_temp_table())
+          {
+            /* As all updated tables are temporary, nothing will be logged */
+            set_current_stmt_binlog_format_row();
+          }
+          else if (IF_WSREP((!WSREP(this) ||
+                             wsrep_exec_mode == LOCAL_STATE),1))
 	  {
             my_error((error= ER_BINLOG_STMT_MODE_AND_ROW_ENGINE), MYF(0), "");
 	  }
@@ -7177,11 +7193,12 @@ void THD::issue_unsafe_warnings()
 
   @see decide_logging_format
 
+  @retval < 0 No logging of query (ok)
   @retval 0 Success
-
-  @retval nonzero If there is a failure when writing the query (e.g.,
-  write failure), then the error code is returned.
+  @retval > 0  If there is a failure when writing the query (e.g.,
+               write failure), then the error code is returned.
 */
+
 int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                       ulong query_len, bool is_trans, bool direct, 
                       bool suppress_use, int errcode)
@@ -7207,7 +7224,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
       The current statement is to be ignored, and not written to
       the binlog. Do not call issue_unsafe_warnings().
     */
-    DBUG_RETURN(0);
+    DBUG_RETURN(-1);
   }
 
   /*
@@ -7223,7 +7240,10 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
   {
     int error;
     if (unlikely(error= binlog_flush_pending_rows_event(TRUE, is_trans)))
+    {
+      DBUG_ASSERT(error > 0);
       DBUG_RETURN(error);
+    }
   }
 
   /*
@@ -7266,7 +7286,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                ("is_current_stmt_binlog_format_row: %d",
                 is_current_stmt_binlog_format_row()));
     if (is_current_stmt_binlog_format_row())
-      DBUG_RETURN(0);
+      DBUG_RETURN(-1);
     /* Fall through */
 
     /*
@@ -7307,7 +7327,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
       }
 
       binlog_table_maps= 0;
-      DBUG_RETURN(error);
+      DBUG_RETURN(error >= 0 ? error : 1);
     }
 
   case THD::QUERY_TYPE_COUNT:
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index bec90dcf4e1..e3d17174b8b 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -851,7 +851,7 @@ cleanup:
                                         transactional_table, FALSE, FALSE,
                                         errcode);
 
-      if (log_result)
+      if (log_result > 0)
       {
 	error=1;
       }
@@ -1522,7 +1522,7 @@ bool multi_delete::send_eof()
       if (unlikely(thd->binlog_query(THD::ROW_QUERY_TYPE,
                                      thd->query(), thd->query_length(),
                                      transactional_tables, FALSE, FALSE,
-                                     errcode)) &&
+                                     errcode) > 0) &&
           !normal_tables)
       {
 	local_error=1;  // Log write failed: roll back the SQL statement
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 59cc88f9db3..d928dfbc3ae 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1186,13 +1186,13 @@ values_loop_end:
           else if (thd->binlog_query(THD::ROW_QUERY_TYPE,
                                      log_query.c_ptr(), log_query.length(),
                                      transactional_table, FALSE, FALSE,
-                                     errcode))
+                                     errcode) > 0)
             error= 1;
         }
         else if (thd->binlog_query(THD::ROW_QUERY_TYPE,
 			           thd->query(), thd->query_length(),
 			           transactional_table, FALSE, FALSE,
-                                   errcode))
+                                   errcode) > 0)
 	  error= 1;
       }
     }
@@ -3931,6 +3931,7 @@ bool select_insert::prepare_eof()
   int error;
   bool const trans_table= table->file->has_transactions();
   bool changed;
+  bool binary_logged= 0;
   killed_state killed_status= thd->killed;
 
   DBUG_ENTER("select_insert::prepare_eof");
@@ -3978,18 +3979,22 @@ bool select_insert::prepare_eof()
       (likely(!error) || thd->transaction.stmt.modified_non_trans_table))
   {
     int errcode= 0;
+    int res;
     if (likely(!error))
       thd->clear_error();
     else
       errcode= query_error_code(thd, killed_status == NOT_KILLED);
-    if (thd->binlog_query(THD::ROW_QUERY_TYPE,
-                      thd->query(), thd->query_length(),
-                      trans_table, FALSE, FALSE, errcode))
+    res= thd->binlog_query(THD::ROW_QUERY_TYPE,
+                           thd->query(), thd->query_length(),
+                           trans_table, FALSE, FALSE, errcode);
+    if (res > 0)
     {
       table->file->ha_release_auto_increment();
       DBUG_RETURN(true);
     }
+    binary_logged= res == 0 || !table->s->tmp_table;
   }
+  table->s->table_creation_was_logged|= binary_logged;
   table->file->ha_release_auto_increment();
 
   if (unlikely(error))
@@ -4040,8 +4045,9 @@ bool select_insert::send_eof()
   DBUG_RETURN(res);
 }
 
-void select_insert::abort_result_set() {
-
+void select_insert::abort_result_set()
+{
+  bool binary_logged= 0;
   DBUG_ENTER("select_insert::abort_result_set");
   /*
     If the creation of the table failed (due to a syntax error, for
@@ -4093,16 +4099,20 @@ void select_insert::abort_result_set() {
         if(WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open())
         {
           int errcode= query_error_code(thd, thd->killed == NOT_KILLED);
+          int res;
           /* error of writing binary log is ignored */
-          (void) thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
-                                   thd->query_length(),
-                                   transactional_table, FALSE, FALSE, errcode);
+          res= thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
+                                 thd->query_length(),
+                                 transactional_table, FALSE, FALSE, errcode);
+          binary_logged= res == 0 || !table->s->tmp_table;
         }
 	if (changed)
 	  query_cache_invalidate3(thd, table, 1);
     }
     DBUG_ASSERT(transactional_table || !changed ||
 		thd->transaction.stmt.modified_non_trans_table);
+
+    table->s->table_creation_was_logged|= binary_logged;
     table->file->ha_release_auto_increment();
   }
 
@@ -4173,6 +4183,7 @@ TABLE *select_create::create_table_from_items(THD *thd,
   /* Add selected items to field list */
   List_iterator_fast<Item> it(*items);
   Item *item;
+  bool save_table_creation_was_logged;
   DBUG_ENTER("select_create::create_table_from_items");
 
   tmp_table.s= &share;
@@ -4326,6 +4337,14 @@ TABLE *select_create::create_table_from_items(THD *thd,
 
   table->reginfo.lock_type=TL_WRITE;
   hooks->prelock(&table, 1);                    // Call prelock hooks
+
+  /*
+    Ensure that decide_logging_format(), called by mysql_lock_tables(), works
+    with temporary tables that will be logged later if needed.
+  */
+  save_table_creation_was_logged= table->s->table_creation_was_logged;
+  table->s->table_creation_was_logged= 1;
+
   /*
     mysql_lock_tables() below should never fail with request to reopen table
     since it won't wait for the table lock (we have exclusive metadata lock on
@@ -4349,6 +4368,7 @@ TABLE *select_create::create_table_from_items(THD *thd,
     DBUG_RETURN(0);
     /* purecov: end */
   }
+  table->s->table_creation_was_logged= save_table_creation_was_logged;
   DBUG_RETURN(table);
 }
 
@@ -4554,7 +4574,7 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
                               /* is_trans */ TRUE,
                               /* direct */ FALSE,
                               /* suppress_use */ FALSE,
-                              errcode);
+                              errcode) > 0;
   }
 
   ha_fake_trx_id(thd);
@@ -4667,8 +4687,6 @@ bool select_create::send_eof()
     }
 #endif /* WITH_WSREP */
   }
-  else if (!thd->is_current_stmt_binlog_format_row())
-    table->s->table_creation_was_logged= 1;
 
   /*
     exit_done must only be set after last potential call to
@@ -4754,7 +4772,8 @@ void select_create::abort_result_set()
   if (table)
   {
     bool tmp_table= table->s->tmp_table;
-
+    bool table_creation_was_logged= (!tmp_table ||
+                                     table->s->table_creation_was_logged);
     if (tmp_table)
     {
       DBUG_ASSERT(saved_tmp_table_share);
@@ -4783,7 +4802,9 @@ void select_create::abort_result_set()
       /* Remove logging of drop, create + insert rows */
       binlog_reset_cache(thd);
       /* Original table was deleted. We have to log it */
-      log_drop_table(thd, &create_table->db, &create_table->table_name, tmp_table);
+      if (table_creation_was_logged)
+        log_drop_table(thd, &create_table->db, &create_table->table_name,
+                       tmp_table);
     }
   }
   DBUG_VOID_RETURN;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 8b6b2ca9df0..03e73e23d3d 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2018,6 +2018,14 @@ public:
                 ((1U << STMT_READS_TEMP_TRANS_TABLE) |
                  (1U << STMT_WRITES_TEMP_TRANS_TABLE))) != 0);
   }
+  inline bool stmt_writes_to_non_temp_table()
+  {
+    DBUG_ENTER("THD::stmt_writes_to_non_temp_table");
+
+    DBUG_RETURN((stmt_accessed_table_flag &
+                ((1U << STMT_WRITES_TRANS_TABLE) |
+                 (1U << STMT_WRITES_NON_TRANS_TABLE))));
+  }
 
   /**
     Checks if a temporary non-transactional table is about to be accessed
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c2c26222a82..ebf268fe8c8 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1993,7 +1993,7 @@ int write_bin_log(THD *thd, bool clear_error,
       errcode= query_error_code(thd, TRUE);
     error= thd->binlog_query(THD::STMT_QUERY_TYPE,
                              query, query_length, is_trans, FALSE, FALSE,
-                             errcode);
+                             errcode) > 0;
     thd_proc_info(thd, 0);
   }
   return error;
@@ -2616,12 +2616,12 @@ err:
 #ifdef WITH_WSREP
           thd->wsrep_skip_wsrep_GTID = true;
 #endif /* WITH_WSREP */
-          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                     built_non_trans_tmp_query.ptr(),
-                                     built_non_trans_tmp_query.length(),
-                                     FALSE, FALSE,
-                                     is_drop_tmp_if_exists_added,
-                                     0);
+          error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                      built_non_trans_tmp_query.ptr(),
+                                      built_non_trans_tmp_query.length(),
+                                      FALSE, FALSE,
+                                      is_drop_tmp_if_exists_added,
+                                      0) > 0);
       }
       if (trans_tmp_table_deleted)
       {
@@ -2631,12 +2631,12 @@ err:
 #ifdef WITH_WSREP
           thd->wsrep_skip_wsrep_GTID = true;
 #endif /* WITH_WSREP */
-          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                     built_trans_tmp_query.ptr(),
-                                     built_trans_tmp_query.length(),
-                                     TRUE, FALSE,
-                                     is_drop_tmp_if_exists_added,
-                                     0);
+          error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                      built_trans_tmp_query.ptr(),
+                                      built_trans_tmp_query.length(),
+                                      TRUE, FALSE,
+                                      is_drop_tmp_if_exists_added,
+                                      0) > 0);
       }
       if (non_tmp_table_deleted)
       {
@@ -2648,11 +2648,11 @@ err:
 #ifdef WITH_WSREP
           thd->wsrep_skip_wsrep_GTID = false;
 #endif /* WITH_WSREP */
-          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                     built_query.ptr(),
-                                     built_query.length(),
-                                     TRUE, FALSE, FALSE,
-                                     error_code);
+          error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                      built_query.ptr(),
+                                      built_query.length(),
+                                      TRUE, FALSE, FALSE,
+                                      error_code) > 0);
       }
     }
   }
@@ -2738,7 +2738,7 @@ bool log_drop_table(THD *thd, const LEX_CSTRING *db_name,
                                "failed CREATE OR REPLACE */"));
   error= thd->binlog_query(THD::STMT_QUERY_TYPE,
                            query.ptr(), query.length(),
-                           FALSE, FALSE, temporary_table, 0);
+                           FALSE, FALSE, temporary_table, 0) > 0;
   DBUG_RETURN(error);
 }
 
@@ -5215,9 +5215,13 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   }
 
 err:
-  /* In RBR we don't need to log CREATE TEMPORARY TABLE */
-  if (!result && thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
+  /* In RBR or readonly server we don't need to log CREATE TEMPORARY TABLE */
+  if (!result && create_info->tmp_table() &&
+      (thd->is_current_stmt_binlog_format_row() || opt_readonly))
+  {
+    /* Note that table->s->table_creation_was_logged is not set! */
     DBUG_RETURN(result);
+  }
 
   if (create_info->tmp_table())
     thd->transaction.stmt.mark_created_temp_table();
@@ -5234,11 +5238,13 @@ err:
       */
       thd->locked_tables_list.unlock_locked_table(thd, mdl_ticket);
     }
-    else if (likely(!result) && create_info->tmp_table() && create_info->table)
+    else if (likely(!result) && create_info->table)
     {
       /*
-        Remember that tmp table creation was logged so that we know if
+        Remember that table creation was logged so that we know if
         we should log a delete of it.
+        If create_info->table was not set, it's a normal table and
+        table_creation_was_logged will be set when the share is created.
       */
       create_info->table->s->table_creation_was_logged= 1;
     }
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 18ab69d3557..32adc4420b3 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1156,7 +1156,7 @@ update_end:
 
       if (thd->binlog_query(THD::ROW_QUERY_TYPE,
                             thd->query(), thd->query_length(),
-                            transactional_table, FALSE, FALSE, errcode))
+                            transactional_table, FALSE, FALSE, errcode) > 0)
       {
         error=1;				// Rollback update
       }
@@ -2885,7 +2885,7 @@ bool multi_update::send_eof()
 
       if (thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
                             thd->query_length(), transactional_tables, FALSE,
-                            FALSE, errcode))
+                            FALSE, errcode) > 0)
 	local_error= 1;				// Rollback update
       thd->set_current_stmt_binlog_format(save_binlog_format);
     }
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 98c1e816f31..1ab72fad17b 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -698,7 +698,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
     thd->reset_unsafe_warnings();
     if (thd->binlog_query(THD::STMT_QUERY_TYPE,
                           buff.ptr(), buff.length(), FALSE, FALSE, FALSE,
-                          errcode))
+                          errcode) > 0)
       res= TRUE;
   }
 
diff --git a/sql/table.cc b/sql/table.cc
index 34523767eb4..0e713f60d21 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -675,7 +675,11 @@ enum open_frm_error open_table_def(THD *thd, TABLE_SHARE *share, uint flags)
   frmlen= read_length + sizeof(head);
 
   share->init_from_binary_frm_image(thd, false, buf, frmlen);
-  error_given= true; // init_from_binary_frm_image has already called my_error()
+  /*
+    Don't give any additional errors. If there would be a problem,
+    init_from_binary_frm_image would call my_error() itself.
+  */
+  error_given= true;
   my_free(buf);
 
   goto err_not_open;
@@ -684,6 +688,9 @@ err:
   mysql_file_close(file, MYF(MY_WME));
 
 err_not_open:
+  /* Mark that table was created earlier and thus should have been logged */
+  share->table_creation_was_logged= 1;
+
   if (unlikely(share->error && !error_given))
   {
     share->open_errno= my_errno;
@@ -2838,6 +2845,8 @@ ret:
              sql_copy);
     DBUG_RETURN(HA_ERR_GENERIC);
   }
+  /* Treat the table as normal table from binary logging point of view */
+  table_creation_was_logged= 1;
   DBUG_RETURN(0);
 }
 
diff --git a/sql/table.h b/sql/table.h
index 0a0f0aab77f..64a49fb2596 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -732,16 +732,20 @@ struct TABLE_SHARE
   bool null_field_first;
   bool system;                          /* Set if system table (one record) */
   bool not_usable_by_query_cache;
+  /*
+    This is used by log tables, for tables that have their own internal
+    binary logging or for tables that doesn't support statement or row logging
+   */
   bool no_replicate;
   bool crashed;
   bool is_view;
   bool can_cmp_whole_record;
+  /* This is set for temporary tables where CREATE was binary logged */
   bool table_creation_was_logged;
   bool non_determinstic_insert;
   bool vcols_need_refixing;
   bool has_update_default_function;
   bool can_do_row_logging;              /* 1 if table supports RBR */
-
   ulong table_map_id;                   /* for row-based replication */
 
   /*