/* Copyright (C) 2009-2020 Kentoku Shiba
   Copyright (C) 2019-2020 MariaDB corp

  This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; version 2 of the License.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */

#define MYSQL_SERVER 1
#include <my_global.h>
#include "mysql_version.h"
#include "sql_priv.h"
#include "create_options.h"
#include "probes_mysql.h"
#include "sql_class.h"
#include "sql_partition.h"
#include "sql_base.h"
#include "sql_servers.h"
#include "tztime.h"
#include "spd_err.h"
#include "spd_param.h"
#include "spd_db_include.h"
#include "spd_include.h"
#include "spd_sys_table.h"
#include "ha_spider.h"
#include "spd_db_conn.h"
#include "spd_trx.h"
#include "spd_conn.h"
#include "spd_table.h"
#include "spd_direct_sql.h"
#include "spd_udf.h"
#include "spd_malloc.h"

#define SPIDER_NEED_INIT_ONE_TABLE_FOR_FIND_TEMPORARY_TABLE

extern const char **spd_defaults_extra_file;
extern const char **spd_defaults_file;

extern handlerton *spider_hton_ptr;
extern SPIDER_DBTON spider_dbton[SPIDER_DBTON_SIZE];

#ifdef HAVE_PSI_INTERFACE
extern PSI_mutex_key spd_key_mutex_mta_conn;
extern PSI_mutex_key spd_key_mutex_bg_direct_sql;
extern PSI_cond_key spd_key_cond_bg_direct_sql;
#endif

extern HASH spider_open_connections;
extern HASH spider_ipport_conns;
extern pthread_mutex_t spider_conn_mutex;
extern pthread_mutex_t spider_conn_id_mutex;
extern pthread_mutex_t spider_ipport_conn_mutex;
extern ulonglong spider_conn_id;

/* UTC time zone for timestamp columns */
extern Time_zone *UTC;

uint spider_udf_calc_hash(
  char *key,
  uint mod
) {
  uint sum = 0;
  DBUG_ENTER("spider_udf_calc_hash");
  while (*key != '\0')
  {
    sum += *key;
    key++;
  }
  DBUG_PRINT("info",("spider calc hash = %u", sum % mod));
  DBUG_RETURN(sum % mod);
}

int spider_udf_direct_sql_create_table_list(
  SPIDER_DIRECT_SQL *direct_sql,
  char *table_name_list,
  uint table_name_list_length
) {
  int table_count, roop_count, length;
  char *tmp_ptr, *tmp_ptr2, *tmp_ptr3, *tmp_name_ptr;
  THD *thd = direct_sql->trx->thd;
  DBUG_ENTER("spider_udf_direct_sql_create_table_list");
  tmp_ptr = table_name_list;
  while (*tmp_ptr == ' ')
    tmp_ptr++;
  if (*tmp_ptr)
    table_count = 1;
  else {
    direct_sql->table_count = 0;
    DBUG_RETURN(0);
  }

  while (TRUE)
  {
    if ((tmp_ptr2 = strchr(tmp_ptr, ' ')))
    {
      table_count++;
      tmp_ptr = tmp_ptr2 + 1;
      while (*tmp_ptr == ' ')
        tmp_ptr++;
    } else
      break;
  }
  if (!(direct_sql->db_names = (char**)
    spider_bulk_malloc(spider_current_trx, SPD_MID_UDF_DIRECT_SQL_CREATE_TABLE_LIST_2, MYF(MY_WME | MY_ZEROFILL),
      &direct_sql->db_names, (uint) (sizeof(char*) * table_count),
      &direct_sql->table_names, (uint) (sizeof(char*) * table_count),
      &direct_sql->tables, (uint) (sizeof(TABLE*) * table_count),
      &tmp_name_ptr, (uint) (sizeof(char) * (
        table_name_list_length +
        SPIDER_THD_db_length(thd) * table_count +
        2 * table_count
      )),
      &direct_sql->iop, (uint) (sizeof(int) * table_count),
      &direct_sql->table_list, (uint) (sizeof(TABLE_LIST) * table_count),
      &direct_sql->real_table_bitmap,
        (uint) (sizeof(uchar) * ((table_count + 7) / 8)),
      NullS))
  )
    DBUG_RETURN(HA_ERR_OUT_OF_MEM);

  tmp_ptr = table_name_list;
  while (*tmp_ptr == ' ')
    tmp_ptr++;
  roop_count = 0;
  while (TRUE)
  {
    if ((tmp_ptr2 = strchr(tmp_ptr, ' ')))
      *tmp_ptr2 = '\0';

    direct_sql->db_names[roop_count] = tmp_name_ptr;

    if ((tmp_ptr3 = strchr(tmp_ptr, '.')))
    {
      /* exist database name */
      *tmp_ptr3 = '\0';
      length = strlen(tmp_ptr);
      memcpy(tmp_name_ptr, tmp_ptr, length + 1);
      tmp_name_ptr += length + 1;
      tmp_ptr = tmp_ptr3 + 1;
    } else {
      if (SPIDER_THD_db_str(thd))
      {
        memcpy(tmp_name_ptr, SPIDER_THD_db_str(thd),
          SPIDER_THD_db_length(thd) + 1);
        tmp_name_ptr += SPIDER_THD_db_length(thd) + 1;
      } else {
        direct_sql->db_names[roop_count] = (char *) "";
      }
    }

    direct_sql->table_names[roop_count] = tmp_name_ptr;
    length = strlen(tmp_ptr);
    memcpy(tmp_name_ptr, tmp_ptr, length + 1);
    tmp_name_ptr += length + 1;

    DBUG_PRINT("info",("spider db=%s",
      direct_sql->db_names[roop_count]));
    DBUG_PRINT("info",("spider table_name=%s",
      direct_sql->table_names[roop_count]));

    if (!tmp_ptr2)
      break;
    tmp_ptr = tmp_ptr2 + 1;
    while (*tmp_ptr == ' ')
      tmp_ptr++;
    roop_count++;
  }
  direct_sql->table_count = table_count;
  DBUG_RETURN(0);
}

int spider_udf_direct_sql_create_conn_key(
  SPIDER_DIRECT_SQL *direct_sql
) {
  char *tmp_name, port_str[6];
  DBUG_ENTER("spider_udf_direct_sql_create_conn_key");

  uint roop_count2;
  bool tables_on_different_db_are_joinable = TRUE;
  direct_sql->dbton_id = SPIDER_DBTON_SIZE;
  DBUG_PRINT("info",("spider direct_sql->tgt_wrapper=%s",
    direct_sql->tgt_wrapper));
  for (roop_count2 = 0; roop_count2 < SPIDER_DBTON_SIZE; roop_count2++)
  {
    DBUG_PRINT("info",("spider spider_dbton[%d].wrapper=%s", roop_count2,
      spider_dbton[roop_count2].wrapper ?
        spider_dbton[roop_count2].wrapper : "NULL"));
    if (
      spider_dbton[roop_count2].wrapper &&
      !strcasecmp(direct_sql->tgt_wrapper, spider_dbton[roop_count2].wrapper)
    ) {
        if (spider_dbton[roop_count2].db_access_type ==
          SPIDER_DB_ACCESS_TYPE_SQL)
        {
          direct_sql->dbton_id = roop_count2;
          break;
        }
    }
  }
  if (direct_sql->dbton_id == SPIDER_DBTON_SIZE)
  {
      my_printf_error(
        ER_SPIDER_SQL_WRAPPER_IS_INVALID_NUM,
        ER_SPIDER_SQL_WRAPPER_IS_INVALID_STR,
        MYF(0), direct_sql->tgt_wrapper);
      DBUG_RETURN(ER_SPIDER_SQL_WRAPPER_IS_INVALID_NUM);
  }
    tables_on_different_db_are_joinable =
      spider_dbton[direct_sql->dbton_id].db_util->
        tables_on_different_db_are_joinable();
    direct_sql->conn_key_length
      = 1
      + (direct_sql->tgt_wrapper ? direct_sql->tgt_wrapper_length + 2 : 0)
      + (direct_sql->tgt_host ? direct_sql->tgt_host_length + 2 : 0)
      + 5 + 2
      + (direct_sql->tgt_socket ? direct_sql->tgt_socket_length + 2 : 0)
      + (!tables_on_different_db_are_joinable && direct_sql->tgt_default_db_name ?
         direct_sql->tgt_default_db_name_length + 2 : 0)
      + (direct_sql->tgt_username ? direct_sql->tgt_username_length + 2 : 0)
      + (direct_sql->tgt_password ? direct_sql->tgt_password_length + 2 : 0)
      + (direct_sql->tgt_ssl_ca ? direct_sql->tgt_ssl_ca_length + 2 : 0)
      + (direct_sql->tgt_ssl_capath ? direct_sql->tgt_ssl_capath_length + 2 : 0)
      + (direct_sql->tgt_ssl_cert ? direct_sql->tgt_ssl_cert_length + 2 : 0)
      + (direct_sql->tgt_ssl_cipher ? direct_sql->tgt_ssl_cipher_length + 2 : 0)
      + (direct_sql->tgt_ssl_key ? direct_sql->tgt_ssl_key_length + 2 : 0)
      + 1 + 2
      + (direct_sql->tgt_default_file ? direct_sql->tgt_default_file_length + 2 : 0)
      + (direct_sql->tgt_default_group ? direct_sql->tgt_default_group_length + 2 : 0)
      + (direct_sql->tgt_dsn ? direct_sql->tgt_dsn_length + 2 : 0)
      + (direct_sql->tgt_filedsn ? direct_sql->tgt_filedsn_length + 2 : 0)
      + (direct_sql->tgt_driver ? direct_sql->tgt_driver_length + 2 : 0)
      + (direct_sql->tgt_odbc_conn_str ? direct_sql->tgt_odbc_conn_str_length + 2 : 0);
  if (!(direct_sql->conn_key = (char *)
    spider_malloc(spider_current_trx, SPD_MID_UDF_DIRECT_SQL_CREATE_CONN_KEY_1, direct_sql->conn_key_length + 1,
      MYF(MY_WME | MY_ZEROFILL)))
  )
    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
  if (direct_sql->connection_channel > 48)
    *direct_sql->conn_key = '0' + 48 - direct_sql->connection_channel;
  else
    *direct_sql->conn_key = '0' + direct_sql->connection_channel;
  int counter= 0;
  tmp_name= direct_sql->conn_key + 1;
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_wrapper);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_host);
  snprintf(port_str, sizeof(port_str), "%05ld", direct_sql->tgt_port);
  spider_create_conn_key_add_one(&counter, &tmp_name, port_str);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_socket);
  counter++;
  if (!tables_on_different_db_are_joinable && direct_sql->tgt_default_db_name)
  {
    *tmp_name= (char) counter;
    tmp_name = strmov(tmp_name + 1, direct_sql->tgt_default_db_name);
    tmp_name++;
  }
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_username);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_password);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_ssl_ca);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_ssl_capath);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_ssl_cert);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_ssl_cipher);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_ssl_key);
  counter++;
  *tmp_name= (char) counter;
  tmp_name++;
  *tmp_name = '0' + ((char) direct_sql->tgt_ssl_vsc);
  tmp_name++;
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_default_file);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_default_group);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_dsn);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_filedsn);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_driver);
  spider_create_conn_key_add_one(&counter, &tmp_name, direct_sql->tgt_odbc_conn_str);
  tmp_name++;
  direct_sql->conn_key_hash_value = my_calc_hash(&spider_open_connections,
    (uchar*) direct_sql->conn_key, direct_sql->conn_key_length);
  DBUG_RETURN(0);
}

static inline void spider_maybe_memcpy_string(
  char **dest,
  char *src,
  char *tmp,
  uint *dest_len,
  uint src_len)
{
    *dest_len= src_len;
    if (src)
    {
      *dest= tmp;
      memcpy(*dest, src, src_len);
    } else
      *dest= NULL;
}


SPIDER_CONN *spider_udf_direct_sql_create_conn(
  const SPIDER_DIRECT_SQL *direct_sql,
  int *error_num
) {
  SPIDER_CONN *conn;
  SPIDER_IP_PORT_CONN *ip_port_conn;
  char *tmp_name, *tmp_host, *tmp_username, *tmp_password, *tmp_socket;
  char *tmp_wrapper, *tmp_db, *tmp_ssl_ca, *tmp_ssl_capath, *tmp_ssl_cert;
  char *tmp_ssl_cipher, *tmp_ssl_key, *tmp_default_file, *tmp_default_group;
  char *tmp_dsn, *tmp_filedsn, *tmp_driver, *tmp_odbc_conn_str;
  int *need_mon;
  bool tables_on_different_db_are_joinable = TRUE;
  DBUG_ENTER("spider_udf_direct_sql_create_conn");

  if (unlikely(!UTC))
  {
    /* UTC time zone for timestamp columns */
    String tz_00_name(STRING_WITH_LEN("+00:00"), &my_charset_bin);
    UTC = my_tz_find(current_thd, &tz_00_name);
  }

    tables_on_different_db_are_joinable =
      spider_dbton[direct_sql->dbton_id].db_util->
        tables_on_different_db_are_joinable();
    if (!(conn = (SPIDER_CONN *)
      spider_bulk_malloc(spider_current_trx, SPD_MID_UDF_DIRECT_SQL_CREATE_CONN_1, MYF(MY_WME | MY_ZEROFILL),
        &conn, (uint) (sizeof(*conn)),
        &tmp_name, (uint) (direct_sql->conn_key_length + 1),
        &tmp_host, (uint) (direct_sql->tgt_host_length + 1),
        &tmp_username, (uint) (direct_sql->tgt_username_length + 1),
        &tmp_password, (uint) (direct_sql->tgt_password_length + 1),
        &tmp_socket, (uint) (direct_sql->tgt_socket_length + 1),
        &tmp_wrapper, (uint) (direct_sql->tgt_wrapper_length + 1),
        &tmp_db, (uint) (tables_on_different_db_are_joinable ?
          0 : direct_sql->tgt_default_db_name_length + 1),
        &tmp_ssl_ca, (uint) (direct_sql->tgt_ssl_ca_length + 1),
        &tmp_ssl_capath, (uint) (direct_sql->tgt_ssl_capath_length + 1),
        &tmp_ssl_cert, (uint) (direct_sql->tgt_ssl_cert_length + 1),
        &tmp_ssl_cipher, (uint) (direct_sql->tgt_ssl_cipher_length + 1),
        &tmp_ssl_key, (uint) (direct_sql->tgt_ssl_key_length + 1),
        &tmp_default_file,
          (uint) (direct_sql->tgt_default_file_length + 1),
        &tmp_default_group,
          (uint) (direct_sql->tgt_default_group_length + 1),
        &tmp_dsn,
          (uint) (direct_sql->tgt_dsn_length + 1),
        &tmp_filedsn,
          (uint) (direct_sql->tgt_filedsn_length + 1),
        &tmp_driver,
          (uint) (direct_sql->tgt_driver_length + 1),
        &tmp_odbc_conn_str,
          (uint) (direct_sql->tgt_odbc_conn_str_length + 1),
        &need_mon, (uint) (sizeof(int)),
        NullS))
    ) {
      *error_num = HA_ERR_OUT_OF_MEM;
      goto error_alloc_conn;
    }
    conn->default_database.init_calc_mem(SPD_MID_UDF_DIRECT_SQL_CREATE_CONN_2);

  conn->conn_key_length = direct_sql->conn_key_length;
  conn->conn_key = tmp_name;
  memcpy(conn->conn_key, direct_sql->conn_key, direct_sql->conn_key_length);
    spider_maybe_memcpy_string(
      &conn->tgt_wrapper, direct_sql->tgt_wrapper, tmp_wrapper,
      &conn->tgt_wrapper_length, direct_sql->tgt_wrapper_length);
    spider_maybe_memcpy_string(
      &conn->tgt_host, direct_sql->tgt_host, tmp_host,
      &conn->tgt_host_length, direct_sql->tgt_host_length);
    conn->tgt_port = direct_sql->tgt_port;
    spider_maybe_memcpy_string(
      &conn->tgt_socket, direct_sql->tgt_socket, tmp_socket,
      &conn->tgt_socket_length, direct_sql->tgt_socket_length);
    if (!tables_on_different_db_are_joinable)
      spider_maybe_memcpy_string(
        &conn->tgt_db, direct_sql->tgt_default_db_name, tmp_db,
        &conn->tgt_db_length, direct_sql->tgt_default_db_name_length);
    spider_maybe_memcpy_string(
      &conn->tgt_username, direct_sql->tgt_username, tmp_username,
      &conn->tgt_username_length, direct_sql->tgt_username_length);
    spider_maybe_memcpy_string(
      &conn->tgt_password, direct_sql->tgt_password, tmp_password,
      &conn->tgt_password_length, direct_sql->tgt_password_length);
     spider_maybe_memcpy_string(
      &conn->tgt_ssl_ca, direct_sql->tgt_ssl_ca, tmp_ssl_ca,
      &conn->tgt_ssl_ca_length, direct_sql->tgt_ssl_ca_length);
    spider_maybe_memcpy_string(
      &conn->tgt_ssl_capath, direct_sql->tgt_ssl_capath, tmp_ssl_capath,
      &conn->tgt_ssl_capath_length, direct_sql->tgt_ssl_capath_length);
    spider_maybe_memcpy_string(
      &conn->tgt_ssl_cert, direct_sql->tgt_ssl_cert, tmp_ssl_cert,
      &conn->tgt_ssl_cert_length, direct_sql->tgt_ssl_cert_length);
    spider_maybe_memcpy_string(
      &conn->tgt_ssl_cipher, direct_sql->tgt_ssl_cipher, tmp_ssl_cipher,
      &conn->tgt_ssl_cipher_length, direct_sql->tgt_ssl_cipher_length);
    spider_maybe_memcpy_string(
      &conn->tgt_ssl_key, direct_sql->tgt_ssl_key, tmp_ssl_key,
      &conn->tgt_ssl_key_length, direct_sql->tgt_ssl_key_length);
    spider_maybe_memcpy_string(
      &conn->tgt_default_file, direct_sql->tgt_default_file, tmp_default_file,
      &conn->tgt_default_file_length, direct_sql->tgt_default_file_length);
    spider_maybe_memcpy_string(
      &conn->tgt_default_group, direct_sql->tgt_default_group, tmp_default_group,
      &conn->tgt_default_group_length, direct_sql->tgt_default_group_length);
    spider_maybe_memcpy_string(
      &conn->tgt_dsn, direct_sql->tgt_dsn, tmp_dsn,
      &conn->tgt_dsn_length, direct_sql->tgt_dsn_length);
    spider_maybe_memcpy_string(
      &conn->tgt_filedsn, direct_sql->tgt_filedsn, tmp_filedsn,
      &conn->tgt_filedsn_length, direct_sql->tgt_filedsn_length);
    spider_maybe_memcpy_string(
      &conn->tgt_driver, direct_sql->tgt_driver, tmp_driver,
      &conn->tgt_driver_length, direct_sql->tgt_driver_length);
    spider_maybe_memcpy_string(
      &conn->tgt_odbc_conn_str, direct_sql->tgt_odbc_conn_str, tmp_odbc_conn_str,
      &conn->tgt_odbc_conn_str_length, direct_sql->tgt_odbc_conn_str_length);
    conn->tgt_ssl_vsc = direct_sql->tgt_ssl_vsc;
  conn->dbton_id = direct_sql->dbton_id;
  conn->conn_need_mon = need_mon;
  conn->need_mon = need_mon;
  if (!(conn->db_conn = spider_dbton[conn->dbton_id].create_db_conn(conn)))
  {
    *error_num = HA_ERR_OUT_OF_MEM;
    goto error_db_conn_create;
  }
  if ((*error_num = conn->db_conn->init()))
  {
    goto error_db_conn_init;
  }
  conn->join_trx = 0;
  conn->thd = NULL;
  conn->table_lock = 0;
  conn->semi_trx_isolation = -2;
  conn->semi_trx_isolation_chk = FALSE;
  conn->semi_trx_chk = FALSE;

  if (mysql_mutex_init(spd_key_mutex_mta_conn, &conn->mta_conn_mutex,
    MY_MUTEX_INIT_FAST))
  {
    *error_num = HA_ERR_OUT_OF_MEM;
    goto error_mta_conn_mutex_init;
  }

  if (unlikely((*error_num = spider_conn_init(conn))))
  {
    goto error_conn_init;
  }

  if ((*error_num = spider_db_udf_direct_sql_connect(direct_sql, conn)))
    goto error;
  conn->ping_time = (time_t) time((time_t*) 0);
  conn->connect_error_time = conn->ping_time;
  pthread_mutex_lock(&spider_conn_id_mutex);
  conn->conn_id = spider_conn_id;
  ++spider_conn_id;
  pthread_mutex_unlock(&spider_conn_id_mutex);

  pthread_mutex_lock(&spider_ipport_conn_mutex);
  if ((ip_port_conn = (SPIDER_IP_PORT_CONN*) my_hash_search_using_hash_value(
    &spider_ipport_conns, conn->conn_key_hash_value,
    (uchar*)conn->conn_key, conn->conn_key_length)))
  { /* exists, +1 */
    pthread_mutex_unlock(&spider_ipport_conn_mutex);
    pthread_mutex_lock(&ip_port_conn->mutex);
    if (spider_param_max_connections())
    { /* enable conncetion pool */
      if (ip_port_conn->ip_port_count >= spider_param_max_connections())
      { /* bigger than the max num of connections, free conn and return NULL */
        pthread_mutex_unlock(&ip_port_conn->mutex);
        goto error_too_many_ipport_count;
      }
    }
    ip_port_conn->ip_port_count++;
    pthread_mutex_unlock(&ip_port_conn->mutex);
  }
  else
  {// do not exist
    ip_port_conn = spider_create_ipport_conn(conn);
    if (!ip_port_conn) {
      /* failed, always do not effect 'create conn' */
      pthread_mutex_unlock(&spider_ipport_conn_mutex);
      DBUG_RETURN(conn);
    }
    if (my_hash_insert(&spider_ipport_conns, (uchar *)ip_port_conn)) {
      /* insert failed, always do not effect 'create conn' */
      pthread_mutex_unlock(&spider_ipport_conn_mutex);
      DBUG_RETURN(conn);
    }
    pthread_mutex_unlock(&spider_ipport_conn_mutex);
  }
  conn->ip_port_conn = ip_port_conn;

  DBUG_RETURN(conn);

error:
error_too_many_ipport_count:
  spider_conn_done(conn);
error_conn_init:
  pthread_mutex_destroy(&conn->mta_conn_mutex);
error_mta_conn_mutex_init:
error_db_conn_init:
  delete conn->db_conn;
error_db_conn_create:
  spider_free(spider_current_trx, conn, MYF(0));
error_alloc_conn:
  DBUG_RETURN(NULL);
}

SPIDER_CONN *spider_udf_direct_sql_get_conn(
  const SPIDER_DIRECT_SQL *direct_sql,
  SPIDER_TRX *trx,
  int *error_num
) {
  SPIDER_CONN *conn = NULL;
  DBUG_ENTER("spider_udf_direct_sql_get_conn");

  if (
      !(conn = (SPIDER_CONN*) my_hash_search_using_hash_value(
        &trx->trx_conn_hash, direct_sql->conn_key_hash_value,
        (uchar*) direct_sql->conn_key, direct_sql->conn_key_length))
  )
  {
    if (
        (
          (spider_param_conn_recycle_mode(trx->thd) & 1) ||
          spider_param_conn_recycle_strict(trx->thd)
        )
    ) {
        pthread_mutex_lock(&spider_conn_mutex);
        if (!(conn = (SPIDER_CONN*) my_hash_search_using_hash_value(
          &spider_open_connections, direct_sql->conn_key_hash_value,
          (uchar*) direct_sql->conn_key, direct_sql->conn_key_length)))
        {
          pthread_mutex_unlock(&spider_conn_mutex);
          DBUG_PRINT("info",("spider create new conn"));
          if(!(conn = spider_udf_direct_sql_create_conn(direct_sql,
            error_num)))
            goto error;
        } else {
          my_hash_delete(&spider_open_connections, (uchar*) conn);
          pthread_mutex_unlock(&spider_conn_mutex);
          DBUG_PRINT("info",("spider get global conn"));
        }
    } else {
      DBUG_PRINT("info",("spider create new conn"));
      /* conn_recycle_strict = 0 and conn_recycle_mode = 0 or 2 */
      if(!(conn = spider_udf_direct_sql_create_conn(direct_sql, error_num)))
        goto error;
    }
    conn->thd = trx->thd;
    conn->priority = direct_sql->priority;

      uint old_elements = trx->trx_conn_hash.array.max_element;
      if (my_hash_insert(&trx->trx_conn_hash, (uchar*) conn))
      {
        spider_free_conn(conn);
        *error_num = HA_ERR_OUT_OF_MEM;
        goto error;
      }
      if (trx->trx_conn_hash.array.max_element > old_elements)
      {
        spider_alloc_calc_mem(spider_current_trx,
          trx->trx_conn_hash,
          (trx->trx_conn_hash.array.max_element - old_elements) *
          trx->trx_conn_hash.array.size_of_element);
      }
  }

  if (conn->queued_connect)
  {
    if ((*error_num = spider_db_udf_direct_sql_connect(direct_sql, conn)))
      goto error;
    conn->queued_connect = FALSE;
  }

  if (conn->queued_ping)
    conn->queued_ping = FALSE;

  DBUG_PRINT("info",("spider conn=%p", conn));
  DBUG_RETURN(conn);

error:
  DBUG_RETURN(NULL);
}

int spider_udf_direct_sql_get_server(
  SPIDER_DIRECT_SQL *direct_sql
) {
  MEM_ROOT mem_root;
  int error_num, length;
  FOREIGN_SERVER *server, server_buf;
  DBUG_ENTER("spider_udf_direct_sql_get_server");
  SPD_INIT_ALLOC_ROOT(&mem_root, 65, 0, MYF(MY_WME));

  if (!(server
       = get_server_by_name(&mem_root, direct_sql->server_name, &server_buf)))
  {
    error_num = ER_FOREIGN_SERVER_DOESNT_EXIST;
    goto error_get_server;
  }

  if (!direct_sql->tgt_wrapper && server->scheme)
  {
    direct_sql->tgt_wrapper_length = strlen(server->scheme);
    if (!(direct_sql->tgt_wrapper =
      spider_create_string(server->scheme, direct_sql->tgt_wrapper_length)))
    {
      error_num = HA_ERR_OUT_OF_MEM;
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    DBUG_PRINT("info",("spider tgt_wrapper=%s", direct_sql->tgt_wrapper));
  }

  if (!direct_sql->tgt_host && server->host)
  {
    direct_sql->tgt_host_length = strlen(server->host);
    if (!(direct_sql->tgt_host =
      spider_create_string(server->host, direct_sql->tgt_host_length)))
    {
      error_num = HA_ERR_OUT_OF_MEM;
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    DBUG_PRINT("info",("spider tgt_host=%s", direct_sql->tgt_host));
  }

  if (direct_sql->tgt_port == -1)
  {
    direct_sql->tgt_port = server->port;
    DBUG_PRINT("info",("spider tgt_port=%ld", direct_sql->tgt_port));
  }

  if (!direct_sql->tgt_socket && server->socket)
  {
    direct_sql->tgt_socket_length = strlen(server->socket);
    if (!(direct_sql->tgt_socket =
      spider_create_string(server->socket, direct_sql->tgt_socket_length)))
    {
      error_num = HA_ERR_OUT_OF_MEM;
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    DBUG_PRINT("info",("spider tgt_socket=%s", direct_sql->tgt_socket));
  }

  if (!direct_sql->tgt_default_db_name && server->db &&
    (length = strlen(server->db)))
  {
    direct_sql->tgt_default_db_name_length = length;
    if (!(direct_sql->tgt_default_db_name =
      spider_create_string(server->db, length)))
    {
      error_num = HA_ERR_OUT_OF_MEM;
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    DBUG_PRINT("info",("spider tgt_default_db_name=%s",
      direct_sql->tgt_default_db_name));
  }

  if (!direct_sql->tgt_username && server->username)
  {
    direct_sql->tgt_username_length = strlen(server->username);
    if (!(direct_sql->tgt_username =
      spider_create_string(server->username, direct_sql->tgt_username_length)))
    {
      error_num = HA_ERR_OUT_OF_MEM;
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    DBUG_PRINT("info",("spider tgt_username=%s", direct_sql->tgt_username));
  }

  if (!direct_sql->tgt_password && server->password)
  {
    direct_sql->tgt_password_length = strlen(server->password);
    if (!(direct_sql->tgt_password =
      spider_create_string(server->password, direct_sql->tgt_password_length)))
    {
      error_num = HA_ERR_OUT_OF_MEM;
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    DBUG_PRINT("info",("spider tgt_password=%s", direct_sql->tgt_password));
  }

  free_root(&mem_root, MYF(0));
  DBUG_RETURN(0);

error_get_server:
  my_error(error_num, MYF(0), direct_sql->server_name);
error:
  free_root(&mem_root, MYF(0));
  DBUG_RETURN(error_num);
}

#define SPIDER_PARAM_LEN(name) name ## _length
#define SPIDER_PARAM_STR(title_name, param_name)                        \
  if (!strncasecmp(parse.start_title, title_name, title_length))        \
  {                                                                     \
    DBUG_PRINT("info",("spider " title_name " start"));                 \
    if (!direct_sql->param_name)                                        \
    {                                                                   \
      if ((direct_sql->param_name = spider_create_string(parse.start_value, \
                                                         value_length))) \
        direct_sql->SPIDER_PARAM_LEN(param_name) = strlen(direct_sql->param_name); \
      else {                                                            \
        error_num= parse.fail(true);                                    \
        goto error;                                                     \
      }                                                                 \
      DBUG_PRINT("info",("spider " title_name "=%s", direct_sql->param_name)); \
    } \
    break; \
  }
#define SPIDER_PARAM_HINT_WITH_MAX(title_name, param_name, check_length, max_size, min_val, max_val) \
  if (!strncasecmp(parse.start_title, title_name, check_length)) \
  { \
    DBUG_PRINT("info",("spider " title_name " start")); \
    DBUG_PRINT("info",("spider max_size=%d", max_size)); \
    int hint_num = atoi(parse.start_title + check_length) - 1; \
    DBUG_PRINT("info",("spider hint_num=%d", hint_num)); \
    DBUG_PRINT("info",("spider direct_sql->param_name=%p", \
                       direct_sql->param_name)); \
    if (direct_sql->param_name) \
    { \
      if (hint_num < 0 || hint_num >= max_size) \
      { \
        error_num= parse.fail(true); \
        goto error; \
      } else if (direct_sql->param_name[hint_num] != -1) \
        break; \
      direct_sql->param_name[hint_num] = atoi(parse.start_value); \
      if (direct_sql->param_name[hint_num] < min_val) \
        direct_sql->param_name[hint_num] = min_val; \
      else if (direct_sql->param_name[hint_num] > max_val) \
        direct_sql->param_name[hint_num] = max_val; \
      DBUG_PRINT("info",("spider " title_name "[%d]=%d", hint_num, \
                         direct_sql->param_name[hint_num])); \
    } else { \
      error_num= parse.fail(true); \
      goto error; \
    } \
    break; \
  }
#define SPIDER_PARAM_INT_WITH_MAX(title_name, param_name, min_val, max_val) \
  if (!strncasecmp(parse.start_title, title_name, title_length)) \
  { \
    DBUG_PRINT("info",("spider " title_name " start")); \
    if (direct_sql->param_name == -1) \
    { \
      direct_sql->param_name = atoi(parse.start_value); \
      if (direct_sql->param_name < min_val) \
        direct_sql->param_name = min_val; \
      else if (direct_sql->param_name > max_val) \
        direct_sql->param_name = max_val; \
      DBUG_PRINT("info",("spider " title_name "=%d", \
                         (int) direct_sql->param_name)); \
    } \
    break; \
  }
#define SPIDER_PARAM_INT(title_name, param_name, min_val) \
  if (!strncasecmp(parse.start_title, title_name, title_length)) \
  { \
    DBUG_PRINT("info",("spider " title_name " start")); \
    if (direct_sql->param_name == -1) \
    { \
      direct_sql->param_name = atoi(parse.start_value); \
      if (direct_sql->param_name < min_val) \
        direct_sql->param_name = min_val; \
      DBUG_PRINT("info",("spider " title_name "=%d", direct_sql->param_name)); \
    } \
    break; \
  }
#define SPIDER_PARAM_LONGLONG(title_name, param_name, min_val) \
  if (!strncasecmp(parse.start_title, title_name, title_length)) \
  { \
    DBUG_PRINT("info",("spider " title_name " start")); \
    if (direct_sql->param_name == -1) \
    { \
      direct_sql->param_name = my_strtoll10(parse.start_value, (char**) NULL, \
                                            &error_num); \
      if (direct_sql->param_name < min_val) \
        direct_sql->param_name = min_val; \
      DBUG_PRINT("info",("spider " title_name "=%lld", direct_sql->param_name)); \
    } \
    break; \
  }

static void spider_minus_1(SPIDER_DIRECT_SQL *direct_sql)
{
  direct_sql->tgt_port = -1;
  direct_sql->tgt_ssl_vsc = -1;
  direct_sql->table_loop_mode = -1;
  direct_sql->priority = -1;
  direct_sql->connect_timeout = -1;
  direct_sql->net_read_timeout = -1;
  direct_sql->net_write_timeout = -1;
  direct_sql->bulk_insert_rows = -1;
  direct_sql->connection_channel = -1;
  direct_sql->use_real_table = -1;
  direct_sql->error_rw_mode = -1;
  for (int i = 0; i < direct_sql->table_count; i++)
    direct_sql->iop[i] = -1;
}

int spider_udf_parse_direct_sql_param(
  SPIDER_TRX *trx,
  SPIDER_DIRECT_SQL *direct_sql,
  const char *param,
  int param_length
) {
  int error_num = 0;
  char *param_string = NULL;
  char *start_param;
  int title_length, value_length;
  SPIDER_PARAM_STRING_PARSE parse;
  DBUG_ENTER("spider_udf_parse_direct_sql_param");

  spider_minus_1(direct_sql);
  if (param_length == 0)
    goto set_default;
  DBUG_PRINT("info",("spider create param_string string"));
  if (!(param_string = spider_create_string(param, param_length)))
  {
    error_num = HA_ERR_OUT_OF_MEM;
    my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
    goto error_alloc_param_string;
  }
  DBUG_PRINT("info",("spider param_string=%s", param_string));

  start_param = param_string;
  parse.error_num = ER_SPIDER_INVALID_UDF_PARAM_NUM;
  while (*start_param != '\0')
  {
    if (parse.locate_param_def(start_param))
    {
      error_num= parse.fail(false);
      goto error;
    }
    /* Null the end of the parameter value. */
    *parse.end_value= '\0';
    value_length= (int) (parse.end_value - parse.start_value);

    switch (title_length = (int) (parse.end_title - parse.start_title))
    {
      case 0:
        error_num= parse.fail(true);
        goto error;
      case 3:
        SPIDER_PARAM_LONGLONG("bir", bulk_insert_rows, 0);
        SPIDER_PARAM_INT_WITH_MAX("cch", connection_channel, 0, 63);
        SPIDER_PARAM_INT("cto", connect_timeout, 0);
        SPIDER_PARAM_STR("dff", tgt_default_file);
        SPIDER_PARAM_STR("dfg", tgt_default_group);
        SPIDER_PARAM_STR("drv", tgt_driver);
        SPIDER_PARAM_STR("dsn", tgt_dsn);
        SPIDER_PARAM_STR("fds", tgt_filedsn);
        SPIDER_PARAM_LONGLONG("prt", priority, 0);
        SPIDER_PARAM_INT("rto", net_read_timeout, 0);
        SPIDER_PARAM_STR("sca", tgt_ssl_ca);
        SPIDER_PARAM_STR("sch", tgt_ssl_cipher);
        SPIDER_PARAM_STR("scp", tgt_ssl_capath);
        SPIDER_PARAM_STR("scr", tgt_ssl_cert);
        SPIDER_PARAM_STR("sky", tgt_ssl_key);
        SPIDER_PARAM_STR("srv", server_name);
        SPIDER_PARAM_INT_WITH_MAX("svc", tgt_ssl_vsc, 0, 1);
        SPIDER_PARAM_INT_WITH_MAX("tlm", table_loop_mode, 0, 2);
        SPIDER_PARAM_INT_WITH_MAX("urt", use_real_table, 0, 1);
        SPIDER_PARAM_INT("wto", net_write_timeout, 0);
        error_num= parse.fail(true);
        goto error;
      case 4:
        SPIDER_PARAM_INT_WITH_MAX("erwm", error_rw_mode, 0, 1);
        SPIDER_PARAM_STR("host", tgt_host);
        SPIDER_PARAM_INT_WITH_MAX("port", tgt_port, 0, 65535);
        SPIDER_PARAM_STR("user", tgt_username);
        error_num= parse.fail(true);
        goto error;
      case 6:
        SPIDER_PARAM_STR("driver", tgt_driver);
        SPIDER_PARAM_STR("server", server_name);
        SPIDER_PARAM_STR("socket", tgt_socket);
        SPIDER_PARAM_HINT_WITH_MAX("iop", iop, 3, direct_sql->table_count, 0, 2);
        SPIDER_PARAM_STR("ssl_ca", tgt_ssl_ca);
        error_num= parse.fail(true);
        goto error;
      case 7:
        SPIDER_PARAM_STR("filedsn", tgt_filedsn);
        SPIDER_PARAM_STR("wrapper", tgt_wrapper);
        SPIDER_PARAM_STR("ssl_key", tgt_ssl_key);
        error_num= parse.fail(true);
        goto error;
      case 8:
        SPIDER_PARAM_STR("database", tgt_default_db_name);
        SPIDER_PARAM_STR("password", tgt_password);
        SPIDER_PARAM_LONGLONG("priority", priority, 0);
        SPIDER_PARAM_STR("ssl_cert", tgt_ssl_cert);
        error_num= parse.fail(true);
        goto error;
      case 10:
        SPIDER_PARAM_STR("ssl_cipher", tgt_ssl_cipher);
        SPIDER_PARAM_STR("ssl_capath", tgt_ssl_capath);
        error_num= parse.fail(true);
        goto error;
      case 11:
        error_num= parse.fail(true);
        goto error;
      case 12:
        SPIDER_PARAM_STR("default_file", tgt_default_file);
        error_num= parse.fail(true);
        goto error;
      case 13:
        SPIDER_PARAM_STR("default_group", tgt_default_group);
        SPIDER_PARAM_INT_WITH_MAX("error_rw_mode", error_rw_mode, 0, 1);
        error_num= parse.fail(true);
        goto error;
      case 14:
        SPIDER_PARAM_INT_WITH_MAX("use_real_table", use_real_table, 0, 1);
        error_num= parse.fail(true);
        goto error;
      case 15:
        SPIDER_PARAM_INT_WITH_MAX("table_loop_mode", table_loop_mode, 0, 2);
        SPIDER_PARAM_INT("connect_timeout", connect_timeout, 0);
        error_num= parse.fail(true);
        goto error;
      case 16:
        SPIDER_PARAM_LONGLONG("bulk_insert_rows", bulk_insert_rows, 1);
        SPIDER_PARAM_INT("net_read_timeout", net_read_timeout, 0);
        error_num= parse.fail(true);
        goto error;
      case 17:
        SPIDER_PARAM_INT("net_write_timeout", net_write_timeout, 0);
        error_num= parse.fail(true);
        goto error;
      case 18:
        SPIDER_PARAM_INT_WITH_MAX(
          "connection_channel", connection_channel, 0, 63);
        error_num= parse.fail(true);
        goto error;
      case 22:
        SPIDER_PARAM_INT_WITH_MAX("ssl_verify_server_cert", tgt_ssl_vsc, 0, 1);
        error_num= parse.fail(true);
        goto error;
      default:
        error_num= parse.fail(true);
        goto error;
    }
    /* Restore delim */
    *parse.end_value= parse.delim_value;
  }

set_default:
  error_num = spider_udf_set_direct_sql_param_default(trx, direct_sql);
error:
  if (param_string)
    spider_free(spider_current_trx, param_string, MYF(0));
error_alloc_param_string:
  DBUG_RETURN(error_num);
}

int spider_udf_set_direct_sql_param_default(
  SPIDER_TRX *trx,
  SPIDER_DIRECT_SQL *direct_sql
) {
  bool check_socket;
  bool check_database;
  bool check_default_file;
  bool check_host;
  bool check_port;
  bool socket_has_default_value;
  bool database_has_default_value;
  bool default_file_has_default_value;
  bool host_has_default_value;
  bool port_has_default_value;
  int error_num, roop_count;
  DBUG_ENTER("spider_udf_set_direct_sql_param_default");
  if (direct_sql->server_name)
  {
    if ((error_num = spider_udf_direct_sql_get_server(direct_sql)))
      DBUG_RETURN(error_num);
  }

  if (
    !direct_sql->tgt_socket &&
    (!direct_sql->tgt_host || !strcmp(direct_sql->tgt_host, my_localhost))
  ) {
    check_socket = TRUE;
  } else {
    check_socket = FALSE;
  }
  if (!direct_sql->tgt_default_db_name)
  {
    check_database = TRUE;
  } else {
    check_database = FALSE;
  }
  if (
    !direct_sql->tgt_default_file &&
    direct_sql->tgt_default_group &&
    (*spd_defaults_file || *spd_defaults_extra_file)
  ) {
    check_default_file = TRUE;
  } else {
    check_default_file = FALSE;
  }
  if (!direct_sql->tgt_host)
  {
    check_host = TRUE;
  } else {
    check_host = FALSE;
  }
  if (direct_sql->tgt_port == -1)
  {
    check_port = TRUE;
  } else {
    check_port = FALSE;
  }
  if (check_socket || check_database || check_default_file || check_host ||
    check_port)
  {
    socket_has_default_value = check_socket;
    database_has_default_value = check_database;
    default_file_has_default_value = check_default_file;
    host_has_default_value = check_host;
    port_has_default_value = check_port;
    if (direct_sql->tgt_wrapper)
    {
      for (roop_count = 0; roop_count < SPIDER_DBTON_SIZE; roop_count++)
      {
        DBUG_PRINT("info",("spider direct_sql->tgt_wrapper=%s",
          direct_sql->tgt_wrapper));
        DBUG_PRINT("info",("spider spider_dbton[%d].wrapper=%s", roop_count,
          spider_dbton[roop_count].wrapper ?
            spider_dbton[roop_count].wrapper : "NULL"));
        if (
          spider_dbton[roop_count].wrapper &&
          !strcasecmp(direct_sql->tgt_wrapper,
            spider_dbton[roop_count].wrapper)
        ) {
          if (spider_dbton[roop_count].db_access_type ==
            SPIDER_DB_ACCESS_TYPE_SQL)
          {
            if (check_socket)
            {
              socket_has_default_value = spider_dbton[roop_count].
                db_util->socket_has_default_value();
            }
            if (check_database)
            {
              database_has_default_value = spider_dbton[roop_count].
                db_util->database_has_default_value();
            }
            if (check_default_file)
            {
              default_file_has_default_value = spider_dbton[roop_count].
                db_util->default_file_has_default_value();
            }
            if (check_host)
            {
              host_has_default_value = spider_dbton[roop_count].
                db_util->host_has_default_value();
            }
            if (check_port)
            {
              port_has_default_value = spider_dbton[roop_count].
                db_util->port_has_default_value();
            }
            break;
          }
        }
      }
    }
  } else {
    socket_has_default_value = FALSE;
    database_has_default_value = FALSE;
    default_file_has_default_value = FALSE;
    host_has_default_value = FALSE;
    port_has_default_value = FALSE;
  }

  if (database_has_default_value)
  {
    DBUG_PRINT("info",("spider create default tgt_default_db_name"));
    direct_sql->tgt_default_db_name_length = SPIDER_THD_db_length(trx->thd);
    if (
      !(direct_sql->tgt_default_db_name = spider_create_string(
          trx->thd->get_db(),
        direct_sql->tgt_default_db_name_length))
    ) {
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
    }
  }

  if (!direct_sql->tgt_wrapper)
  {
    DBUG_PRINT("info",("spider create default tgt_wrapper"));
    direct_sql->tgt_wrapper_length = SPIDER_DB_WRAPPER_LEN;
    if (
      !(direct_sql->tgt_wrapper = spider_create_string(
        SPIDER_DB_WRAPPER_STR,
        direct_sql->tgt_wrapper_length))
    ) {
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
    }
  }

  if (host_has_default_value)
  {
    DBUG_PRINT("info",("spider create default tgt_host"));
    direct_sql->tgt_host_length = strlen(my_localhost);
    if (
      !(direct_sql->tgt_host = spider_create_string(
        my_localhost,
        direct_sql->tgt_host_length))
    ) {
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
    }
  }

  if (default_file_has_default_value)
  {
    DBUG_PRINT("info",("spider create default tgt_default_file"));
    if (*spd_defaults_extra_file)
    {
      direct_sql->tgt_default_file_length = strlen(*spd_defaults_extra_file);
      if (
        !(direct_sql->tgt_default_file = spider_create_string(
          *spd_defaults_extra_file,
          direct_sql->tgt_default_file_length))
      ) {
        my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
        DBUG_RETURN(HA_ERR_OUT_OF_MEM);
      }
    } else {
      direct_sql->tgt_default_file_length = strlen(*spd_defaults_file);
      if (
        !(direct_sql->tgt_default_file = spider_create_string(
          *spd_defaults_file,
          direct_sql->tgt_default_file_length))
      ) {
        my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
        DBUG_RETURN(HA_ERR_OUT_OF_MEM);
      }
    }
  }

  if (port_has_default_value)
  {
      direct_sql->tgt_port = MYSQL_PORT;
  }
  else if (direct_sql->tgt_port < 0)
    direct_sql->tgt_port = 0;
  else if (direct_sql->tgt_port > 65535)
    direct_sql->tgt_port = 65535;

  if (direct_sql->tgt_ssl_vsc == -1)
    direct_sql->tgt_ssl_vsc = 0;

  if (socket_has_default_value)
  {
    DBUG_PRINT("info",("spider create default tgt_socket"));
    direct_sql->tgt_socket_length = strlen((char *) MYSQL_UNIX_ADDR);
    if (
      !(direct_sql->tgt_socket = spider_create_string(
        (char *) MYSQL_UNIX_ADDR,
        direct_sql->tgt_socket_length))
    ) {
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
    }
  }

  if (direct_sql->table_loop_mode == -1)
    direct_sql->table_loop_mode = 0;
  if (direct_sql->priority == -1)
    direct_sql->priority = 1000000;
  if (direct_sql->bulk_insert_rows == -1)
    direct_sql->bulk_insert_rows = 3000;
  if (direct_sql->connection_channel == -1)
    direct_sql->connection_channel = 0;
  if (direct_sql->use_real_table == -1)
    direct_sql->use_real_table = 0;
  if (direct_sql->error_rw_mode == -1)
    direct_sql->error_rw_mode = 0;
  for (roop_count = 0; roop_count < direct_sql->table_count; roop_count++)
  {
    if (direct_sql->iop[roop_count] == -1)
      direct_sql->iop[roop_count] = 0;
  }
  DBUG_RETURN(0);
}

void spider_udf_free_direct_sql_alloc(
  SPIDER_DIRECT_SQL *direct_sql,
  my_bool bg
) {
  SPIDER_BG_DIRECT_SQL *bg_direct_sql;
  DBUG_ENTER("spider_udf_free_direct_sql_alloc");
  if (bg)
  {
    pthread_mutex_lock(direct_sql->bg_mutex);
    bg_direct_sql = (SPIDER_BG_DIRECT_SQL *) direct_sql->parent;
    if (bg_direct_sql->direct_sql == direct_sql)
      bg_direct_sql->direct_sql = direct_sql->next;
    if (direct_sql->next)
      direct_sql->next->prev = direct_sql->prev;
    if (direct_sql->prev)
      direct_sql->prev->next = direct_sql->next;
    pthread_cond_signal(direct_sql->bg_cond);
    pthread_mutex_unlock(direct_sql->bg_mutex);
  }
  if (direct_sql->real_table_used && direct_sql->open_tables_thd)
  {
    spider_sys_close_table(direct_sql->open_tables_thd,
      &direct_sql->open_tables_backup);
  }
  if (direct_sql->server_name)
  {
    spider_free(spider_current_trx, direct_sql->server_name, MYF(0));
  }
  if (direct_sql->tgt_default_db_name)
  {
    spider_free(spider_current_trx, direct_sql->tgt_default_db_name, MYF(0));
  }
  if (direct_sql->tgt_host)
  {
    spider_free(spider_current_trx, direct_sql->tgt_host, MYF(0));
  }
  if (direct_sql->tgt_username)
  {
    spider_free(spider_current_trx, direct_sql->tgt_username, MYF(0));
  }
  if (direct_sql->tgt_password)
  {
    spider_free(spider_current_trx, direct_sql->tgt_password, MYF(0));
  }
  if (direct_sql->tgt_socket)
  {
    spider_free(spider_current_trx, direct_sql->tgt_socket, MYF(0));
  }
  if (direct_sql->tgt_wrapper)
  {
    spider_free(spider_current_trx, direct_sql->tgt_wrapper, MYF(0));
  }
  if (direct_sql->tgt_ssl_ca)
  {
    spider_free(spider_current_trx, direct_sql->tgt_ssl_ca, MYF(0));
  }
  if (direct_sql->tgt_ssl_capath)
  {
    spider_free(spider_current_trx, direct_sql->tgt_ssl_capath, MYF(0));
  }
  if (direct_sql->tgt_ssl_cert)
  {
    spider_free(spider_current_trx, direct_sql->tgt_ssl_cert, MYF(0));
  }
  if (direct_sql->tgt_ssl_cipher)
  {
    spider_free(spider_current_trx, direct_sql->tgt_ssl_cipher, MYF(0));
  }
  if (direct_sql->tgt_ssl_key)
  {
    spider_free(spider_current_trx, direct_sql->tgt_ssl_key, MYF(0));
  }
  if (direct_sql->tgt_default_file)
  {
    spider_free(spider_current_trx, direct_sql->tgt_default_file, MYF(0));
  }
  if (direct_sql->tgt_default_group)
  {
    spider_free(spider_current_trx, direct_sql->tgt_default_group, MYF(0));
  }
  if (direct_sql->tgt_dsn)
  {
    spider_free(spider_current_trx, direct_sql->tgt_dsn, MYF(0));
  }
  if (direct_sql->tgt_filedsn)
  {
    spider_free(spider_current_trx, direct_sql->tgt_filedsn, MYF(0));
  }
  if (direct_sql->tgt_driver)
  {
    spider_free(spider_current_trx, direct_sql->tgt_driver, MYF(0));
  }
  if (direct_sql->tgt_odbc_conn_str)
  {
    spider_free(spider_current_trx, direct_sql->tgt_odbc_conn_str, MYF(0));
  }
  if (direct_sql->conn_key)
  {
    spider_free(spider_current_trx, direct_sql->conn_key, MYF(0));
  }
  if (direct_sql->db_names)
  {
    spider_free(spider_current_trx, direct_sql->db_names, MYF(0));
  }
  spider_free(spider_current_trx, direct_sql, MYF(0));
  DBUG_VOID_RETURN;
}

long long spider_direct_sql_body(
  UDF_INIT *initid,
  UDF_ARGS *args,
  unsigned char *is_null,
  unsigned char *error,
  my_bool bg
) {
  int error_num, roop_count;
  SPIDER_DIRECT_SQL *direct_sql = NULL, *tmp_direct_sql;
  THD *thd = current_thd;
  SPIDER_TRX *trx;
  SPIDER_CONN *conn;
  char *sql;
  TABLE_LIST table_list;
  SPIDER_BG_DIRECT_SQL *bg_direct_sql;
  TABLE_LIST *real_table_list_last = NULL;
  uint use_real_table = 0;
  DBUG_ENTER("spider_direct_sql_body");
  SPIDER_BACKUP_DASTATUS;
  if (!(direct_sql = (SPIDER_DIRECT_SQL *)
    spider_bulk_malloc(spider_current_trx, SPD_MID_DIRECT_SQL_BODY_1, MYF(MY_WME | MY_ZEROFILL),
      &direct_sql, (uint) (sizeof(SPIDER_DIRECT_SQL)),
      &sql, (uint) (sizeof(char) * args->lengths[0]),
      NullS))
  ) {
    error_num = HA_ERR_OUT_OF_MEM;
    my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
    goto error;
  }
  if (bg)
  {
    bg_direct_sql = (SPIDER_BG_DIRECT_SQL *) initid->ptr;
    pthread_mutex_lock(&bg_direct_sql->bg_mutex);
    tmp_direct_sql = (SPIDER_DIRECT_SQL *) bg_direct_sql->direct_sql;
    bg_direct_sql->direct_sql = direct_sql;
    if (tmp_direct_sql)
    {
      tmp_direct_sql->prev = direct_sql;
      direct_sql->next = tmp_direct_sql;
    }
    pthread_mutex_unlock(&bg_direct_sql->bg_mutex);
    direct_sql->bg_mutex = &bg_direct_sql->bg_mutex;
    direct_sql->bg_cond = &bg_direct_sql->bg_cond;
    direct_sql->parent = bg_direct_sql;
    bg_direct_sql->called_cnt++;
  }
  if (!(trx = spider_get_trx(thd, TRUE, &error_num)))
  {
    if (error_num == HA_ERR_OUT_OF_MEM)
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
    goto error;
  }
  direct_sql->trx = trx;

  if (args->args[1])
  {
    if ((error_num = spider_udf_direct_sql_create_table_list(
      direct_sql,
      args->args[1],
      args->lengths[1]
    ))) {
      if (error_num == HA_ERR_OUT_OF_MEM)
        my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
  } else {
    if ((error_num = spider_udf_direct_sql_create_table_list(
      direct_sql,
      (char *) "",
      0
    ))) {
      if (error_num == HA_ERR_OUT_OF_MEM)
        my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
  }
  if (args->args[2])
  {
    if ((error_num = spider_udf_parse_direct_sql_param(
      trx,
      direct_sql,
      args->args[2],
      args->lengths[2]
    ))) {
      goto error;
    }
  } else {
    if ((error_num = spider_udf_parse_direct_sql_param(
      trx,
      direct_sql,
      "",
      0
    ))) {
      goto error;
    }
  }
    trx->updated_in_this_trx = TRUE;
    DBUG_PRINT("info",("spider trx->updated_in_this_trx=TRUE"));
    use_real_table= direct_sql->use_real_table;
    for (roop_count= 0; roop_count < direct_sql->table_count; roop_count++)
    {
#ifdef SPIDER_NEED_INIT_ONE_TABLE_FOR_FIND_TEMPORARY_TABLE
#ifdef SPIDER_use_LEX_CSTRING_for_database_tablename_alias
    LEX_CSTRING db_name =
    {
      direct_sql->db_names[roop_count],
      strlen(direct_sql->db_names[roop_count])
    };
    LEX_CSTRING tbl_name =
    {
      direct_sql->table_names[roop_count],
      strlen(direct_sql->table_names[roop_count])
    };
    table_list.init_one_table(&db_name, &tbl_name, 0, TL_WRITE);
#else
    table_list.init_one_table(direct_sql->db_names[roop_count],
      strlen(direct_sql->db_names[roop_count]),
      direct_sql->table_names[roop_count],
      strlen(direct_sql->table_names[roop_count]),
      direct_sql->table_names[roop_count], TL_WRITE);
#endif
#else
    SPIDER_TABLE_LIST_db_str(&table_list) = direct_sql->db_names[roop_count];
    SPIDER_TABLE_LIST_table_name_str(&table_list) =
      direct_sql->table_names[roop_count];
#endif
    if (!(direct_sql->tables[roop_count] =
      spider_find_temporary_table(thd, &table_list)))
    {
      if (!use_real_table)
      {
        error_num = ER_SPIDER_UDF_TMP_TABLE_NOT_FOUND_NUM;
        my_printf_error(ER_SPIDER_UDF_TMP_TABLE_NOT_FOUND_NUM,
          ER_SPIDER_UDF_TMP_TABLE_NOT_FOUND_STR,
          MYF(0), SPIDER_TABLE_LIST_db_str(&table_list),
          SPIDER_TABLE_LIST_table_name_str(&table_list));
        goto error;
      }
      TABLE_LIST *tables = &direct_sql->table_list[roop_count];
      MDL_REQUEST_INIT(&tables->mdl_request, MDL_key::TABLE,
        SPIDER_TABLE_LIST_db_str(&table_list),
        SPIDER_TABLE_LIST_table_name_str(&table_list),
        MDL_SHARED_WRITE, MDL_TRANSACTION);
      if (!direct_sql->table_list_first)
      {
        direct_sql->table_list_first = tables;
      } else {
        real_table_list_last->next_global = tables;
      }
      real_table_list_last = tables;
      spider_set_bit(direct_sql->real_table_bitmap, roop_count);
      direct_sql->real_table_used = TRUE;
    }
  }
  if ((error_num = spider_udf_direct_sql_create_conn_key(direct_sql)))
  {
    if (error_num == HA_ERR_OUT_OF_MEM)
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
    goto error;
  }
  if (!(conn = spider_udf_direct_sql_get_conn(direct_sql, trx, &error_num)))
  {
    if (error_num == HA_ERR_OUT_OF_MEM)
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
    goto error;
  }
  conn->error_mode = 0;
  direct_sql->conn = conn;
  if ((error_num = spider_db_udf_check_and_set_set_names(trx)))
  {
    if (error_num == HA_ERR_OUT_OF_MEM)
      my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
    goto error;
  }
  if (args->args[0])
  {
    direct_sql->sql_length = args->lengths[0];
    memcpy(sql, args->args[0], direct_sql->sql_length);
  } else
    direct_sql->sql_length = 0;
  direct_sql->sql = sql;

  if (bg)
  {
    if ((error_num = spider_udf_bg_direct_sql(direct_sql)))
    {
      if (error_num == HA_ERR_OUT_OF_MEM)
        my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
  } else {
    if (conn->bg_init)
      pthread_mutex_lock(&conn->bg_conn_mutex);
    if ((error_num = spider_db_udf_direct_sql(direct_sql)))
    {
      if (conn->bg_init)
        pthread_mutex_unlock(&conn->bg_conn_mutex);
      if (direct_sql->modified_non_trans_table)
        thd->transaction->stmt.modified_non_trans_table = TRUE;
      if (error_num == HA_ERR_OUT_OF_MEM)
        my_error(ER_OUT_OF_RESOURCES, MYF(0), HA_ERR_OUT_OF_MEM);
      goto error;
    }
    if (conn->bg_init)
      pthread_mutex_unlock(&conn->bg_conn_mutex);
    if (direct_sql->modified_non_trans_table)
      thd->transaction->stmt.modified_non_trans_table = TRUE;
  }
  if (!bg)
  {
    spider_udf_free_direct_sql_alloc(direct_sql, FALSE);
  }
  DBUG_RETURN(1);

error:
  if (direct_sql)
  {
    if (
      direct_sql->error_rw_mode &&
      spider_db_conn_is_network_error(error_num)
    ) {
      SPIDER_RESTORE_DASTATUS;
      spider_udf_free_direct_sql_alloc(direct_sql, bg);
      DBUG_RETURN(1);
    }
    spider_udf_free_direct_sql_alloc(direct_sql, bg);
  }
  *error = 1;
  DBUG_RETURN(0);
}

my_bool spider_direct_sql_init_body(
  UDF_INIT *initid,
  UDF_ARGS *args,
  char *message,
  my_bool bg
) {
  SPIDER_BG_DIRECT_SQL *bg_direct_sql;
  DBUG_ENTER("spider_direct_sql_init_body");
  if (!spider_hton_ptr)
  {
    strcpy(message, "Plugin 'SPIDER' is not loaded");
    goto error;
  }
  if (args->arg_count != 3)
  {
    strcpy(message, "spider_(bg)_direct_sql() requires 3 arguments");
    goto error;
  }
  if (
    args->arg_type[0] != STRING_RESULT ||
    args->arg_type[1] != STRING_RESULT ||
    args->arg_type[2] != STRING_RESULT
  ) {
    strcpy(message, "spider_(bg)_direct_sql() requires string arguments");
    goto error;
  }
  if (bg)
  {
    if (!(bg_direct_sql = (SPIDER_BG_DIRECT_SQL *)
      spider_malloc(spider_current_trx, SPD_MID_DIRECT_SQL_INIT_BODY_1, sizeof(SPIDER_BG_DIRECT_SQL),
      MYF(MY_WME | MY_ZEROFILL)))
    ) {
      strcpy(message, "spider_bg_direct_sql() out of memory");
      goto error;
    }
    if (mysql_mutex_init(spd_key_mutex_bg_direct_sql,
      &bg_direct_sql->bg_mutex, MY_MUTEX_INIT_FAST))
    {
      strcpy(message, "spider_bg_direct_sql() out of memory");
      goto error_mutex_init;
    }
    if (mysql_cond_init(spd_key_cond_bg_direct_sql,
      &bg_direct_sql->bg_cond, NULL))
    {
      strcpy(message, "spider_bg_direct_sql() out of memory");
      goto error_cond_init;
    }
    initid->ptr = (char *) bg_direct_sql;
  }
  DBUG_RETURN(FALSE);

error_cond_init:
  pthread_mutex_destroy(&bg_direct_sql->bg_mutex);
error_mutex_init:
  spider_free(spider_current_trx, bg_direct_sql, MYF(0));
error:
  DBUG_RETURN(TRUE);
}

void spider_direct_sql_deinit_body(
  UDF_INIT *initid
) {
  SPIDER_BG_DIRECT_SQL *bg_direct_sql = (SPIDER_BG_DIRECT_SQL *) initid->ptr;
  DBUG_ENTER("spider_direct_sql_deinit_body");
  if (bg_direct_sql)
  {
    pthread_mutex_lock(&bg_direct_sql->bg_mutex);
    while (bg_direct_sql->direct_sql)
      pthread_cond_wait(&bg_direct_sql->bg_cond, &bg_direct_sql->bg_mutex);
    pthread_mutex_unlock(&bg_direct_sql->bg_mutex);
    if (bg_direct_sql->modified_non_trans_table)
    {
      THD *thd = current_thd;
      thd->transaction->stmt.modified_non_trans_table = TRUE;
    }
    pthread_cond_destroy(&bg_direct_sql->bg_cond);
    pthread_mutex_destroy(&bg_direct_sql->bg_mutex);
    spider_free(spider_current_trx, bg_direct_sql, MYF(0));
  }
  DBUG_VOID_RETURN;
}

void spider_direct_sql_bg_start(
  UDF_INIT *initid
) {
  SPIDER_BG_DIRECT_SQL *bg_direct_sql = (SPIDER_BG_DIRECT_SQL *) initid->ptr;
  DBUG_ENTER("spider_direct_sql_bg_start");
  bg_direct_sql->called_cnt = 0;
  bg_direct_sql->bg_error = 0;
  DBUG_VOID_RETURN;
}

long long spider_direct_sql_bg_end(
  UDF_INIT *initid
) {
  THD *thd = current_thd;
  SPIDER_BG_DIRECT_SQL *bg_direct_sql = (SPIDER_BG_DIRECT_SQL *) initid->ptr;
  DBUG_ENTER("spider_direct_sql_bg_end");
  pthread_mutex_lock(&bg_direct_sql->bg_mutex);
  while (bg_direct_sql->direct_sql)
    pthread_cond_wait(&bg_direct_sql->bg_cond, &bg_direct_sql->bg_mutex);
  pthread_mutex_unlock(&bg_direct_sql->bg_mutex);
  if (bg_direct_sql->modified_non_trans_table)
    thd->transaction->stmt.modified_non_trans_table = TRUE;
  if (bg_direct_sql->bg_error)
  {
    my_message(bg_direct_sql->bg_error, bg_direct_sql->bg_error_msg, MYF(0));
    DBUG_RETURN(0);
  }
  DBUG_RETURN(bg_direct_sql->called_cnt);
}

int spider_udf_bg_direct_sql(
  SPIDER_DIRECT_SQL *direct_sql
) {
  int error_num;
  SPIDER_CONN *conn = direct_sql->conn;
  DBUG_ENTER("spider_udf_bg_direct_sql");
  if ((error_num = spider_create_conn_thread(conn)))
    DBUG_RETURN(error_num);
  if (!pthread_mutex_trylock(&conn->bg_conn_mutex))
  {
    DBUG_PRINT("info",("spider get bg_conn_mutex"));
    conn->bg_target = direct_sql;
    conn->bg_direct_sql = TRUE;
    conn->bg_caller_sync_wait = TRUE;
    pthread_mutex_lock(&conn->bg_conn_sync_mutex);
    pthread_cond_signal(&conn->bg_conn_cond);
    pthread_mutex_unlock(&conn->bg_conn_mutex);
    pthread_cond_wait(&conn->bg_conn_sync_cond, &conn->bg_conn_sync_mutex);
    pthread_mutex_unlock(&conn->bg_conn_sync_mutex);
    conn->bg_caller_sync_wait = FALSE;
  } else {
    DBUG_PRINT("info",("spider get put job stack"));
    bool bg_get_job_stack = FALSE;
    pthread_mutex_lock(&conn->bg_job_stack_mutex);
    uint old_elements = conn->bg_job_stack.max_element;
    if (insert_dynamic(&conn->bg_job_stack, (uchar *) &direct_sql))
    {
      pthread_mutex_unlock(&conn->bg_job_stack_mutex);
      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
    }
    if (conn->bg_job_stack.max_element > old_elements)
    {
      spider_alloc_calc_mem(spider_current_trx,
        conn->bg_job_stack,
        (conn->bg_job_stack.max_element - old_elements) *
        conn->bg_job_stack.size_of_element);
    }
    if (!conn->bg_get_job_stack_off)
      bg_get_job_stack = TRUE;
    pthread_mutex_unlock(&conn->bg_job_stack_mutex);
    if (bg_get_job_stack)
    {
      DBUG_PRINT("info",("spider get bg_conn_mutex"));
      pthread_mutex_lock(&conn->bg_conn_mutex);
      conn->bg_target = NULL;
      conn->bg_get_job_stack = TRUE;
      conn->bg_direct_sql = TRUE;
      conn->bg_caller_sync_wait = TRUE;
      pthread_mutex_lock(&conn->bg_conn_sync_mutex);
      pthread_cond_signal(&conn->bg_conn_cond);
      pthread_mutex_unlock(&conn->bg_conn_mutex);
      pthread_cond_wait(&conn->bg_conn_sync_cond, &conn->bg_conn_sync_mutex);
      pthread_mutex_unlock(&conn->bg_conn_sync_mutex);
      conn->bg_caller_sync_wait = FALSE;
    }
  }
  DBUG_RETURN(0);
}