/***********************************************************************

Copyright (c) 2011, 2020, Oracle and/or its affiliates. All rights reserved.

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

This program is also distributed with certain software (including
but not limited to OpenSSL) that is licensed under separate terms,
as designated in a particular file or component or in included license
documentation.  The authors of MySQL hereby grant you an additional
permission to link the program and your derivative works with the
separately licensed software that they have included with MySQL.

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, version 2.0, 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 St, Fifth Floor, Boston, MA 02110-1301  USA

***********************************************************************/

/**************************************************/ /**
 @file
 InnoDB Memcached configurations

 Created 04/12/2011 Jimmy Yang
 *******************************************************/

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "my_compiler.h"

#include "innodb_api.h"
#include "innodb_cb_api.h"
#include "innodb_config.h"
#include "innodb_utility.h"

/** Configure options enum IDs, their "names" and their default value */
option_t config_option_names[] = {
    {OPTION_ID_COL_SEP, COLUMN_SEPARATOR, {"|", 1}},
    {OPTION_ID_TBL_MAP_SEP, TABLE_MAP_SEPARATOR, {".", 1}}};

/**********************************************************************/ /**
 Makes a NUL-terminated copy of a nonterminated string.
 @return own: a copy of the string, must be deallocated by caller */
static char *my_strdupl(
    /*=======*/
    const char *str, /*!< in: string to be copied */
    int len)         /*!< in: length of str, in bytes */
{
  char *s = (char *)malloc(len + 1);

  if (!s) {
    return (NULL);
  }

  s[len] = 0;

  return ((char *)memcpy(s, str, len));
}

/**********************************************************************/ /**
 This function frees meta info structures */
void innodb_config_free(
    /*===============*/
    meta_cfg_info_t *item) /*!< in: meta info structure */
{
  int i;

  for (i = 0; i < CONTAINER_NUM_COLS; i++) {
    if (item->col_info[i].col_name) {
      free(item->col_info[i].col_name);
      item->col_info[i].col_name = NULL;
    }
  }

  if (item->index_info.idx_name) {
    free(item->index_info.idx_name);
    item->index_info.idx_name = NULL;
  }

  if (item->extra_col_info) {
    for (i = 0; i < item->n_extra_col; i++) {
      free(item->extra_col_info[i].col_name);
      item->extra_col_info[i].col_name = NULL;
    }

    free(item->extra_col_info);
    item->extra_col_info = NULL;
  }
}

/**********************************************************************/ /**
 This function parses possible multiple column name separated by ",", ";"
 or " " in the input "str" for the memcached "value" field.
 @return true if everything works out fine */
static bool innodb_config_parse_value_col(
    /*==========================*/
    meta_cfg_info_t *item, /*!< in: meta info structure */
    char *str,             /*!< in: column name(s) string */
    int len)               /*!< in: length of above string */
{
  static const char *sep = " ;,|\n";
  char *last;
  char *column_str;
  int num_cols = 0;
  char *my_str = my_strdupl(str, len);

  /* Find out how many column names in the string */
  for (column_str = strtok_r(my_str, sep, &last); column_str;
       column_str = strtok_r(NULL, sep, &last)) {
    num_cols++;
  }

  free(my_str);

  my_str = str;

  if (num_cols > 1) {
    int i = 0;
    item->extra_col_info =
        (meta_column_t *)malloc(num_cols * sizeof(*item->extra_col_info));

    if (!item->extra_col_info) {
      return (false);
    }

    for (column_str = strtok_r(my_str, sep, &last); column_str;
         column_str = strtok_r(NULL, sep, &last)) {
      item->extra_col_info[i].col_name_len = strlen(column_str);
      item->extra_col_info[i].col_name =
          my_strdupl(column_str, item->extra_col_info[i].col_name_len);
      item->extra_col_info[i].field_id = -1;
      i++;
    }

    item->n_extra_col = num_cols;
  } else {
    item->extra_col_info = NULL;
    item->n_extra_col = 0;
  }

  return (true);
}

/**********************************************************************/ /**
 This function opens the cache_policy configuration table, and find the
 table and column info that used for memcached data
 @return true if everything works out fine */
static bool innodb_read_cache_policy(
    /*=====================*/
    meta_cfg_info_t *item, /*!< in: meta info structure */
    void *thd)             /*!< in/out: MySQL THD */
{
  ib_trx_t ib_trx;
  ib_crsr_t crsr = NULL;
  ib_crsr_t idx_crsr = NULL;
  ib_tpl_t tpl = NULL;
  ib_err_t err = DB_SUCCESS;
  int n_cols MY_ATTRIBUTE((unused));
  int i;
  ib_ulint_t data_len;
  ib_col_meta_t col_meta;

  ib_trx = ib_cb_trx_begin(IB_TRX_READ_COMMITTED, true, false, thd);

  err = innodb_api_begin(NULL, MCI_CFG_DB_NAME, MCI_CFG_CACHE_POLICIES, NULL,
                         ib_trx, &crsr, &idx_crsr, IB_LOCK_S);

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: Cannot open config table"
            "'%s' in database '%s'. Error %d\n",
            MCI_CFG_CACHE_POLICIES, MCI_CFG_DB_NAME, err);
    err = DB_ERROR;
    goto func_exit;
  }

  tpl = innodb_cb_read_tuple_create(crsr);

  /* Currently, we support one table per memcached setup.
  We could extend that limit later */
  err = innodb_cb_cursor_first(crsr);

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate entry in"
            " config table '%s' in database '%s' \n",
            MCI_CFG_CACHE_POLICIES, MCI_CFG_DB_NAME);
    err = DB_ERROR;
    goto func_exit;
  }

  err = ib_cb_cursor_read_row(crsr, tpl, NULL, 0, NULL, NULL, NULL);

  n_cols = innodb_cb_tuple_get_n_cols(tpl);

  assert(n_cols >= CACHE_POLICY_NUM_COLS);

  for (i = 0; i < CACHE_POLICY_NUM_COLS; ++i) {
    char opt_name;
    meta_cache_opt_t opt_val;

    /* Skip cache policy name for now, We could have
    different cache policy stored, and switch dynamically */
    if (i == CACHE_POLICY_NAME) {
      continue;
    }

    data_len = innodb_cb_col_get_meta(tpl, i, &col_meta);

    if (data_len == IB_SQL_NULL) {
      opt_val = META_CACHE_OPT_INNODB;
    } else {
      opt_name = *(char *)innodb_cb_col_get_value(tpl, i);

      opt_val = (meta_cache_opt_t)opt_name;
    }

    if (opt_val >= META_CACHE_NUM_OPT || opt_val < META_CACHE_OPT_INNODB) {
      fprintf(stderr,
              " InnoDB_Memcached: Invalid Cache"
              " Policy %d. Reset to innodb_only\n",
              (int)opt_val);
      opt_val = META_CACHE_OPT_INNODB;
    }

    switch (i) {
      case CACHE_POLICY_GET:
        item->get_option = opt_val;
        break;
      case CACHE_POLICY_SET:
        item->set_option = opt_val;
        break;
      case CACHE_POLICY_DEL:
        item->del_option = opt_val;
        break;
      case CACHE_POLICY_FLUSH:
        item->flush_option = opt_val;
        break;
      default:
        assert(0);
    }
  }

func_exit:

  innodb_cb_cursor_close(crsr);

  if (tpl) {
    innodb_cb_tuple_delete(tpl);
  }

  innodb_cb_trx_commit(ib_trx);
  ib_cb_trx_release(ib_trx);

  return (err == DB_SUCCESS || err == DB_END_OF_INDEX);
}

/**********************************************************************/ /**
 This function opens the config_options configuration table, and find the
 table and column info that used for memcached data
 @return true if everything works out fine */
static bool innodb_read_config_option(
    /*======================*/
    meta_cfg_info_t *item, /*!< in: meta info structure */
    void *thd)             /*!< in/out: MySQL THD */
{
  ib_trx_t ib_trx;
  ib_crsr_t crsr = NULL;
  ib_crsr_t idx_crsr = NULL;
  ib_tpl_t tpl = NULL;
  ib_err_t err = DB_SUCCESS;
  int n_cols MY_ATTRIBUTE((unused));
  int i;
  ib_ulint_t data_len;
  ib_col_meta_t col_meta;
  int current_option = -1;

  ib_trx = ib_cb_trx_begin(IB_TRX_READ_COMMITTED, true, false, thd);
  err = innodb_api_begin(NULL, MCI_CFG_DB_NAME, MCI_CFG_CONFIG_OPTIONS, NULL,
                         ib_trx, &crsr, &idx_crsr, IB_LOCK_S);

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: Cannot open config table"
            "'%s' in database '%s'\n",
            MCI_CFG_CONFIG_OPTIONS, MCI_CFG_DB_NAME);
    err = DB_ERROR;
    goto func_exit;
  }

  tpl = innodb_cb_read_tuple_create(crsr);

  err = innodb_cb_cursor_first(crsr);

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate entry in"
            " config table '%s' in database '%s' \n",
            MCI_CFG_CONFIG_OPTIONS, MCI_CFG_DB_NAME);
    err = DB_ERROR;
    goto func_exit;
  }

  do {
    err = ib_cb_cursor_read_row(crsr, tpl, NULL, 0, NULL, NULL, NULL);

    if (err != DB_SUCCESS) {
      fprintf(stderr,
              " InnoDB_Memcached: failed to read"
              " row from config table '%s' in"
              " database '%s' \n",
              MCI_CFG_CONFIG_OPTIONS, MCI_CFG_DB_NAME);
      err = DB_ERROR;
      goto func_exit;
    }

    n_cols = innodb_cb_tuple_get_n_cols(tpl);

    assert(n_cols >= CONFIG_OPT_NUM_COLS);

    for (i = 0; i < CONFIG_OPT_NUM_COLS; ++i) {
      char *key;

      data_len = innodb_cb_col_get_meta(tpl, i, &col_meta);

      assert(data_len != IB_SQL_NULL);

      if (i == CONFIG_OPT_KEY) {
        int j;
        key = (char *)innodb_cb_col_get_value(tpl, i);
        current_option = -1;

        for (j = 0; j < OPTION_ID_NUM_OPTIONS; j++) {
          /* Currently, we only support one
          configure option, that is the string
          "separator" */
          if (strcmp(key, config_option_names[j].name) == 0) {
            current_option = config_option_names[j].id;
            break;
          }
        }
      }

      if (i == CONFIG_OPT_VALUE && current_option >= 0) {
        int max_len;

        /* The maximum length for delimiter is
        MAX_DELIMITER_LEN */
        max_len = (data_len > MAX_DELIMITER_LEN) ? MAX_DELIMITER_LEN : data_len;

        memcpy(item->options[current_option].value,
               innodb_cb_col_get_value(tpl, i), max_len);

        item->options[current_option].value[max_len] = 0;

        item->options[current_option].value_len = max_len;
      }
    }

    err = ib_cb_cursor_next(crsr);

  } while (err == DB_SUCCESS);

func_exit:

  innodb_cb_cursor_close(crsr);

  if (tpl) {
    innodb_cb_tuple_delete(tpl);
  }

  innodb_cb_trx_commit(ib_trx);
  ib_cb_trx_release(ib_trx);

  return (err == DB_SUCCESS || err == DB_END_OF_INDEX);
}

/**********************************************************************/ /**
 This function opens the "containers" configuration table, and find the
 table and column info that used for memcached data, and instantiates
 meta_cfg_info_t structure for such metadata.
 @return instantiated configure info item if everything works out fine,
 callers are responsible to free the memory returned by this function */
static meta_cfg_info_t *innodb_config_add_item(
    /*===================*/
    ib_tpl_t tpl,                /*!< in: container row we are fetching
                                 row from */
    hash_table_t *eng_meta_hash, /*!< in/out: hash table to insert
                                 the row */
    void *thd)                   /*!< in/out: MySQL THD */
{
  ib_err_t err = DB_SUCCESS;
  int n_cols;
  int i;
  ib_ulint_t data_len;
  meta_cfg_info_t *item = NULL;
  ib_col_meta_t col_meta;
  int fold;

  n_cols = innodb_cb_tuple_get_n_cols(tpl);

  if (n_cols < CONTAINER_NUM_COLS) {
    fprintf(stderr,
            " InnoDB_Memcached: config table '%s' in"
            " database '%s' has only %d column(s),"
            " server is expecting %d columns\n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME, n_cols,
            CONTAINER_NUM_COLS);
    err = DB_ERROR;
    goto func_exit;
  }

  item = (meta_cfg_info_t *)malloc(sizeof(*item));

  memset(item, 0, sizeof(*item));

  /* Get the column mappings (column for each memcached data */
  for (i = 0; i < CONTAINER_NUM_COLS; ++i) {
    data_len = innodb_cb_col_get_meta(tpl, i, &col_meta);

    if (data_len == IB_SQL_NULL) {
      fprintf(stderr,
              " InnoDB_Memcached: column %d in"
              " the entry for config table '%s' in"
              " database '%s' has an invalid"
              " NULL value\n",
              i, MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME);

      err = DB_ERROR;
      goto func_exit;
    }

    item->col_info[i].col_name_len = data_len;

    item->col_info[i].col_name =
        my_strdupl((char *)innodb_cb_col_get_value(tpl, i), data_len);

    item->col_info[i].field_id = -1;

    if (i == CONTAINER_VALUE) {
      innodb_config_parse_value_col(item, item->col_info[i].col_name, data_len);
    }
  }

  /* Last column is about the unique index name on key column */
  data_len = innodb_cb_col_get_meta(tpl, i, &col_meta);

  if (data_len == IB_SQL_NULL) {
    fprintf(stderr,
            " InnoDB_Memcached: There must be a unique"
            " index on memcached table's key column\n");
    err = DB_ERROR;
    goto func_exit;
  }

  item->index_info.idx_name =
      my_strdupl((char *)innodb_cb_col_get_value(tpl, i), data_len);

  if (!innodb_verify(item, thd)) {
    err = DB_ERROR;
    goto func_exit;
  }

  fold = ut_fold_string(item->col_info[0].col_name);
  HASH_INSERT(meta_cfg_info_t, name_hash, eng_meta_hash, fold, item);

func_exit:
  if (err != DB_SUCCESS && item) {
    free(item);
    item = NULL;
  }

  return (item);
}

/**********************************************************************/ /**
 This function opens the "containers" table, reads in all rows and
 instantiates the metadata hash table.
 @return the default configuration setting (whose mapping name is "default") */
meta_cfg_info_t *innodb_config_meta_hash_init(
    /*=========================*/
    hash_table_t *meta_hash, /*!< in/out: InnoDB Memcached engine */
    void *thd)               /*!< in/out: MySQL THD */
{
  ib_trx_t ib_trx;
  ib_crsr_t crsr = NULL;
  ib_crsr_t idx_crsr = NULL;
  ib_tpl_t tpl = NULL;
  ib_err_t err = DB_SUCCESS;
  meta_cfg_info_t *default_item = NULL;

  ib_trx = ib_cb_trx_begin(IB_TRX_READ_COMMITTED, true, false, thd);
  err = innodb_api_begin(NULL, MCI_CFG_DB_NAME, MCI_CFG_CONTAINER_TABLE, NULL,
                         ib_trx, &crsr, &idx_crsr, IB_LOCK_S);

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: Please create config table"
            "'%s' in database '%s' by running"
            " 'innodb_memcached_config.sql. error %s'\n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME, ib_cb_ut_strerr(err));
    err = DB_ERROR;
    goto func_exit;
  }

  tpl = innodb_cb_read_tuple_create(crsr);

  /* If name field is NULL, just read the first row */
  err = innodb_cb_cursor_first(crsr);

  while (err == DB_SUCCESS) {
    meta_cfg_info_t *item;

    err = ib_cb_cursor_read_row(crsr, tpl, NULL, 0, NULL, NULL, NULL);

    if (err != DB_SUCCESS) {
      fprintf(stderr,
              " InnoDB_Memcached: failed to read row"
              " from config table '%s' in database"
              " '%s' \n",
              MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME);
      err = DB_ERROR;
      goto func_exit;
    }

    item = innodb_config_add_item(tpl, meta_hash, thd);

    /* First initialize default setting to be the first row
    of the table */
    /* If there are any setting whose name is "default",
    then set default_item to point to this setting, otherwise
    point it to the first row of the table */
    if (default_item == NULL ||
        (item && strcmp(item->col_info[0].col_name, "default") == 0)) {
      default_item = item;
    }

    err = ib_cb_cursor_next(crsr);
  }

  if (err == DB_END_OF_INDEX) {
    err = DB_SUCCESS;
  }

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate entry in"
            " config table '%s' in database '%s' \n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME);
    err = DB_ERROR;
  }

func_exit:

  innodb_cb_cursor_close(crsr);

  if (tpl) {
    innodb_cb_tuple_delete(tpl);
  }

  innodb_cb_trx_commit(ib_trx);
  ib_cb_trx_release(ib_trx);

  return (default_item);
}

/**********************************************************************/ /**
 This function opens the "containers" configuration table, and find the
 table and column info that used for memcached data
 @return true if everything works out fine */
static meta_cfg_info_t *innodb_config_container(
    /*====================*/
    const char *name,        /*!< in: option name to look for */
    size_t name_len,         /*!< in: option name length */
    hash_table_t *meta_hash, /*!< in: engine hash table */
    void *thd)               /*!< in/out: MySQL THD */
{
  ib_trx_t ib_trx;
  ib_crsr_t crsr = NULL;
  ib_crsr_t idx_crsr = NULL;
  ib_tpl_t tpl = NULL;
  ib_err_t err = DB_SUCCESS;
  int n_cols;
  int i;
  ib_ulint_t data_len;
  ib_col_meta_t col_meta;
  ib_tpl_t read_tpl = NULL;
  meta_cfg_info_t *item = NULL;

  if (name != NULL) {
    ib_ulint_t fold;

    assert(meta_hash);

    fold = ut_fold_string(name);
    HASH_SEARCH(name_hash, meta_hash, fold, meta_cfg_info_t *, item,
                (name_len == item->col_info[0].col_name_len &&
                 strcmp(name, item->col_info[0].col_name) == 0));

    if (item) {
      return (item);
    }
  }

  ib_trx = ib_cb_trx_begin(IB_TRX_READ_COMMITTED, true, false, thd);
  err = innodb_api_begin(NULL, MCI_CFG_DB_NAME, MCI_CFG_CONTAINER_TABLE, NULL,
                         ib_trx, &crsr, &idx_crsr, IB_LOCK_S);

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: Please create config table"
            "'%s' in database '%s' by running"
            " 'innodb_memcached_config.sql. error %d'\n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME, err);
    err = DB_ERROR;
    goto func_exit;
  }

  if (!name) {
    tpl = innodb_cb_read_tuple_create(crsr);

    /* If name field is NULL, just read the first row */
    err = innodb_cb_cursor_first(crsr);
  } else {
    /* User supplied a config option name, find it */
    tpl = ib_cb_sec_search_tuple_create(crsr);

    err = ib_cb_col_set_value(tpl, 0, name, name_len, true);

    ib_cb_cursor_set_match_mode(crsr, IB_EXACT_MATCH);
    err = ib_cb_cursor_moveto(crsr, tpl, IB_CUR_GE, 0);
  }

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate entry in"
            " config table '%s' in database '%s' \n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME);
    err = DB_ERROR;
    goto func_exit;
  }

  if (!name) {
    read_tpl = tpl;
    err = ib_cb_cursor_read_row(crsr, tpl, NULL, 0, NULL, NULL, NULL);
  } else {
    read_tpl = ib_cb_clust_read_tuple_create(crsr);

    err = ib_cb_cursor_read_row(crsr, read_tpl, NULL, 0, NULL, NULL, NULL);
  }

  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to read row from"
            " config table '%s' in database '%s' \n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME);
    err = DB_ERROR;
    goto func_exit;
  }

  n_cols = innodb_cb_tuple_get_n_cols(read_tpl);

  if (n_cols < CONTAINER_NUM_COLS) {
    fprintf(stderr,
            " InnoDB_Memcached: config table '%s' in"
            " database '%s' has only %d column(s),"
            " server is expecting %d columns\n",
            MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME, n_cols,
            CONTAINER_NUM_COLS);
    err = DB_ERROR;
    goto func_exit;
  }

  item = (meta_cfg_info_t *)malloc(sizeof(*item));
  memset(item, 0, sizeof(*item));

  /* Get the column mappings (column for each memcached data */
  for (i = 0; i < CONTAINER_NUM_COLS; ++i) {
    data_len = innodb_cb_col_get_meta(read_tpl, i, &col_meta);

    if (data_len == IB_SQL_NULL) {
      fprintf(stderr,
              " InnoDB_Memcached: column %d in"
              " the entry for config table '%s' in"
              " database '%s' has an invalid"
              " NULL value\n",
              i, MCI_CFG_CONTAINER_TABLE, MCI_CFG_DB_NAME);

      err = DB_ERROR;
      goto func_exit;
    }

    item->col_info[i].col_name_len = data_len;

    item->col_info[i].col_name =
        my_strdupl((char *)innodb_cb_col_get_value(read_tpl, i), data_len);

    item->col_info[i].field_id = -1;

    if (i == CONTAINER_VALUE) {
      innodb_config_parse_value_col(item, item->col_info[i].col_name, data_len);
    }
  }

  /* Last column is about the unique index name on key column */
  data_len = innodb_cb_col_get_meta(read_tpl, i, &col_meta);

  if (data_len == IB_SQL_NULL) {
    fprintf(stderr,
            " InnoDB_Memcached: There must be a unique"
            " index on memcached table's key column\n");
    err = DB_ERROR;
    goto func_exit;
  }

  item->index_info.idx_name =
      my_strdupl((char *)innodb_cb_col_get_value(read_tpl, i), data_len);

  if (!innodb_verify(item, thd)) {
    err = DB_ERROR;
  }

func_exit:

  innodb_cb_cursor_close(crsr);

  if (tpl) {
    innodb_cb_tuple_delete(tpl);
  }

  innodb_cb_trx_commit(ib_trx);
  ib_cb_trx_release(ib_trx);

  if (err != DB_SUCCESS) {
    free(item);
    item = NULL;
  } else {
    ib_ulint_t fold;

    fold = ut_fold_string(item->col_info[0].col_name);
    HASH_INSERT(meta_cfg_info_t, name_hash, meta_hash, fold, item);
  }

  return (item);
}

/**********************************************************************/ /**
 This function verifies "value" column(s) specified by configure table are of
 the correct type
 @return DB_SUCCESS if everything is verified */
static ib_err_t innodb_config_value_col_verify(
    /*===========================*/
    const char *name,           /*!< in: column name */
    meta_cfg_info_t *meta_info, /*!< in: meta info structure */
    ib_col_meta_t *col_meta,    /*!< in: column metadata */
    int col_id,                 /*!< in: column ID */
    meta_column_t *col_verify)  /*!< in: verify structure */
{
  ib_err_t err = DB_NOT_FOUND;
  char table_name[MAX_TABLE_NAME_LEN + MAX_DATABASE_NAME_LEN];
  char *dbname;
  char *tname;

  /* Get table name. */
  dbname = meta_info->col_info[CONTAINER_DB].col_name;
  tname = meta_info->col_info[CONTAINER_TABLE].col_name;
#ifdef __WIN__
  sprintf(table_name, "%s\%s", dbname, tname);
#else
  snprintf(table_name, sizeof(table_name), "%s/%s", dbname, tname);
#endif

  if (!meta_info->n_extra_col) {
    meta_column_t *cinfo = meta_info->col_info;

    /* "value" column must be of CHAR, VARCHAR or BLOB type */
    if (strcmp(name, cinfo[CONTAINER_VALUE].col_name) == 0) {
      if (col_meta->type != IB_VARCHAR && col_meta->type != IB_CHAR &&
          col_meta->type != IB_BLOB && col_meta->type != IB_CHAR_ANYCHARSET &&
          col_meta->type != IB_VARCHAR_ANYCHARSET && col_meta->type != IB_INT) {
        fprintf(stderr,
                " InnoDB_Memcached: the value"
                " column %s in table %s"
                " should be INTEGER, CHAR or"
                " VARCHAR.\n",
                name, table_name);
        err = DB_DATA_MISMATCH;
      }

      cinfo[CONTAINER_VALUE].field_id = col_id;
      cinfo[CONTAINER_VALUE].col_meta = *col_meta;
      err = DB_SUCCESS;
    }
  } else {
    int i;

    for (i = 0; i < meta_info->n_extra_col; i++) {
      if (strcmp(name, meta_info->extra_col_info[i].col_name) == 0) {
        if (col_meta->type != IB_VARCHAR && col_meta->type != IB_CHAR &&
            col_meta->type != IB_BLOB && col_meta->type != IB_CHAR_ANYCHARSET &&
            col_meta->type != IB_VARCHAR_ANYCHARSET &&
            col_meta->type != IB_INT) {
          fprintf(stderr,
                  " InnoDB_Memcached: the value"
                  " column %s in table %s"
                  " should be INTEGER, CHAR or"
                  " VARCHAR.\n",
                  name, table_name);
          err = DB_DATA_MISMATCH;
          break;
        }

        meta_info->extra_col_info[i].field_id = col_id;
        meta_info->extra_col_info[i].col_meta = *col_meta;

        meta_info->col_info[CONTAINER_VALUE].field_id = col_id;
        meta_info->col_info[CONTAINER_VALUE].col_meta = *col_meta;

        if (col_verify) {
          col_verify[i].field_id = col_id;
        }

        err = DB_SUCCESS;
      }
    }
  }

  return (err);
}

/**********************************************************************/ /**
 This function verifies the table configuration information on an opened
 table, and fills in columns used for memcached functionalities (cas, exp etc.)
 @return true if everything works out fine */
ib_err_t innodb_verify_low(
    /*==============*/
    meta_cfg_info_t *info, /*!< in/out: meta info structure */
    ib_crsr_t crsr,        /*!< in: crsr */
    bool runtime)          /*!< in: verify at the runtime */
{
  ib_crsr_t idx_crsr = NULL;
  ib_tpl_t tpl = NULL;
  ib_col_meta_t col_meta;
  int n_cols;
  int i;
  bool is_key_col = false;
  bool is_value_col = false;
  bool is_flag_col = false;
  bool is_cas_col = false;
  bool is_exp_col = false;
  int index_type;
  ib_id_u64_t index_id;
  ib_err_t err = DB_SUCCESS;
  const char *name;
  meta_column_t *cinfo = info->col_info;
  meta_column_t *col_verify = NULL;
  char table_name[MAX_TABLE_NAME_LEN + MAX_DATABASE_NAME_LEN];
  char *dbname;
  char *tname;

  tpl = innodb_cb_read_tuple_create(crsr);

  if (runtime && info->n_extra_col) {
    col_verify =
        (meta_column_t *)malloc(info->n_extra_col * sizeof(meta_column_t));

    if (!col_verify) {
      return (DB_ERROR);
    }

    for (i = 0; i < info->n_extra_col; i++) {
      col_verify[i].field_id = -1;
    }
  }

  /* Get table name. */
  dbname = info->col_info[CONTAINER_DB].col_name;
  tname = info->col_info[CONTAINER_TABLE].col_name;
#ifdef __WIN__
  sprintf(table_name, "%s\%s", dbname, tname);
#else
  snprintf(table_name, sizeof(table_name), "%s/%s", dbname, tname);
#endif

  n_cols = innodb_cb_tuple_get_n_cols(tpl);

  /* Verify each mapped column */
  for (i = 0; i < n_cols; i++) {
    ib_err_t result = DB_SUCCESS;

    name = innodb_cb_col_get_name(crsr, i);
    innodb_cb_col_get_meta(tpl, i, &col_meta);

    result =
        innodb_config_value_col_verify(name, info, &col_meta, i, col_verify);

    if (result == DB_SUCCESS) {
      is_value_col = true;

      if (strcmp(name, cinfo[CONTAINER_KEY].col_name)) {
        continue;
      }
    } else if (result == DB_DATA_MISMATCH) {
      err = DB_DATA_MISMATCH;
      goto func_exit;
    }

    if (strcmp(name, cinfo[CONTAINER_KEY].col_name) == 0) {
      /* Key column must be CHAR or VARCHAR type */
      if (col_meta.type != IB_VARCHAR && col_meta.type != IB_CHAR &&
          col_meta.type != IB_VARCHAR_ANYCHARSET &&
          col_meta.type != IB_CHAR_ANYCHARSET && col_meta.type != IB_INT) {
        fprintf(stderr,
                " InnoDB_Memcached: the key"
                " column %s in table %s should"
                " be INTEGER, CHAR or VARCHAR.\n",
                name, table_name);
        err = DB_DATA_MISMATCH;
        goto func_exit;
      }
      cinfo[CONTAINER_KEY].field_id = i;
      cinfo[CONTAINER_KEY].col_meta = col_meta;
      is_key_col = true;
    } else if (strcmp(name, cinfo[CONTAINER_FLAG].col_name) == 0) {
      /* Flag column must be integer type */
      if (col_meta.type != IB_INT) {
        fprintf(stderr,
                " InnoDB_Memcached: the flag"
                " column %s in table %s should"
                " be INTEGER.\n",
                name, table_name);
        err = DB_DATA_MISMATCH;
        goto func_exit;
      }
      cinfo[CONTAINER_FLAG].field_id = i;
      cinfo[CONTAINER_FLAG].col_meta = col_meta;
      info->flag_enabled = true;
      is_flag_col = true;
    } else if (strcmp(name, cinfo[CONTAINER_CAS].col_name) == 0) {
      /* CAS column must be integer type */
      if (col_meta.type != IB_INT) {
        fprintf(stderr,
                " InnoDB_Memcached: the cas"
                " column %s in table %s should"
                " be INTEGER.\n",
                name, table_name);
        err = DB_DATA_MISMATCH;
        goto func_exit;
      }
      cinfo[CONTAINER_CAS].field_id = i;
      cinfo[CONTAINER_CAS].col_meta = col_meta;
      info->cas_enabled = true;
      is_cas_col = true;
    } else if (strcmp(name, cinfo[CONTAINER_EXP].col_name) == 0) {
      /* EXP column must be integer type */
      if (col_meta.type != IB_INT) {
        fprintf(stderr,
                " InnoDB_Memcached: the expire"
                " column %s in table %s should"
                " be INTEGER.\n",
                name, table_name);
        err = DB_DATA_MISMATCH;
        goto func_exit;
      }
      cinfo[CONTAINER_EXP].field_id = i;
      cinfo[CONTAINER_EXP].col_meta = col_meta;
      info->exp_enabled = true;
      is_exp_col = true;
    }
  }

  /* Key column and Value column must present */
  if (!is_key_col || !is_value_col) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate key"
            " column or value column in table"
            " as specified by config table \n");

    err = DB_ERROR;
    goto func_exit;
  }

  if (info->n_extra_col) {
    meta_column_t *col_check;

    col_check = (runtime && col_verify) ? col_verify : info->extra_col_info;

    for (i = 0; i < info->n_extra_col; i++) {
      if (col_check[i].field_id < 0) {
        fprintf(stderr,
                " InnoDB_Memcached: failed to"
                " locate value column %s"
                " as specified by config"
                " table \n",
                info->extra_col_info[i].col_name);
        err = DB_ERROR;
        goto func_exit;
      }
    }
  }

  if (info->flag_enabled && !is_flag_col) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate flag"
            " column as specified by config table \n");
    err = DB_ERROR;
    goto func_exit;
  }

  if (info->cas_enabled && !is_cas_col) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate cas"
            " column as specified by config table \n");
    err = DB_ERROR;
    goto func_exit;
  }

  if (info->exp_enabled && !is_exp_col) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to locate exp"
            " column as specified by config table \n");
    err = DB_ERROR;
    goto func_exit;
  }

  /* Test the specified index */
  innodb_cb_cursor_open_index_using_name(crsr, info->index_info.idx_name,
                                         &idx_crsr, &index_type, &index_id);

  if (index_type & IB_CLUSTERED) {
    info->index_info.srch_use_idx = META_USE_CLUSTER;
  } else if (!idx_crsr || !(index_type & IB_UNIQUE)) {
    fprintf(stderr,
            " InnoDB_Memcached: Index on key column"
            " must be a Unique index\n");
    info->index_info.srch_use_idx = META_USE_NO_INDEX;
    err = DB_ERROR;
  } else {
    info->index_info.idx_id = index_id;
    info->index_info.srch_use_idx = META_USE_SECONDARY;
  }

  if (idx_crsr) {
    /* We use the same function even if index_type & IB_CLUSTERED,
    because the variant for _clust_ is not exposed via API, and it turns
    out that we don't really need it, as long as the clustered index
    has one column. */
    ib_tpl_t idx_tpl = ib_cb_sec_search_tuple_create(idx_crsr);

    n_cols = ib_cb_tuple_get_n_user_cols(idx_tpl);

    name = ib_cb_get_idx_field_name(idx_crsr, 0);

    if (n_cols > 1) {
      fprintf(stderr,
              " InnoDB_Memcached: The unique_idx_name_on_key (%s)"
              " must be on key column (%s) only but it is on %d columns\n",
              info->index_info.idx_name, cinfo[CONTAINER_KEY].col_name, n_cols);
      info->index_info.srch_use_idx = META_USE_NO_INDEX;
      err = DB_ERROR;
    }
    if (strcmp(name, cinfo[CONTAINER_KEY].col_name)) {
      fprintf(stderr,
              " InnoDB_Memcached: The unique_idx_name_on_key (%s)"
              " must be on key column (%s) but it is on (%s)\n",
              info->index_info.idx_name, cinfo[CONTAINER_KEY].col_name, name);
      info->index_info.srch_use_idx = META_USE_NO_INDEX;
      err = DB_ERROR;
    }

    innodb_cb_tuple_delete(idx_tpl);
    innodb_cb_cursor_close(idx_crsr);
  }
func_exit:

  if (runtime && col_verify) {
    free(col_verify);
  }

  if (tpl) {
    innodb_cb_tuple_delete(tpl);
  }

  return (err);
}

/** This function verifies the table configuration information, and fills
in columns used for memcached functionalities (cas, exp etc.)
@param[in]	info	meta info structure
@param[in,out]	thd	MySQL THD
@return true if everything works out fine */
bool innodb_verify(
    /*==========*/
    meta_cfg_info_t *info, void *thd) {
  ib_crsr_t crsr = NULL;
  char table_name[MAX_TABLE_NAME_LEN + MAX_DATABASE_NAME_LEN];
  char *dbname;
  char *name;
  ib_err_t err = DB_SUCCESS;
  ib_trx_t ib_trx = ib_cb_trx_begin(IB_TRX_READ_COMMITTED, false, false, thd);

  dbname = info->col_info[CONTAINER_DB].col_name;
  name = info->col_info[CONTAINER_TABLE].col_name;
  info->flag_enabled = false;
  info->cas_enabled = false;
  info->exp_enabled = false;

#ifdef _WIN32
  sprintf(table_name, "%s\%s", dbname, name);
#else
  snprintf(table_name, sizeof(table_name), "%s/%s", dbname, name);
#endif

  err = innodb_cb_open_table(table_name, ib_trx, &crsr);

  /* Mapped InnoDB table must be able to open */
  if (err != DB_SUCCESS) {
    fprintf(stderr,
            " InnoDB_Memcached: failed to open table"
            " '%s' \n",
            table_name);
    err = DB_ERROR;
    goto func_exit;
  }

  if (ib_cb_is_virtual_table(crsr)) {
    fprintf(stderr,
            " InnoDB_Memcached: table '%s' cannot"
            " be mapped since it contains virtual"
            " columns. \n",
            table_name);
    err = DB_ERROR;
    goto func_exit;
  }

  err = innodb_verify_low(info, crsr, false);
func_exit:
  innodb_cb_cursor_close(crsr);

  innodb_cb_trx_commit(ib_trx);
  ib_cb_trx_release(ib_trx);

  return (err == DB_SUCCESS);
}

/**********************************************************************/ /**
 If the hash table (meta_hash) is NULL, then initialise the hash table with
 data in the configure tables. And return the "default" item.  If there is
 no setting named "default" then use the first row in the table. This is
 currently only used at the engine initialization time.
 If the hash table (meta_hash) is created, then look for the meta-data based on
 specified configuration name parameter. If such metadata does not exist in
 the hash table, then add such metadata into hash table.
 @return meta_cfg_info_t* structure if configure option found, otherwise NULL */
meta_cfg_info_t *innodb_config(
    /*==========*/
    const char *name,         /*!< in: config option name */
    size_t name_len,          /*!< in: name length */
    hash_table_t **meta_hash) /*!< in/out: engine hash
                              table. If NULL, it will be
                              created and initialized */
{
  meta_cfg_info_t *item;
  bool success;
  void *thd = handler_create_thd(false);

  if (*meta_hash == NULL) {
    *meta_hash = hash_create(100);
  }

  if (!name) {
    item = innodb_config_meta_hash_init(*meta_hash, thd);
  } else {
    ib_ulint_t fold;

    fold = ut_fold_string(name);
    HASH_SEARCH(name_hash, *meta_hash, fold, meta_cfg_info_t *, item,
                (name_len == item->col_info[0].col_name_len &&
                 strcmp(name, item->col_info[0].col_name) == 0));

    if (item) {
      handler_close_thd(thd);
      return (item);
    }

    item = innodb_config_container(name, name_len, *meta_hash, thd);
  }

  if (!item) {
    handler_close_thd(thd);
    return (NULL);
  }

  /* Following two configure operations are optional, and can be
  failed */
  success = innodb_read_cache_policy(item, thd);

  if (!success) {
    handler_close_thd(thd);
    return (NULL);
  }

  success = innodb_read_config_option(item, thd);

  handler_close_thd(thd);

  if (!success) {
    return (NULL);
  }

  return (item);
}