Add rows_read

This adds an additional info bit that is JSON serialized and contains
the number of rows read for a query so that the client can get access to
this data.

Signed-off-by: Dirkjan Bussink <[email protected]>

Author: piki

client/mysql.cc

@@ -3412,7 +3412,13 @@ static int com_go(String *buffer, char *line [[maybe_unused]]) {
     }
     my_stpcpy(pos, time_buff);
     put_info(buff, INFO_RESULT);
-    if (mysql_info(&mysql)) put_info(mysql_info(&mysql), INFO_RESULT);
+    const char *info = mysql_info(&mysql);
+    if (info) {
+      if (info[0])
+        put_info(mysql_info(&mysql), INFO_RESULT);
+      else if (info[1] && getenv("MYSQL_METRICS"))
+        printf("Metrics: %s\n", info+1);
+    }
     put_info("", INFO_RESULT);  // Empty row
 
     if (result && !mysql_eof(result)) /* Something wrong when using quick */

client/mysqltest.cc

@@ -8381,7 +8381,7 @@ static void append_info(DYNAMIC_STRING *ds, ulonglong affected_rows,
   char buf[40], buff2[21];
   sprintf(buf, "affected rows: %s\n", llstr(affected_rows, buff2));
   dynstr_append(ds, buf);
-  if (info) {
+  if (info && info[0]) {
     dynstr_append(ds, "info: ");
     dynstr_append(ds, info);
     dynstr_append_mem(ds, "\n", 1);

router/src/harness/tests/test_net_ts_internet.cc

@@ -661,6 +661,11 @@ TEST(NetTS_internet, tcp_ipv4_socket_bind_accept_connect_dynbuffer) {
 }
 
 TEST(NetTS_internet, tcp_ipv6_socket_bind_accept_connect) {
+  if (getenv("DISABLE_IPV6_TESTS")) {
+    // fails when ipv6 is present but disabled
+    GTEST_SKIP();
+  }
+
   net::io_context io_ctx;
 
   net::ip::tcp::endpoint endp = net_ipv6_any_port_endpoint();
@@ -748,6 +753,11 @@ TEST(NetTS_internet, tcp_ipv6_socket_bind_accept_connect) {
 }
 
 TEST(NetTS_internet, udp_ipv6_socket_bind_accept_connect) {
+  if (getenv("DISABLE_IPV6_TESTS")) {
+    // fails when ipv6 is present but disabled
+    GTEST_SKIP();
+  }
+
   net::io_context io_ctx;
 
   // any ip, any port

router/tests/component/test_http_server.cc

@@ -69,6 +69,9 @@ static_assert(sizeof(kInvalidBindAddress) > 7 + 1,
               "kInvalidBindAddress is too short");
 #endif
 
+const std::string localhost_ipv4("127.0.0.1");
+const std::string localhost_ipv6("::1");
+
 using namespace std::chrono_literals;
 
 const std::string kHttpBasedir(kPlaceholderHttpBaseDir);
@@ -211,6 +214,11 @@ TempDirectory HttpServerPlainTest::http_base_dir_;
  * - make a client connect to the http-server
  */
 TEST_P(HttpServerPlainTest, ensure) {
+  if (getenv("DISABLE_IPV6_TESTS") && GetParam().http_hostname == localhost_ipv6) {
+    // fails when ipv6 is present but disabled
+    GTEST_SKIP();
+  }
+
   std::vector<std::pair<std::string, std::string>> http_section;
   http_section.reserve(GetParam().http_section.size());
 
@@ -310,9 +318,6 @@ TEST_P(HttpServerPlainTest, ensure) {
   }
 }
 
-const std::string localhost_ipv4("127.0.0.1");
-const std::string localhost_ipv6("::1");
-
 static const HttpServerPlainParams http_server_static_files_params[]{
     {"bind-address-ipv4-any",
      "WL11891::TS-3",

sql/auth/sql_security_ctx.cc

@@ -93,6 +93,7 @@ void Security_context::init() {
   m_has_drop_policy = false;
   m_executed_drop_policy = false;
   m_registration_sandbox_mode = false;
+  m_exclude_user_from_rows_read = false;
 }
 
 void Security_context::logout() {
@@ -815,6 +816,8 @@ void Security_context::set_user_ptr(const char *user_arg,
 
   // set new user value to m_user.
   m_user.set(user_arg, user_arg_length, system_charset_info);
+
+  recheck_exclude_rows_read();
 }
 
 /**
@@ -837,6 +840,25 @@ void Security_context::assign_user(const char *user_arg,
     m_user.copy(user_arg, user_arg_length, system_charset_info);
   else
     m_user.set((const char *)nullptr, 0, system_charset_info);
+
+  recheck_exclude_rows_read();
+}
+
+char *rows_read_exclude_users = nullptr;
+
+void Security_context::recheck_exclude_rows_read() {
+  m_exclude_user_from_rows_read = false;
+  if (rows_read_exclude_users) {
+    char *copy = strdup(rows_read_exclude_users);
+    char *saveptr, *tok;
+    for (tok=my_strtok_r(copy, ",", &saveptr); tok; tok=my_strtok_r(NULL, ",", &saveptr)) {
+      if (!strcmp(tok, m_user.ptr())) {
+        m_exclude_user_from_rows_read = true;
+        break;
+      }
+    }
+    free(copy);
+  }
 }
 
 /**
@@ -880,7 +902,7 @@ void Security_context::set_host_ptr(const char *host_arg,
 /**
   Setter method for member m_host.
 
-  Copies host_arg value to the m_host if it is not null else m_user is set
+  Copies host_arg value to the m_host if it is not null else m_host is set
   to empty string.
 
 

sql/auth/sql_security_ctx.h

@@ -62,6 +62,7 @@ class Security_context {
   void skip_grants(const char *user = "skip-grants user",
                    const char *host = "skip-grants host");
   bool is_skip_grants_user();
+  bool exclude_user_from_rows_read() const { return m_exclude_user_from_rows_read; }
 
   /**
     Getter method for member m_user.
@@ -380,6 +381,9 @@ class Security_context {
   */
   bool m_is_skip_grants_user;
 
+  bool m_exclude_user_from_rows_read;
+  void recheck_exclude_rows_read();
+
   bool m_executed_drop_policy;
   bool m_has_drop_policy;
   std::unique_ptr<std::function<void(Security_context *)>> m_drop_policy;

sql/protocol_classic.cc

@@ -459,6 +459,7 @@
 #include "sql/sql_lex.h"
 #include "sql/sql_list.h"
 #include "sql/sql_prepare.h"  // Prepared_statement
+#include "sql/sql_thd_internal_api.h"
 #include "sql/system_variables.h"
 #include "sql_string.h"
 #include "template_utils.h"
@@ -858,7 +859,7 @@ bool net_send_error(NET *net, uint sql_errno, const char *err) {
 
 static bool net_send_ok(THD *thd, uint server_status, uint statement_warn_count,
                         ulonglong affected_rows, ulonglong id,
-                        const char *message, bool eof_identifier) {
+                        const char *message, size_t msglen, bool eof_identifier) {
   Protocol *protocol = thd->get_protocol();
   NET *net = thd->get_protocol_classic()->get_net();
   uchar buff[MYSQL_ERRMSG_SIZE + 10];
@@ -927,9 +928,9 @@ static bool net_send_ok(THD *thd, uint server_status, uint statement_warn_count,
 
   if (protocol->has_client_capability(CLIENT_SESSION_TRACK)) {
     /* the info field */
-    if (state_changed || (message && message[0]))
+    if (state_changed || (message && msglen))
       pos = net_store_data(pos, pointer_cast<const uchar *>(message),
-                           message ? strlen(message) : 0);
+                           message ? msglen : 0);
     /* session state change information */
     if (unlikely(state_changed)) {
       store.set_charset(thd->variables.collation_database);
@@ -946,10 +947,10 @@ static bool net_send_ok(THD *thd, uint server_status, uint statement_warn_count,
       start = (uchar *)store.ptr();
       pos = start + store.length();
     }
-  } else if (message && message[0]) {
+  } else if (message && msglen > 0 && (message[0] || message[1])) {
     /* the info field, if there is a message to store */
     pos = net_store_data(pos, pointer_cast<const uchar *>(message),
-                         strlen(message));
+                         msglen);
   }
 
   /* OK packet length will be restricted to 16777215 bytes */
@@ -1299,7 +1300,7 @@ bool Protocol_classic::send_ok(uint server_status, uint statement_warn_count,
   DBUG_TRACE;
   const bool retval =
       net_send_ok(m_thd, server_status, statement_warn_count, affected_rows,
-                  last_insert_id, message, false);
+                  last_insert_id, message, strlen(message), false);
   // Reclaim some memory
   convert.shrink(m_thd->variables.net_buffer_length);
   return retval;
@@ -1321,9 +1322,19 @@ bool Protocol_classic::send_eof(uint server_status, uint statement_warn_count) {
   */
   if (has_client_capability(CLIENT_DEPRECATE_EOF) &&
       (m_thd->get_command() != COM_BINLOG_DUMP &&
-       m_thd->get_command() != COM_BINLOG_DUMP_GTID))
-    retval = net_send_ok(m_thd, server_status, statement_warn_count, 0, 0,
-                         nullptr, true);
+       m_thd->get_command() != COM_BINLOG_DUMP_GTID)) {
+    ulonglong rows_read = m_thd->get_stmt_da()->rows_read();
+    if (rows_read) {
+      char message[64] = "";
+      snprintf(message+1, sizeof(message)-1, "{\"rows_read\":%llu}", rows_read);
+      retval = net_send_ok(m_thd, server_status, statement_warn_count, 0, 0,
+                           message, 1+strlen(message+1), true);
+    }
+    else {
+      retval = net_send_ok(m_thd, server_status, statement_warn_count, 0, 0,
+                           NULL, 0, true);
+    }
+  }
   else
     retval = net_send_eof(m_thd, server_status, statement_warn_count);
   // Reclaim some memory
@@ -3708,7 +3719,7 @@ bool Protocol_binary::send_parameters(List<Item_param> *parameters,
                        (m_thd->server_status | SERVER_PS_OUT_PARAMS |
                         SERVER_MORE_RESULTS_EXISTS),
                        m_thd->get_stmt_da()->current_statement_cond_count(), 0,
-                       0, nullptr, true);
+                       0, nullptr, 0, true);
   else
     /*
       In case of old clients send EOF packet.

sql/sql_error.cc

@@ -336,6 +336,7 @@ Diagnostics_area::Diagnostics_area(bool allow_unlimited_conditions)
       m_status(DA_EMPTY),
       m_mysql_errno(0),
       m_affected_rows(0),
+      m_rows_read(0),
       m_last_insert_id(0),
       m_last_statement_cond_count(0),
       m_current_statement_cond_count(0),
@@ -364,6 +365,7 @@ void Diagnostics_area::reset_diagnostics_area() {
   m_last_insert_id = 0;
   m_last_statement_cond_count = 0;
 #endif
+  m_rows_read = 0;
   set_is_sent(false);
   // Tiny reset in debug mode to see garbage right away.
   m_status = DA_EMPTY;

sql/sql_error.h

@@ -397,6 +397,11 @@ class Diagnostics_area {
     return m_affected_rows;
   }
 
+  ulonglong rows_read() const {
+    assert(m_status == DA_OK || m_status == DA_EOF);
+    return m_rows_read;
+  }
+
   ulonglong last_insert_id() const {
     assert(m_status == DA_OK);
     return m_last_insert_id;
@@ -456,6 +461,9 @@ class Diagnostics_area {
   /** Increment the current row counter to point at the next row. */
   void inc_current_row_for_condition() { m_current_row_for_condition++; }
 
+  /** Increment the rows-read counter. */
+  void inc_rows_read() { m_rows_read++; }
+
   /** Set the current row counter to point to the given row number. */
   void set_current_row_for_condition(ulong rowno) {
     m_current_row_for_condition = rowno;
@@ -659,6 +667,11 @@ class Diagnostics_area {
   */
   ulonglong m_affected_rows;
 
+  /**
+    The number of rows read by the last statement.
+  */
+  ulonglong m_rows_read;
+
   /**
     Similarly to the previous member, this is a replacement of
     thd->first_successful_insert_id_in_prev_stmt, which is used

sql/sys_vars.cc

@@ -1080,6 +1080,13 @@ static Sys_var_charptr Sys_my_bind_addr(
     READ_ONLY NON_PERSIST GLOBAL_VAR(my_bind_addr_str), CMD_LINE(REQUIRED_ARG),
     IN_FS_CHARSET, DEFAULT(MY_BIND_ALL_ADDRESSES));
 
+extern char *rows_read_exclude_users;
+static Sys_var_charptr Sys_my_rows_read_exclude_users(
+    "rows_read_exclude_users",
+    "Comma-separated list of users whose queries do not increment rows_read",
+    READ_ONLY NON_PERSIST GLOBAL_VAR(rows_read_exclude_users), CMD_LINE(REQUIRED_ARG),
+    IN_FS_CHARSET, DEFAULT(nullptr));
+
 static Sys_var_charptr Sys_admin_addr(
     "admin_address",
     "IP address to bind to for service connection. Address can be an IPv4"

storage/innobase/handler/ha_innodb.cc

@@ -10295,9 +10295,10 @@ int ha_innobase::index_read(
       if (m_prebuilt->table->is_system_table) {
         srv_stats.n_system_rows_read.add(
             thd_get_thread_id(m_prebuilt->trx->mysql_thd), 1);
-      } else {
+      } else if (!m_user_thd->security_context()->exclude_user_from_rows_read()) {
         srv_stats.n_rows_read.add(thd_get_thread_id(m_prebuilt->trx->mysql_thd),
                                   1);
+        m_user_thd->get_stmt_da()->inc_rows_read();
       }
       break;
 
@@ -10542,9 +10543,10 @@ int ha_innobase::general_fetch(
       if (m_prebuilt->table->is_system_table) {
         srv_stats.n_system_rows_read.add(
             thd_get_thread_id(m_prebuilt->trx->mysql_thd), 1);
-      } else {
+      } else if (!m_user_thd->security_context()->exclude_user_from_rows_read()) {
         srv_stats.n_rows_read.add(thd_get_thread_id(m_prebuilt->trx->mysql_thd),
                                   1);
+        m_user_thd->get_stmt_da()->inc_rows_read();
       }
       break;
     case DB_RECORD_NOT_FOUND:

unittest/gunit/mysys_pathfuncs-t.cc

@@ -142,6 +142,12 @@ TEST(Mysys, LoadPathOverflow) {
 
 #ifdef HAVE_O_TMPFILE
 TEST(Mysys, CreateTempFile) {
+  if (getenv("DISABLE_TMPFILE_TESTS")) {
+    // In Docker, /tmp is a plain subdirectory under /, which is an overlayfs
+    // O_TMPFILE does not work on overlayfs
+    GTEST_SKIP();
+  }
+
   char dst[FN_REFLEN];
   aset(dst, 0xaa);