PS-9777: binlog_utils_udf plugin should handle binlog.index entries the same as the server code does

https://perconadev.atlassian.net/browse/PS-9777

Problem:
Some tools like PXB, when moving back a backup, can produce a malformed
binlog.index file. This file contains an entry that is just a binlog
file name without relative/absolute path.
Server is able to handle sucha case in a proper way, however
binlog_utils_udf plugin complains about not being able to find a binlog
file.

Cause:
When the server processes queries like SHOW BINARY LOGS or SHOW BINLOG
EVENTS IN, it does two things:
1. expand provided binlog name to the full binlog path
2. expand entries in binlog.index file to their full path
Then it does comparison (to deduce if the provided binlog file name is
a valid one)

binlog_utils_udf does the expansion of the provided binlog file name,
but gets binlog.index entries as they are. This causes that in
problematic case the entry is not found in binlog.index file causing
the error returned by plugin.

Solution:
Do the full expansion of binlog.index entries in the same way as
the expansion of the provided file name is done.

Author: kamil-holubicki

mysql-test/r/percona_binlog_utils_udf.result

@@ -1655,7 +1655,5 @@ DELETE FROM loaded_ts;
 LOAD DATA LOCAL INFILE 'MYSQLTEST_VARDIR/filtered_binlog_file' INTO TABLE loaded_ts;
 include/assert.inc ['last record timestamp extracted via get_last_record_timestamp_by_binlog() for stage 16 should match the value extracted via mysqlbinlog']
 DROP FUNCTION get_last_record_timestamp_by_binlog;
-DROP TABLE loaded_ts;
 UNINSTALL PLUGIN binlog_utils_udf;
-DROP TABLE captured_gtid;
 DROP TABLE t1;

mysql-test/t/percona_binlog_utils_udf.test

@@ -338,11 +338,59 @@ while($stage <= $number_of_stages)
 }
 DROP FUNCTION get_last_record_timestamp_by_binlog;
 
-# temporary table cleanup
-DROP TABLE loaded_ts;
+
+#
+# PS-9777 - binlog_utils_udf plugin should handle binlog.index entries the same as the server code does
+# Check that binlog_utils_udf plugin handles binlog.index file entries
+# that are filenames (no relative or absolute path) correctly.
+#
+--source include/have_util_sed.inc
+
+# We only expect that the following won't fail at any stage. Actual results don't matter.
+--disable_query_log
+--disable_result_log
+eval CREATE FUNCTION get_gtid_set_by_binlog RETURNS STRING SONAME "$BINLOG_UTILS_UDF_LIB";
+eval CREATE FUNCTION get_binlog_by_gtid_set RETURNS STRING SONAME "$BINLOG_UTILS_UDF_LIB";
+eval CREATE FUNCTION get_binlog_by_gtid RETURNS STRING SONAME "$BINLOG_UTILS_UDF_LIB";
+eval CREATE FUNCTION get_last_gtid_from_binlog RETURNS STRING SONAME "$BINLOG_UTILS_UDF_LIB";
+eval CREATE FUNCTION get_first_record_timestamp_by_binlog RETURNS INTEGER SONAME "$BINLOG_UTILS_UDF_LIB";
+eval CREATE FUNCTION get_last_record_timestamp_by_binlog RETURNS INTEGER SONAME "$BINLOG_UTILS_UDF_LIB";
+
+# Modify the index file (remove leading ./ leaving only binlog file names)
+--let $index_file = `SELECT @@log_bin_index`
+--exec $SED -i 's|^\./binlog\.|binlog.|' $index_file
+
+# There will be no temp tables after restart
+--let $binlog_file_name = `SELECT binlog_name FROM captured_gtid LIMIT 1`
+--let $gtid_value = `SELECT gtid_value FROM captured_gtid LIMIT 1`
+
+# mysqld uses a cache over index file. Restart the server to fetch the new index file content.
+--source include/restart_mysqld.inc
+
+# restart re-enables query and resutl logs
+--disable_query_log
+--disable_result_log
+eval SELECT get_gtid_set_by_binlog('$binlog_file_name');
+--let $gtidset = `SELECT get_gtid_set_by_binlog('$binlog_file_name')`
+eval SELECT get_binlog_by_gtid_set('$gtidset');
+eval SELECT get_binlog_by_gtid('$gtid_value');
+eval SELECT get_last_gtid_from_binlog('$binlog_file_name');
+eval SELECT get_first_record_timestamp_by_binlog('$binlog_file_name');
+eval SELECT get_last_record_timestamp_by_binlog('$binlog_file_name');
+
+DROP FUNCTION get_gtid_set_by_binlog;
+DROP FUNCTION get_binlog_by_gtid_set;
+DROP FUNCTION get_binlog_by_gtid;
+DROP FUNCTION get_last_gtid_from_binlog;
+DROP FUNCTION get_first_record_timestamp_by_binlog;
+DROP FUNCTION get_last_record_timestamp_by_binlog;
+--enable_query_log
+--enable_result_log
+
+
+# temporary table cleanup not needed as we restarted the server
 
 --replace_result $BINLOG_UTILS_UDF_LIB BINLOG_UTILS_UDF_LIB
 UNINSTALL PLUGIN binlog_utils_udf;
 
-DROP TABLE captured_gtid;
 DROP TABLE t1;

plugin/binlog_utils_udf/binlog_utils_udf.cc

@@ -168,6 +168,25 @@ const char *get_short_binlog_name(const std::string &binlog_name) noexcept {
   return binlog_name.c_str() + dirname_length(binlog_name.c_str());
 }
 
+//
+// Retrieves the contents of the index file. Then normalizes its entries (each
+// entry is prefixed with relative or absolute path)
+//
+std::pair<int, std::list<std::string>> get_normalized_log_index() {
+  auto log_index = mysql_bin_log.get_log_index(true /* need_lock_index */);
+
+  auto &binlog_names = log_index.second;
+  auto bg = std::begin(binlog_names);
+  auto en = std::end(binlog_names);
+  std::for_each(bg, en, [](std::string &s) {
+    fn_reflen_buffer buf;
+    check_and_normalize_binlog_name(get_short_binlog_name(s), buf);
+    s.assign(buf);
+  });
+
+  return log_index;
+}
+
 log_event_ptr find_first_event(std::string_view binlog_name) {
   DBUG_TRACE;
 
@@ -378,7 +397,7 @@ mysqlpp::udf_result_t<STRING_RESULT> get_binlog_by_gtid_impl::calculate(
       throw std::runtime_error("Cannot parse 'gtid_executed'");
   }
 
-  auto log_index = mysql_bin_log.get_log_index(true /* need_lock_index */);
+  auto log_index = get_normalized_log_index();
   if (log_index.first != LOG_INFO_EOF)
     throw std::runtime_error("Cannot read binary log index'");
   if (log_index.second.empty())
@@ -480,7 +499,7 @@ mysqlpp::udf_result_t<STRING_RESULT> get_gtid_set_by_binlog_impl::calculate(
     const mysqlpp::udf_context &ctx) {
   DBUG_TRACE;
 
-  auto log_index = mysql_bin_log.get_log_index(true /* need_lock_index */);
+  auto log_index = get_normalized_log_index();
   if (log_index.first != LOG_INFO_EOF)
     throw std::runtime_error("Cannot read binary log index");
   if (log_index.second.empty())
@@ -584,7 +603,7 @@ mysqlpp::udf_result_t<STRING_RESULT> get_binlog_by_gtid_set_impl::calculate(
       throw std::runtime_error("Cannot parse 'gtid_executed'");
   }
 
-  auto log_index = mysql_bin_log.get_log_index(true /* need_lock_index */);
+  auto log_index = get_normalized_log_index();
   if (log_index.first != LOG_INFO_EOF)
     throw std::runtime_error("Cannot read binary log index'");
   if (log_index.second.empty())