Fix for bug #35732: read-only blocks SELECT statements in InnoDB

Problem: SELECTs prohibited for a transactional SE in autocommit mode
if read_only is set.

Fix: allow them.


mysql-test/r/read_only_innodb.result:
  Fix for bug #35732: read-only blocks SELECT statements in InnoDB
    - test result.
mysql-test/t/read_only_innodb.test:
  Fix for bug #35732: read-only blocks SELECT statements in InnoDB
    - test case.
sql/handler.cc:
  Fix for bug #35732: read-only blocks SELECT statements in InnoDB
    - in autocommit mode thd->transaction.all list is empty thus 
      is_real_trans set to TRUE for any SELECTs, so using it in the
      "read_only" check is insufficient.
      ha_check_and_coalesce_trx_read_only() changed to return number
      of engines with read-write changes. This value is used in the
      "read-only" check and checks for GLOBAL READ LOCK.
sql/lock.cc:
  Fix for bug #35732: read-only blocks SELECT statements in InnoDB
    - added assert(protect_against_global_read_lock) before decreasing,
      in order to catch (uint) 0 - 1 situation due to wrong 
      wait_if_global_read_lock()/start_waiting_global_read_lock() call
      sequence.

mysql-test/r/read_only_innodb.result

@@ -16,3 +16,33 @@ ERROR HY000: The MySQL server is running with the --read-only option so it canno
 set global read_only=0;
 drop table table_11733 ;
 drop user test@localhost;
+GRANT CREATE, SELECT, DROP ON *.* TO test@localhost;
+CREATE TABLE t1(a INT) ENGINE=INNODB;
+INSERT INTO t1 VALUES (0), (1);
+SET GLOBAL read_only=1;
+SELECT * FROM t1;
+a
+0
+1
+BEGIN;
+SELECT * FROM t1;
+a
+0
+1
+COMMIT;
+SET GLOBAL read_only=0;
+FLUSH TABLES WITH READ LOCK;
+SELECT * FROM t1;
+a
+0
+1
+BEGIN;
+SELECT * FROM t1;
+a
+0
+1
+COMMIT;
+UNLOCK TABLES;
+DROP TABLE t1;
+DROP USER test@localhost;
+echo End of 5.1 tests 

mysql-test/t/read_only_innodb.test

@@ -41,3 +41,45 @@ set global read_only=0;
 drop table table_11733 ;
 drop user test@localhost;
 
+disconnect con1;
+
+#
+# Bug #35732: read-only blocks SELECT statements in InnoDB
+#
+# Test 1: read only mode
+GRANT CREATE, SELECT, DROP ON *.* TO test@localhost;
+connect(con1, localhost, test, , test);
+
+connection default;
+CREATE TABLE t1(a INT) ENGINE=INNODB;
+INSERT INTO t1 VALUES (0), (1);
+SET GLOBAL read_only=1;
+
+connection con1;
+SELECT * FROM t1;
+BEGIN;
+SELECT * FROM t1;
+COMMIT;
+
+connection default;
+SET GLOBAL read_only=0;
+
+#
+# Test 2: global read lock
+#
+FLUSH TABLES WITH READ LOCK;
+
+connection con1;
+SELECT * FROM t1;
+BEGIN;
+SELECT * FROM t1;
+COMMIT;
+
+connection default;
+UNLOCK TABLES;
+DROP TABLE t1;
+DROP USER test@localhost;
+
+disconnect con1;
+
+--echo echo End of 5.1 tests 

sql/handler.cc

@@ -953,15 +953,20 @@ int ha_prepare(THD *thd)
   As a side effect, propagates the read-only/read-write flags
   of the statement transaction to its enclosing normal transaction.
   
-  @retval TRUE   we must run a two-phase commit. Returned
-                 if we have at least two engines with read-write changes.
-  @retval FALSE  Don't need two-phase commit. Even if we have two
-                 transactional engines, we can run two independent
-                 commits if changes in one of the engines are read-only.
+  If we have at least two engines with read-write changes we must
+  run a two-phase commit. Otherwise we can run several independent
+  commits as the only transactional engine has read-write changes
+  and others are read-only.
+
+  @retval   0   All engines are read-only.
+  @retval   1   We have the only engine with read-write changes.
+  @retval   >1  More than one engine have read-write changes.
+                Note: return value might NOT be the exact number of
+                engines with read-write changes.
 */
 
 static
-bool
+uint
 ha_check_and_coalesce_trx_read_only(THD *thd, Ha_trx_info *ha_list,
                                     bool all)
 {
@@ -998,7 +1003,7 @@ ha_check_and_coalesce_trx_read_only(THD *thd, Ha_trx_info *ha_list,
       break;
     }
   }
-  return rw_ha_count > 1;
+  return rw_ha_count;
 }
 
 
@@ -1061,19 +1066,30 @@ int ha_commit_trans(THD *thd, bool all)
 #ifdef USING_TRANSACTIONS
   if (ha_info)
   {
-    bool must_2pc;
+    uint rw_ha_count;
+    bool rw_trans;
+
+    DBUG_EXECUTE_IF("crash_commit_before", abort(););
+
+    /* Close all cursors that can not survive COMMIT */
+    if (is_real_trans)                          /* not a statement commit */
+      thd->stmt_map.close_transient_cursors();
+
+    rw_ha_count= ha_check_and_coalesce_trx_read_only(thd, ha_info, all);
+    /* rw_trans is TRUE when we in a transaction changing data */
+    rw_trans= is_real_trans && (rw_ha_count > 0);
 
-    if (is_real_trans && wait_if_global_read_lock(thd, 0, 0))
+    if (rw_trans &&
+        wait_if_global_read_lock(thd, 0, 0))
     {
       ha_rollback_trans(thd, all);
       DBUG_RETURN(1);
     }
 
-    if (   is_real_trans
-        && opt_readonly
-        && ! (thd->security_ctx->master_access & SUPER_ACL)
-        && ! thd->slave_thread
-       )
+    if (rw_trans &&
+        opt_readonly &&
+        !(thd->security_ctx->master_access & SUPER_ACL) &&
+        !thd->slave_thread)
     {
       my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
       ha_rollback_trans(thd, all);
@@ -1081,15 +1097,7 @@ int ha_commit_trans(THD *thd, bool all)
       goto end;
     }
 
-    DBUG_EXECUTE_IF("crash_commit_before", abort(););
-
-    /* Close all cursors that can not survive COMMIT */
-    if (is_real_trans)                          /* not a statement commit */
-      thd->stmt_map.close_transient_cursors();
-
-    must_2pc= ha_check_and_coalesce_trx_read_only(thd, ha_info, all);
-
-    if (!trans->no_2pc && must_2pc)
+    if (!trans->no_2pc && (rw_ha_count > 1))
     {
       for (; ha_info && !error; ha_info= ha_info->next())
       {
@@ -1129,7 +1137,7 @@ int ha_commit_trans(THD *thd, bool all)
       tc_log->unlog(cookie, xid);
     DBUG_EXECUTE_IF("crash_commit_after", abort(););
 end:
-    if (is_real_trans)
+    if (rw_trans)
       start_waiting_global_read_lock(thd);
   }
 #endif /* USING_TRANSACTIONS */

sql/lock.cc

@@ -1533,6 +1533,7 @@ void start_waiting_global_read_lock(THD *thd)
   if (unlikely(thd->global_read_lock))
     DBUG_VOID_RETURN;
   (void) pthread_mutex_lock(&LOCK_global_read_lock);
+  DBUG_ASSERT(protect_against_global_read_lock);
   tmp= (!--protect_against_global_read_lock &&
         (waiting_for_read_lock || global_read_lock_blocks_commit));
   (void) pthread_mutex_unlock(&LOCK_global_read_lock);