MDEV-8235: Expose check_slave_start_position as SQL function GTID_CHECK_POS()

mysql-test/main/gtid_check_pos.result

@@ -0,0 +1,51 @@
+#
+# MDEV-8235 Expose check_slave_start_position as SQL function
+#
+# 1. Set up some basic GTID history
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+# The current GTID position is reachable
+SELECT GTID_CHECK_POS('GTID_POS');
+GTID_CHECK_POS('GTID_POS')
+1
+# 2. Invalid input raises an error; NULL input returns NULL
+SELECT GTID_CHECK_POS('invalid-format');
+ERROR HY000: Could not parse GTID list
+SELECT GTID_CHECK_POS(NULL);
+GTID_CHECK_POS(NULL)
+NULL
+# 3. Empty string (empty GTID list) returns 1 (trivially reachable)
+SELECT GTID_CHECK_POS('');
+GTID_CHECK_POS('')
+1
+# 4. Rotate to a new binlog and purge the old one
+FLUSH LOGS;
+# Write a new GTID into the new (post-purge) active binlog
+INSERT INTO t1 VALUES (3);
+# The new position (in the current active log) is reachable
+SELECT GTID_CHECK_POS('NEW_GTID_POS');
+GTID_CHECK_POS('NEW_GTID_POS')
+1
+# The old position (now in a purged log) returns 0
+SELECT GTID_CHECK_POS('OLD_GTID_POS');
+GTID_CHECK_POS('OLD_GTID_POS')
+0
+# A completely fake domain/server/seqno (never existed) returns 0
+SELECT GTID_CHECK_POS('99-99-9999999');
+GTID_CHECK_POS('99-99-9999999')
+0
+# 5. A future sequence number in a known domain returns 0
+SELECT GTID_CHECK_POS('ABSENT_GTID');
+GTID_CHECK_POS('ABSENT_GTID')
+0
+# 6. Mixed list: reachable pos combined with an unknown domain returns 0
+SELECT GTID_CHECK_POS('NEW_GTID_POS,99-99-9999999');
+GTID_CHECK_POS('NEW_GTID_POS,99-99-9999999')
+0
+# 7. List with only the current reachable position returns 1
+SELECT GTID_CHECK_POS('NEW_GTID_POS');
+GTID_CHECK_POS('NEW_GTID_POS')
+1
+# Cleanup
+DROP TABLE t1;

mysql-test/main/gtid_check_pos.test

@@ -0,0 +1,64 @@
+--source include/have_log_bin.inc
+
+--echo #
+--echo # MDEV-8235 Expose check_slave_start_position as SQL function
+--echo #
+
+--echo # 1. Set up some basic GTID history
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+
+--echo # The current GTID position is reachable
+let $pos= `SELECT @@GLOBAL.gtid_binlog_pos`;
+--replace_result $pos GTID_POS
+eval SELECT GTID_CHECK_POS('$pos');
+
+--echo # 2. Invalid input raises an error; NULL input returns NULL
+--error ER_INCORRECT_GTID_STATE
+SELECT GTID_CHECK_POS('invalid-format');
+SELECT GTID_CHECK_POS(NULL);
+
+--echo # 3. Empty string (empty GTID list) returns 1 (trivially reachable)
+SELECT GTID_CHECK_POS('');
+
+--echo # 4. Rotate to a new binlog and purge the old one
+FLUSH LOGS;
+let $purge_binlogs_to= query_get_value(SHOW MASTER STATUS, File, 1);
+--disable_query_log
+--disable_result_log
+--source include/wait_for_purge.inc
+--enable_result_log
+--enable_query_log
+
+--echo # Write a new GTID into the new (post-purge) active binlog
+INSERT INTO t1 VALUES (3);
+
+--echo # The new position (in the current active log) is reachable
+let $new_pos= `SELECT @@GLOBAL.gtid_binlog_pos`;
+--replace_result $new_pos NEW_GTID_POS
+eval SELECT GTID_CHECK_POS('$new_pos');
+
+--echo # The old position (now in a purged log) returns 0
+--replace_result $pos OLD_GTID_POS
+eval SELECT GTID_CHECK_POS('$pos');
+
+--echo # A completely fake domain/server/seqno (never existed) returns 0
+SELECT GTID_CHECK_POS('99-99-9999999');
+
+--echo # 5. A future sequence number in a known domain returns 0
+let $server_id= `SELECT @@GLOBAL.server_id`;
+let $absent_gtid= 0-$server_id-99999999;
+--replace_result $absent_gtid ABSENT_GTID
+eval SELECT GTID_CHECK_POS('$absent_gtid');
+
+--echo # 6. Mixed list: reachable pos combined with an unknown domain returns 0
+--replace_result $new_pos NEW_GTID_POS
+eval SELECT GTID_CHECK_POS('$new_pos,99-99-9999999');
+
+--echo # 7. List with only the current reachable position returns 1
+--replace_result $new_pos NEW_GTID_POS
+eval SELECT GTID_CHECK_POS('$new_pos');
+
+--echo # Cleanup
+DROP TABLE t1;

sql/item_cmpfunc.h

@@ -2920,6 +2920,36 @@ class Item_func_null_predicate :public Item_bool_func
 };
 
 
+/**
+  @brief SQL function GTID_CHECK_POS()
+
+  This class implements the GTID_CHECK_POS() SQL function, which accepts
+  a string representation of one or more GTID lists and verifies if they
+  exist and are reachable (viable) within the current set of binary logs.
+*/
+class Item_func_gtid_check_pos :public Item_bool_func
+{
+  String tmp_value;
+public:
+  Item_func_gtid_check_pos(THD *thd, Item *a): Item_bool_func(thd, a) {}
+  LEX_CSTRING func_name_cstring() const override
+  {
+    static LEX_CSTRING name= {STRING_WITH_LEN("gtid_check_pos") };
+    return name;
+  }
+  bool val_bool() override;
+  bool fix_length_and_dec(THD *thd) override
+  {
+    set_maybe_null();
+    return FALSE;
+  }
+
+protected:
+  Item *shallow_copy(THD *thd) const override
+  { return get_item_copy<Item_func_gtid_check_pos>(thd, this); }
+};
+
+
 class Item_func_isnull :public Item_func_null_predicate
 {
 public:

sql/item_create.cc

@@ -528,6 +528,19 @@ class Create_func_connection_id : public Create_func_arg0
 };
 
 
+class Create_func_gtid_check_pos : public Create_func_arg1
+{
+public:
+  Item *create_1_arg(THD *thd, Item *arg1) override;
+
+  static Create_func_gtid_check_pos s_singleton;
+
+protected:
+  Create_func_gtid_check_pos() = default;
+  ~Create_func_gtid_check_pos() override = default;
+};
+
+
 class Create_func_database : public Create_func_arg0
 {
 public:
@@ -3576,6 +3589,15 @@ Create_func_connection_id::create_builder(THD *thd)
 }
 
 
+Create_func_gtid_check_pos Create_func_gtid_check_pos::s_singleton;
+
+Item*
+Create_func_gtid_check_pos::create_1_arg(THD *thd, Item *arg1)
+{
+  return new (thd->mem_root) Item_func_gtid_check_pos(thd, arg1);
+}
+
+
 Create_func_database Create_func_database::s_singleton;
 
 Item*
@@ -6355,6 +6377,7 @@ const Native_func_registry func_array[] =
   { { STRING_WITH_LEN("CONCAT_OPERATOR_ORACLE") }, BUILDER(Create_func_concat_operator_oracle)},
   { { STRING_WITH_LEN("CONCAT_WS") }, BUILDER(Create_func_concat_ws)},
   { { STRING_WITH_LEN("CONNECTION_ID") }, BUILDER(Create_func_connection_id)},
+  { { STRING_WITH_LEN("GTID_CHECK_POS") }, BUILDER(Create_func_gtid_check_pos)},
   { { STRING_WITH_LEN("CONV") }, BUILDER(Create_func_conv)},
   { { STRING_WITH_LEN("CONVERT_TZ") }, BUILDER(Create_func_convert_tz)},
   { { STRING_WITH_LEN("COS") }, BUILDER(Create_func_cos)},

sql/item_func.cc

@@ -51,6 +51,7 @@
 #include "debug_sync.h"
 #include "sql_base.h"
 #include "sql_cte.h"
+#include "sql_repl.h"
 #ifdef WITH_WSREP
 #include "mysql/service_wsrep.h"
 #endif /* WITH_WSREP */
@@ -824,6 +825,38 @@ String *Item_int_func::val_str(String *str)
 }
 
 
+/**
+  Evaluates the GTID_CHECK_POS() function.
+
+  @retval true   All requested GTIDs are reachable/viable.
+  @retval false  Any requested GTID has been purged or is absent.
+                 Also returns false if the input argument is NULL or if
+                 the check fails (with null_value set to true).
+*/
+bool Item_func_gtid_check_pos::val_bool()
+{
+  DBUG_ASSERT(fixed());
+  String *gtid_str= args[0]->val_str(&tmp_value);
+  if ((null_value= args[0]->null_value))
+    return 0;
+
+#ifdef HAVE_REPLICATION
+  bool is_reachable= false;
+  if (rpl_gtid_pos_check_reachable(gtid_str, &is_reachable))
+  {
+    null_value= 1;
+    return 0;
+  }
+  null_value= 0;
+  return is_reachable;
+#else
+  my_error(ER_NOT_SUPPORTED_YET, MYF(0), "GTID_CHECK_POS");
+  null_value= 1;
+  return 0;
+#endif
+}
+
+
 bool Item_func_connection_id::fix_length_and_dec(THD *thd)
 {
   if (Item_long_func::fix_length_and_dec(thd))

sql/item_func.h

@@ -1505,6 +1505,9 @@ class Item_func_connection_id :public Item_long_func
 };
 
 
+
+
+
 class Item_func_signed :public Item_int_func
 {
 public:

sql/sql_repl.cc

@@ -2122,6 +2122,7 @@ gtid_state_from_binlog_pos(const char *in_name, uint32 pos, String *out_str)
 }
 
 
+
 static bool
 is_until_reached(binlog_send_info *info, ulong *ev_offset,
                  Log_event_type event_type, const char **errmsg,
@@ -5838,4 +5839,116 @@ int compare_log_name(const char *log_1, const char *log_2) {
   return res;
 }
 
+/**
+  @brief Helper function to validate if all GTIDs in a given string are reachable/viable.
+
+  This checks the requested GTID state against the current binlog state. It evaluates:
+  1. If binary logging is enabled.
+  2. If the GTID string can be parsed correctly.
+  3. If each GTID is present in the master's binlog state and has not been written in the future.
+  4. If all requested GTIDs have not been purged from the binary log files.
+
+  @param gtid_str      The input string containing GTIDs.
+  @param is_reachable  [OUT] Set to true if all requested GTIDs exist and are not purged.
+
+  @sa gtid_find_binlog_pos()
+  @sa Item_func_gtid_check_pos::val_bool()
+
+  @retval true   An error occurred (binary logging disabled or malformed string).
+  @retval false  Success. The reachability state is stored in `is_reachable`.
+*/
+bool rpl_gtid_pos_check_reachable(String *gtid_str, bool *is_reachable)
+{
+
+  slave_connection_state state;
+  char buf[FN_REFLEN] = {0};
+  const char *errormsg;
+
+  if (!mysql_bin_log.is_open())
+  {
+    my_error(ER_NO_BINARY_LOGGING, MYF(0));
+    return true;
+  }
+
+  if (state.load(gtid_str->ptr(), gtid_str->length()))
+  {
+    /* 
+      We purposefully do not swallow the error here. If the user passes an 
+      ill-formed string, state.load throws ER_INCORRECT_GTID_STATE which 
+      will correctly abort the statement and notify the user.
+    */
+    return true;
+  }
+
+  /*
+    Verify that each requested GTID actually exists in the binlog state
+    and is not in the future.
+  */
+  for (uint32 i= 0; i < state.hash.records; ++i)
+  {
+    slave_connection_state::entry *entry=
+      (slave_connection_state::entry *)my_hash_element(&state.hash, i);
+    rpl_gtid *slave_gtid= &entry->gtid;
+    rpl_gtid master_gtid;
+    rpl_gtid master_replication_gtid;
+
+    bool in_binlog= mysql_bin_log.find_in_binlog_state(slave_gtid->domain_id,
+                                                       slave_gtid->server_id,
+                                                       &master_gtid);
+    if (in_binlog)
+    {
+      if (slave_gtid->seq_no > master_gtid.seq_no)
+      {
+        *is_reachable= false;
+        return false;
+      }
+    }
+    else
+    {
+      bool start_at_own_slave_pos=
+        rpl_global_gtid_slave_state->domain_to_gtid(slave_gtid->domain_id,
+                                                    &master_replication_gtid) &&
+        slave_gtid->server_id == master_replication_gtid.server_id &&
+        slave_gtid->seq_no == master_replication_gtid.seq_no;
+      if (!start_at_own_slave_pos)
+      {
+        *is_reachable= false;
+        return false;
+      }
+    }
+  }
+
+  /*
+    gtid_find_binlog_pos will natively iterate over the entire `state` hash,
+    evaluating every GTID provided in the comma-separated list.
+
+    If ANY of the requested GTIDs are found to be purged from the binary logs, 
+    it returns a non-null error message pinpointing the purged GTID.
+
+    If ALL GTIDs within the requested state natively resolve (meaning they exist
+    in our index/binlogs, or are safely in the future), it returns a null errormsg.
+
+    We disregard `found_in_index` and `out_start_seek` counts since we are only doing 
+    boolean viability checking of the provided GTIDs, not actually initializing a dump thread.
+  */
+  bool found_in_index= false;
+  uint32 out_start_seek= 0;
+  rpl_binlog_state until_binlog_state;
+  until_binlog_state.init();
+  slave_connection_state until_gtid_state;
+
+  errormsg= gtid_find_binlog_pos(&state, buf, &until_gtid_state, &until_binlog_state, &found_in_index, &out_start_seek);
+
+  until_binlog_state.free();
+
+  if (errormsg)
+  {
+    *is_reachable= false;
+    return false; // At least one requested GTID has been purged.
+  }
+
+  *is_reachable= true;
+  return false; // All requested GTIDs are currently viable/reachable.
+}
+
 #endif /* HAVE_REPLICATION */

sql/sql_repl.h

@@ -72,6 +72,8 @@ int rpl_append_gtid_state(String *dest, bool use_binlog);
 int rpl_load_gtid_state(slave_connection_state *state, bool use_binlog);
 bool rpl_gtid_pos_check(THD *thd, char *str, size_t len);
 bool rpl_gtid_pos_update(THD *thd, char *str, size_t len);
+bool rpl_gtid_pos_check_reachable(String *gtid_str, bool *is_reachable);
+
 #else
 
 struct LOAD_FILE_IO_CACHE : public IO_CACHE { };