Revert "BUG#34737535 Move GTID related SQL functions to separate header/source files."

This reverts commit 4e58a1733f965452f9bbe60df9a4110929ef69a4.

Change-Id: I15d97ad1cc137e039fc22abee0a5304c444f0279

Author: Slawomir Maludzinski

sql/CMakeLists.txt

@@ -397,7 +397,6 @@ SET(SQL_SHARED_SOURCES
   item_cmpfunc.cc
   item_create.cc
   item_func.cc
-  item_gtid_func.cc
   item_pfs_func.cc
   item_json_func.cc
   item_regexp_func.cc

sql/item_create.cc

@@ -56,13 +56,12 @@
 #include "mysql_time.h"
 #include "mysqld_error.h"
 #include "sql/item.h"
-#include "sql/item_cmpfunc.h"    // Item_func_any_value
-#include "sql/item_func.h"       // Item_func_udf_str
-#include "sql/item_geofunc.h"    // Item_func_st_area
-#include "sql/item_gtid_func.h"  // Item_wait_for_executed_gtid_set Item_master_gtid_set_wait Item_func_gtid_subset
-#include "sql/item_inetfunc.h"   // Item_func_inet_ntoa
-#include "sql/item_json_func.h"  // Item_func_json
-#include "sql/item_pfs_func.h"   // Item_pfs_func_thread_id
+#include "sql/item_cmpfunc.h"      // Item_func_any_value
+#include "sql/item_func.h"         // Item_func_udf_str
+#include "sql/item_geofunc.h"      // Item_func_st_area
+#include "sql/item_inetfunc.h"     // Item_func_inet_ntoa
+#include "sql/item_json_func.h"    // Item_func_json
+#include "sql/item_pfs_func.h"     // Item_pfs_func_thread_id
 #include "sql/item_regexp_func.h"  // Item_func_regexp_xxx
 #include "sql/item_strfunc.h"      // Item_func_aes_encrypt
 #include "sql/item_sum.h"          // Item_sum_udf_str

sql/item_func.cc

@@ -5072,6 +5072,259 @@ longlong Item_master_pos_wait::val_int() {
   return Item_source_pos_wait::val_int();
 }
 
+bool Item_wait_for_executed_gtid_set::itemize(Parse_context *pc, Item **res) {
+  if (skip_itemize(res)) return false;
+  if (super::itemize(pc, res)) return true;
+  /*
+    It is unsafe because the return value depends on timing. If the timeout
+    happens, the return value is different from the one in which the function
+    returns with success.
+  */
+  pc->thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
+  pc->thd->lex->safe_to_cache_query = false;
+  return false;
+}
+
+/**
+  Wait until the given gtid_set is found in the executed gtid_set independent
+  of the slave threads.
+*/
+longlong Item_wait_for_executed_gtid_set::val_int() {
+  DBUG_TRACE;
+  assert(fixed);
+  THD *thd = current_thd;
+
+  null_value = false;
+
+  String *gtid_text = args[0]->val_str(&value);
+  if (gtid_text == nullptr) {
+    /*
+      Usually, an argument that is NULL causes an SQL function to return NULL,
+      however since this is a function with side-effects, a NULL value is
+      treated as an error.
+    */
+    if (!thd->is_error()) {
+      my_error(ER_MALFORMED_GTID_SET_SPECIFICATION, MYF(0), "NULL");
+    }
+    return error_int();
+  }
+
+  // Waiting for a GTID in a slave thread could cause the slave to
+  // hang/deadlock.
+  // @todo: Return error instead of NULL
+  if (thd->slave_thread) {
+    return error_int();
+  }
+
+  Gtid_set wait_for_gtid_set(global_sid_map, nullptr);
+
+  global_sid_lock->rdlock();
+  if (global_gtid_mode.get() == Gtid_mode::OFF) {
+    global_sid_lock->unlock();
+    my_error(ER_GTID_MODE_OFF, MYF(0), "use WAIT_FOR_EXECUTED_GTID_SET");
+    return error_int();
+  }
+
+  if (wait_for_gtid_set.add_gtid_text(gtid_text->c_ptr_safe()) !=
+      RETURN_STATUS_OK) {
+    global_sid_lock->unlock();
+    // Error has already been generated.
+    return error_int();
+  }
+
+  // Cannot wait for a GTID that the thread owns since that would
+  // immediately deadlock.
+  if (thd->owned_gtid.sidno > 0 &&
+      wait_for_gtid_set.contains_gtid(thd->owned_gtid)) {
+    char buf[Gtid::MAX_TEXT_LENGTH + 1];
+    thd->owned_gtid.to_string(global_sid_map, buf);
+    global_sid_lock->unlock();
+    my_error(ER_CANT_WAIT_FOR_EXECUTED_GTID_SET_WHILE_OWNING_A_GTID, MYF(0),
+             buf);
+    return error_int();
+  }
+
+  gtid_state->begin_gtid_wait();
+
+  double timeout = (arg_count == 2) ? args[1]->val_real() : 0;
+  if (timeout < 0) {
+    if (thd->is_strict_mode()) {
+      my_error(ER_WRONG_ARGUMENTS, MYF(0), "WAIT_FOR_EXECUTED_GTID_SET.");
+    } else {
+      push_warning_printf(thd, Sql_condition::SL_WARNING, ER_WRONG_ARGUMENTS,
+                          ER_THD(thd, ER_WRONG_ARGUMENTS),
+                          "WAIT_FOR_EXECUTED_GTID_SET.");
+    }
+    gtid_state->end_gtid_wait();
+    global_sid_lock->unlock();
+    return error_int();
+  }
+
+  bool result = gtid_state->wait_for_gtid_set(thd, &wait_for_gtid_set, timeout);
+  global_sid_lock->unlock();
+  gtid_state->end_gtid_wait();
+
+  return result;
+}
+
+Item_master_gtid_set_wait::Item_master_gtid_set_wait(const POS &pos, Item *a)
+    : Item_int_func(pos, a) {
+  push_deprecated_warn(current_thd, "WAIT_UNTIL_SQL_THREAD_AFTER_GTIDS",
+                       "WAIT_FOR_EXECUTED_GTID_SET");
+}
+
+Item_master_gtid_set_wait::Item_master_gtid_set_wait(const POS &pos, Item *a,
+                                                     Item *b)
+    : Item_int_func(pos, a, b) {
+  push_deprecated_warn(current_thd, "WAIT_UNTIL_SQL_THREAD_AFTER_GTIDS",
+                       "WAIT_FOR_EXECUTED_GTID_SET");
+}
+
+Item_master_gtid_set_wait::Item_master_gtid_set_wait(const POS &pos, Item *a,
+                                                     Item *b, Item *c)
+    : Item_int_func(pos, a, b, c) {
+  push_deprecated_warn(current_thd, "WAIT_UNTIL_SQL_THREAD_AFTER_GTIDS",
+                       "WAIT_FOR_EXECUTED_GTID_SET");
+}
+
+bool Item_master_gtid_set_wait::itemize(Parse_context *pc, Item **res) {
+  if (skip_itemize(res)) return false;
+  if (super::itemize(pc, res)) return true;
+  pc->thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
+  pc->thd->lex->safe_to_cache_query = false;
+  return false;
+}
+
+longlong Item_master_gtid_set_wait::val_int() {
+  assert(fixed);
+  DBUG_TRACE;
+  int event_count = 0;
+
+  null_value = false;
+
+  String *gtid = args[0]->val_str(&value);
+  if (gtid == nullptr) {
+    return error_int();
+  }
+
+  THD *thd = current_thd;
+  Master_info *mi = nullptr;
+  double timeout = (arg_count >= 2) ? args[1]->val_real() : 0;
+  if (timeout < 0) {
+    if (thd->is_strict_mode()) {
+      my_error(ER_WRONG_ARGUMENTS, MYF(0),
+               "WAIT_UNTIL_SQL_THREAD_AFTER_GTIDS.");
+    } else {
+      push_warning_printf(thd, Sql_condition::SL_WARNING, ER_WRONG_ARGUMENTS,
+                          ER_THD(thd, ER_WRONG_ARGUMENTS),
+                          "WAIT_UNTIL_SQL_THREAD_AFTER_GTIDS.");
+    }
+    return error_int();
+  }
+
+  if (thd->slave_thread) {
+    return error_int();
+  }
+
+  channel_map.rdlock();
+
+  /* If replication channel is mentioned */
+  if (arg_count == 3) {
+    String *channel_str;
+    if (!(channel_str = args[2]->val_str(&value))) {
+      channel_map.unlock();
+      return error_int();
+    }
+    mi = channel_map.get_mi(channel_str->ptr());
+  } else {
+    if (channel_map.get_num_instances() > 1) {
+      channel_map.unlock();
+      mi = nullptr;
+      my_error(ER_REPLICA_MULTIPLE_CHANNELS_CMD, MYF(0));
+      return error_int();
+    } else
+      mi = channel_map.get_default_channel_mi();
+  }
+
+  if ((mi != nullptr) &&
+      mi->rli->m_assign_gtids_to_anonymous_transactions_info.get_type() >
+          Assign_gtids_to_anonymous_transactions_info::enum_type::AGAT_OFF) {
+    my_error(ER_CANT_SET_ANONYMOUS_TO_GTID_AND_WAIT_UNTIL_SQL_THD_AFTER_GTIDS,
+             MYF(0));
+    channel_map.unlock();
+    return error_int();
+  }
+  if (global_gtid_mode.get() == Gtid_mode::OFF) {
+    channel_map.unlock();
+    return error_int();
+  }
+  gtid_state->begin_gtid_wait();
+
+  if (mi != nullptr) mi->inc_reference();
+
+  channel_map.unlock();
+
+  bool null_result = false;
+
+  if (mi != nullptr && mi->rli != nullptr) {
+    event_count = mi->rli->wait_for_gtid_set(thd, gtid, timeout);
+    if (event_count == -2) {
+      null_result = true;
+    }
+  } else {
+    /*
+      Replication has not been set up, we should return NULL;
+     */
+    null_result = true;
+  }
+  if (mi != nullptr) mi->dec_reference();
+
+  gtid_state->end_gtid_wait();
+
+  return null_result ? error_int() : event_count;
+}
+
+/**
+  Return 1 if both arguments are Gtid_sets and the first is a subset
+  of the second.  Generate an error if any of the arguments is not a
+  Gtid_set.
+*/
+longlong Item_func_gtid_subset::val_int() {
+  DBUG_TRACE;
+
+  assert(fixed);
+
+  null_value = false;
+
+  // Evaluate strings without lock
+  String *string1 = args[0]->val_str(&buf1);
+  if (string1 == nullptr) {
+    return error_int();
+  }
+  String *string2 = args[1]->val_str(&buf2);
+  if (string2 == nullptr) {
+    return error_int();
+  }
+
+  const char *charp1 = string1->c_ptr_safe();
+  assert(charp1 != nullptr);
+  const char *charp2 = string2->c_ptr_safe();
+  assert(charp2 != nullptr);
+  int ret = 1;
+  enum_return_status status;
+
+  Sid_map sid_map(nullptr /*no rwlock*/);
+  // compute sets while holding locks
+  const Gtid_set sub_set(&sid_map, charp1, &status);
+  if (status == RETURN_STATUS_OK) {
+    const Gtid_set super_set(&sid_map, charp2, &status);
+    if (status == RETURN_STATUS_OK) {
+      ret = sub_set.is_subset(&super_set) ? 1 : 0;
+    }
+  }
+  return ret;
+}
+
 /**
   Enables a session to wait on a condition until a timeout or a network
   disconnect occurs.

sql/item_func.h

@@ -2406,6 +2406,75 @@ class Item_master_pos_wait : public Item_source_pos_wait {
   longlong val_int() override;
 };
 
+/**
+  This class is used for implementing the new wait_for_executed_gtid_set
+  function and the functions related to them. This new function is independent
+  of the slave threads.
+*/
+class Item_wait_for_executed_gtid_set final : public Item_int_func {
+  typedef Item_int_func super;
+
+  String value;
+
+ public:
+  Item_wait_for_executed_gtid_set(const POS &pos, Item *a)
+      : Item_int_func(pos, a) {}
+  Item_wait_for_executed_gtid_set(const POS &pos, Item *a, Item *b)
+      : Item_int_func(pos, a, b) {}
+
+  bool itemize(Parse_context *pc, Item **res) override;
+  longlong val_int() override;
+  const char *func_name() const override {
+    return "wait_for_executed_gtid_set";
+  }
+  bool resolve_type(THD *thd) override {
+    if (param_type_is_default(thd, 0, 1)) return true;
+    if (param_type_is_default(thd, 1, 2, MYSQL_TYPE_DOUBLE)) return true;
+    set_nullable(true);
+    return false;
+  }
+};
+
+class Item_master_gtid_set_wait final : public Item_int_func {
+  typedef Item_int_func super;
+
+  String value;
+
+ public:
+  Item_master_gtid_set_wait(const POS &pos, Item *a);
+  Item_master_gtid_set_wait(const POS &pos, Item *a, Item *b);
+  Item_master_gtid_set_wait(const POS &pos, Item *a, Item *b, Item *c);
+
+  bool itemize(Parse_context *pc, Item **res) override;
+  longlong val_int() override;
+  const char *func_name() const override {
+    return "wait_until_sql_thread_after_gtids";
+  }
+  bool resolve_type(THD *thd) override {
+    if (param_type_is_default(thd, 0, 1)) return true;
+    if (param_type_is_default(thd, 1, 2, MYSQL_TYPE_DOUBLE)) return true;
+    if (param_type_is_default(thd, 2, 3)) return true;
+    set_nullable(true);
+    return false;
+  }
+};
+
+class Item_func_gtid_subset final : public Item_int_func {
+  String buf1;
+  String buf2;
+
+ public:
+  Item_func_gtid_subset(const POS &pos, Item *a, Item *b)
+      : Item_int_func(pos, a, b) {}
+  longlong val_int() override;
+  const char *func_name() const override { return "gtid_subset"; }
+  bool resolve_type(THD *thd) override {
+    if (param_type_is_default(thd, 0, ~0U)) return true;
+    return false;
+  }
+  bool is_bool_func() const override { return true; }
+};
+
 /**
  Internal functions used by INFORMATION_SCHEMA implementation to check
  if user have access to given database/table/column.