percona-server-5.6.51-91.0/sql/ha_ndbinfo.cc

/*
   Copyright (C) 2009 Sun Microsystems Inc.
   All rights reserved. Use is subject to license terms.

   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
*/

#include "ha_ndbcluster_glue.h"

#ifdef WITH_NDBCLUSTER_STORAGE_ENGINE
#include "ha_ndbinfo.h"
#include "../storage/ndb/src/ndbapi/NdbInfo.hpp"


static MYSQL_THDVAR_UINT(
  max_rows,                          /* name */
  PLUGIN_VAR_RQCMDARG,
  "Specify max number of rows to fetch per roundtrip to cluster",
  NULL,                              /* check func. */
  NULL,                              /* update func. */
  10,                                /* default */
  1,                                 /* min */
  256,                               /* max */
  0                                  /* block */
);

static MYSQL_THDVAR_UINT(
  max_bytes,                         /* name */
  PLUGIN_VAR_RQCMDARG,
  "Specify approx. max number of bytes to fetch per roundtrip to cluster",
  NULL,                              /* check func. */
  NULL,                              /* update func. */
  0,                                 /* default */
  0,                                 /* min */
  65535,                             /* max */
  0                                  /* block */
);

static MYSQL_THDVAR_BOOL(
  show_hidden,                       /* name */
  PLUGIN_VAR_RQCMDARG,
  "Control if tables should be visible or not",
  NULL,                              /* check func. */
  NULL,                              /* update func. */
  FALSE                              /* default */
);

static char* opt_ndbinfo_dbname = (char*)"ndbinfo";
static MYSQL_SYSVAR_STR(
  database,                         /* name */
  opt_ndbinfo_dbname,               /* var */
  PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
  "Name of the database used by ndbinfo",
  NULL,                             /* check func. */
  NULL,                             /* update func. */
  NULL                              /* default */
);

static char* opt_ndbinfo_table_prefix = (char*)"ndb$";
static MYSQL_SYSVAR_STR(
  table_prefix,                     /* name */
  opt_ndbinfo_table_prefix,         /* var */
  PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
  "Prefix to use for all virtual tables loaded from NDB",
  NULL,                             /* check func. */
  NULL,                             /* update func. */
  NULL                              /* default */
);

static Uint32 opt_ndbinfo_version = NDB_VERSION_D;
static MYSQL_SYSVAR_UINT(
  version,                          /* name */
  opt_ndbinfo_version,              /* var */
  PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_READONLY,
  "Compile version for ndbinfo",
  NULL,                             /* check func. */
  NULL,                             /* update func. */
  0,                                /* default */
  0,                                /* min */
  0,                                /* max */
  0                                 /* block */
);

static my_bool opt_ndbinfo_offline;

static
void
offline_update(THD* thd, struct st_mysql_sys_var* var,
               void* var_ptr, const void* save)
{
  DBUG_ENTER("offline_update");

  const my_bool new_offline =
    (*(static_cast<const my_bool*>(save)) != 0);
  if (new_offline == opt_ndbinfo_offline)
  {
    // No change
    DBUG_VOID_RETURN;
  }

  // Set offline mode, any tables opened from here on will
  // be opened in the new mode
  opt_ndbinfo_offline = new_offline;

  // Close any open tables which may be in the old mode
  (void)close_cached_tables(thd, NULL, false, true, false);

  DBUG_VOID_RETURN;
}

static MYSQL_SYSVAR_BOOL(
  offline,                          /* name */
  opt_ndbinfo_offline,              /* var */
  PLUGIN_VAR_NOCMDOPT,
  "Set ndbinfo in offline mode, tables and views can "
  "be opened even if they don't exist or have different "
  "definition in NDB. No rows will be returned.",
  NULL,                             /* check func. */
  offline_update,                   /* update func. */
  0                                 /* default */
);


static NdbInfo* g_ndbinfo;

extern Ndb_cluster_connection* g_ndb_cluster_connection;

static bool
ndbcluster_is_disabled(void)
{
  /*
    ndbinfo uses the same connection as ndbcluster
    to avoid using up another nodeid, this also means that
    if ndbcluster is not enabled, ndbinfo won't start
  */
  if (g_ndb_cluster_connection)
    return false;
  assert(g_ndbinfo == NULL);
  return true;
}

static handler*
create_handler(handlerton *hton, TABLE_SHARE *table, MEM_ROOT *mem_root)
{
  return new (mem_root) ha_ndbinfo(hton, table);
}

struct ha_ndbinfo_impl
{
  const NdbInfo::Table* m_table;
  NdbInfoScanOperation* m_scan_op;
  Vector<const NdbInfoRecAttr *> m_columns;
  bool m_first_use;

  // Indicates if table has been opened in offline mode
  // can only be reset by closing the table
  bool m_offline;

  ha_ndbinfo_impl() :
    m_table(NULL),
    m_scan_op(NULL),
    m_first_use(true),
    m_offline(false)
  {
  }
};

ha_ndbinfo::ha_ndbinfo(handlerton *hton, TABLE_SHARE *table_arg)
: handler(hton, table_arg), m_impl(*new ha_ndbinfo_impl)
{
}

ha_ndbinfo::~ha_ndbinfo()
{
  delete &m_impl;
}

enum ndbinfo_error_codes {
  ERR_INCOMPAT_TABLE_DEF = 40001
};

struct error_message {
  int error;
  const char* message;
} error_messages[] = {
  { ERR_INCOMPAT_TABLE_DEF, "Incompatible table definitions" },
  { HA_ERR_NO_CONNECTION, "Connection to NDB failed" },

  { 0, 0 }
};

static
const char* find_error_message(int error)
{
  struct error_message* err = error_messages;
  while (err->error && err->message)
  {
    if (err->error == error)
    {
      assert(err->message);
      return err->message;
    }
    err++;
  }
  return NULL;
}

static int err2mysql(int error)
{
  DBUG_ENTER("err2mysql");
  DBUG_PRINT("enter", ("error: %d", error));
  assert(error != 0);
  switch(error)
  {
  case NdbInfo::ERR_ClusterFailure:
    DBUG_RETURN(HA_ERR_NO_CONNECTION);
    break;
  case NdbInfo::ERR_OutOfMemory:
    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
    break;
  default:
    break;
  }
  push_warning_printf(current_thd, Sql_condition::WARN_LEVEL_WARN,
                      ER_GET_ERRNO, ER(ER_GET_ERRNO), error);
  DBUG_RETURN(HA_ERR_INTERNAL_ERROR);
}

bool ha_ndbinfo::get_error_message(int error, String *buf)
{
  DBUG_ENTER("ha_ndbinfo::get_error_message");
  DBUG_PRINT("enter", ("error: %d", error));

  const char* message = find_error_message(error);
  if (!message)
    DBUG_RETURN(false);

  buf->set(message, strlen(message), &my_charset_bin);
  DBUG_PRINT("exit", ("message: %s", buf->ptr()));
  DBUG_RETURN(false);
}

static void
generate_sql(const NdbInfo::Table* ndb_tab, BaseString& sql)
{
  sql.appfmt("'CREATE TABLE `%s`.`%s%s` (",
             opt_ndbinfo_dbname, opt_ndbinfo_table_prefix, ndb_tab->getName());

  const char* separator = "";
  for (unsigned i = 0; i < ndb_tab->columns(); i++)
  {
    const NdbInfo::Column* col = ndb_tab->getColumn(i);

    sql.appfmt("%s", separator);
    separator = ", ";

    sql.appfmt("`%s` ", col->m_name.c_str());

    switch(col->m_type)
    {
    case NdbInfo::Column::Number:
      sql.appfmt("INT UNSIGNED");
      break;
    case NdbInfo::Column::Number64:
      sql.appfmt("BIGINT UNSIGNED");
      break;
    case NdbInfo::Column::String:
      sql.appfmt("VARCHAR(512)");
      break;
    default:
      sql.appfmt("UNKNOWN");
      assert(false);
      break;
    }
  }
  sql.appfmt(") ENGINE=NDBINFO'");
}

/*
  Push a warning with explanation of the problem as well as the
  proper SQL so the user can regenerate the table definition
*/

static void
warn_incompatible(const NdbInfo::Table* ndb_tab, bool fatal,
             const char* format, ...)
{
  BaseString msg;
  DBUG_ENTER("warn_incompatible");
  DBUG_PRINT("enter",("table_name: %s, fatal: %d", ndb_tab->getName(), fatal));
  DBUG_ASSERT(format != NULL);

  va_list args;
  char explanation[128];
  va_start(args,format);
  my_vsnprintf(explanation, sizeof(explanation), format, args);
  va_end(args);

  msg.assfmt("Table '%s%s' is defined differently in NDB, %s. The "
             "SQL to regenerate is: ",
             opt_ndbinfo_table_prefix, ndb_tab->getName(), explanation);
  generate_sql(ndb_tab, msg);

  const Sql_condition::enum_warning_level level =
    (fatal ? Sql_condition::WARN_LEVEL_WARN : Sql_condition::WARN_LEVEL_NOTE);
  push_warning(current_thd, level, ERR_INCOMPAT_TABLE_DEF, msg.c_str());

  DBUG_VOID_RETURN;
}

int ha_ndbinfo::create(const char *name, TABLE *form,
                       HA_CREATE_INFO *create_info)
{
  DBUG_ENTER("ha_ndbinfo::create");
  DBUG_PRINT("enter", ("name: %s", name));

  DBUG_RETURN(0);
}

bool ha_ndbinfo::is_open(void) const
{
  return m_impl.m_table != NULL;
}

bool ha_ndbinfo::is_offline(void) const
{
  return m_impl.m_offline;
}

int ha_ndbinfo::open(const char *name, int mode, uint test_if_locked)
{
  DBUG_ENTER("ha_ndbinfo::open");
  DBUG_PRINT("enter", ("name: %s, mode: %d", name, mode));

  assert(is_closed());
  assert(!is_offline()); // Closed table can not be offline

  if (mode == O_RDWR)
  {
    if (table->db_stat & HA_TRY_READ_ONLY)
    {
      DBUG_PRINT("info", ("Telling server to use readonly mode"));
      DBUG_RETURN(EROFS); // Read only fs
    }
    // Find any commands that does not allow open readonly
    DBUG_ASSERT(false);
  }

  if (opt_ndbinfo_offline ||
      ndbcluster_is_disabled())
  {
    // Mark table as being offline and allow it to be opened
    m_impl.m_offline = true;
    DBUG_RETURN(0);
  }

  int err = g_ndbinfo->openTable(name, &m_impl.m_table);
  if (err)
  {
    assert(m_impl.m_table == 0);
    if (err == NdbInfo::ERR_NoSuchTable)
      DBUG_RETURN(HA_ERR_NO_SUCH_TABLE);
    DBUG_RETURN(err2mysql(err));
  }

  /*
    Check table def. to detect incompatible differences which should
    return an error. Differences which only generate a warning
    is checked on first use
  */
  DBUG_PRINT("info", ("Comparing MySQL's table def against NDB"));
  const NdbInfo::Table* ndb_tab = m_impl.m_table;
  for (uint i = 0; i < table->s->fields; i++)
  {
    const Field* field = table->field[i];

    // Check if field is NULLable
    if (const_cast<Field*>(field)->real_maybe_null() == false)
    {
      // Only NULLable fields supported
      warn_incompatible(ndb_tab, true,
                        "column '%s' is NOT NULL",
                        field->field_name);
      delete m_impl.m_table; m_impl.m_table= 0;
      DBUG_RETURN(ERR_INCOMPAT_TABLE_DEF);
    }

    // Check if column exist in NDB
    const NdbInfo::Column* col = ndb_tab->getColumn(field->field_name);
    if (!col)
    {
      // The column didn't exist
      continue;
    }

    // Check compatible field and column type
    bool compatible = false;
    switch(col->m_type)
    {
    case NdbInfo::Column::Number:
      if (field->type() == MYSQL_TYPE_LONG)
        compatible = true;
      break;
    case NdbInfo::Column::Number64:
      if (field->type() == MYSQL_TYPE_LONGLONG)
        compatible = true;
      break;
    case NdbInfo::Column::String:
      if (field->type() == MYSQL_TYPE_VARCHAR)
        compatible = true;
      break;
    default:
      assert(false);
      break;
    }
    if (!compatible)
    {
      // The column type is not compatible
      warn_incompatible(ndb_tab, true,
                        "column '%s' is not compatible",
                        field->field_name);
      delete m_impl.m_table; m_impl.m_table= 0;
      DBUG_RETURN(ERR_INCOMPAT_TABLE_DEF);
    }
  }

  /* Increase "ref_length" to allow a whole row to be stored in "ref" */
  ref_length = 0;
  for (uint i = 0; i < table->s->fields; i++)
    ref_length += table->field[i]->pack_length();
  DBUG_PRINT("info", ("ref_length: %u", ref_length));

  DBUG_RETURN(0);
}

int ha_ndbinfo::close(void)
{
  DBUG_ENTER("ha_ndbinfo::close");

  if (is_offline())
    DBUG_RETURN(0);

  assert(is_open());
  if (m_impl.m_table)
  {
    g_ndbinfo->closeTable(m_impl.m_table);
    m_impl.m_table = NULL;
  }
  DBUG_RETURN(0);
}

int ha_ndbinfo::rnd_init(bool scan)
{
  DBUG_ENTER("ha_ndbinfo::rnd_init");
  DBUG_PRINT("info", ("scan: %d", scan));

  if (is_offline())
  {
    push_warning(current_thd, Sql_condition::WARN_LEVEL_NOTE, 1,
                 "'NDBINFO' has been started in offline mode "
                 "since the 'NDBCLUSTER' engine is disabled "
                 "or @@global.ndbinfo_offline is turned on "
                 "- no rows can be returned");
    DBUG_RETURN(0);
  }

  assert(is_open());
  assert(m_impl.m_scan_op == NULL); // No scan already ongoing

  if (m_impl.m_first_use)
  {
    m_impl.m_first_use = false;

    /*
      Check table def. and generate warnings for incompatibilites
      which is allowed but should generate a warning.
      (Done this late due to different code paths in MySQL Server for
      prepared statement protocol, where warnings from 'handler::open'
      are lost).
    */
    uint fields_found_in_ndb = 0;
    const NdbInfo::Table* ndb_tab = m_impl.m_table;
    for (uint i = 0; i < table->s->fields; i++)
    {
      const Field* field = table->field[i];
      const NdbInfo::Column* col = ndb_tab->getColumn(field->field_name);
      if (!col)
      {
        // The column didn't exist
        warn_incompatible(ndb_tab, true,
                          "column '%s' does not exist",
                          field->field_name);
        continue;
      }
      fields_found_in_ndb++;
    }

    if (fields_found_in_ndb < ndb_tab->columns())
    {
      // There are more columns available in NDB
      warn_incompatible(ndb_tab, false,
                        "there are more columns available");
    }
  }

  if (!scan)
  {
    // Just an init to read using 'rnd_pos'
    DBUG_PRINT("info", ("not scan"));
    DBUG_RETURN(0);
  }

  THD* thd = current_thd;
  int err;
  NdbInfoScanOperation* scan_op = NULL;
  if ((err = g_ndbinfo->createScanOperation(m_impl.m_table,
                                            &scan_op,
                                            THDVAR(thd, max_rows),
                                            THDVAR(thd, max_bytes))) != 0)
    DBUG_RETURN(err2mysql(err));

  if ((err = scan_op->readTuples()) != 0)
    DBUG_RETURN(err2mysql(err));

  /* Read all columns specified in read_set */
  for (uint i = 0; i < table->s->fields; i++)
  {
    Field *field = table->field[i];
    if (bitmap_is_set(table->read_set, i))
      m_impl.m_columns.push_back(scan_op->getValue(field->field_name));
    else
      m_impl.m_columns.push_back(NULL);
  }

  if ((err = scan_op->execute()) != 0)
    DBUG_RETURN(err2mysql(err));

  m_impl.m_scan_op = scan_op;
  DBUG_RETURN(0);
}

int ha_ndbinfo::rnd_end()
{
  DBUG_ENTER("ha_ndbinfo::rnd_end");

  if (is_offline())
    DBUG_RETURN(0);

  assert(is_open());

  if (m_impl.m_scan_op)
  {
    g_ndbinfo->releaseScanOperation(m_impl.m_scan_op);
    m_impl.m_scan_op = NULL;
  }
  m_impl.m_columns.clear();

  DBUG_RETURN(0);
}

int ha_ndbinfo::rnd_next(uchar *buf)
{
  int err;
  DBUG_ENTER("ha_ndbinfo::rnd_next");

  if (is_offline())
    DBUG_RETURN(HA_ERR_END_OF_FILE);

  assert(is_open());
  assert(m_impl.m_scan_op);

  if ((err = m_impl.m_scan_op->nextResult()) == 0)
    DBUG_RETURN(HA_ERR_END_OF_FILE);

  if (err != 1)
    DBUG_RETURN(err2mysql(err));

  unpack_record(buf);

  DBUG_RETURN(0);
}

int ha_ndbinfo::rnd_pos(uchar *buf, uchar *pos)
{
  DBUG_ENTER("ha_ndbinfo::rnd_pos");
  assert(is_open());
  assert(m_impl.m_scan_op == NULL); // No scan started

  /* Copy the saved row into "buf" and set all fields to not null */
  memcpy(buf, pos, ref_length);
  for (uint i = 0; i < table->s->fields; i++)
    table->field[i]->set_notnull();

  DBUG_RETURN(0);
}

void ha_ndbinfo::position(const uchar *record)
{
  DBUG_ENTER("ha_ndbinfo::position");
  assert(is_open());
  assert(m_impl.m_scan_op);

  /* Save away the whole row in "ref" */
  memcpy(ref, record, ref_length);

  DBUG_VOID_RETURN;
}

int ha_ndbinfo::info(uint flag)
{
  DBUG_ENTER("ha_ndbinfo::info");
  DBUG_PRINT("enter", ("flag: %d", flag));
  DBUG_RETURN(0);
}

void
ha_ndbinfo::unpack_record(uchar *dst_row)
{
  DBUG_ENTER("ha_ndbinfo::unpack_record");
  my_ptrdiff_t dst_offset = dst_row - table->record[0];

  for (uint i = 0; i < table->s->fields; i++)
  {
    Field *field = table->field[i];
    const NdbInfoRecAttr* record = m_impl.m_columns[i];
    if (record && !record->isNULL())
    {
      field->set_notnull();
      field->move_field_offset(dst_offset);
      switch (field->type()) {

      case (MYSQL_TYPE_VARCHAR):
      {
        DBUG_PRINT("info", ("str: %s", record->c_str()));
        Field_varstring* vfield = (Field_varstring *) field;
        /* Field_bit in DBUG requires the bit set in write_set for store(). */
        my_bitmap_map *old_map =
          dbug_tmp_use_all_columns(table, table->write_set);
        (void)vfield->store(record->c_str(),
                            MIN(record->length(), field->field_length)-1,
                            field->charset());
        dbug_tmp_restore_column_map(table->write_set, old_map);
        break;
      }

      case (MYSQL_TYPE_LONG):
      {
        memcpy(field->ptr, record->ptr(), sizeof(Uint32));
        break;
      }

      case (MYSQL_TYPE_LONGLONG):
      {
        memcpy(field->ptr, record->ptr(), sizeof(Uint64));
        break;
      }

      default:
        sql_print_error("Found unexpected field type %u", field->type());
        break;
      }

      field->move_field_offset(-dst_offset);
    }
    else
    {
      field->set_null();
    }
  }
  DBUG_VOID_RETURN;
}


static int
ndbinfo_find_files(handlerton *hton, THD *thd,
                   const char *db, const char *path,
                   const char *wild, bool dir, List<LEX_STRING> *files)
{
  DBUG_ENTER("ndbinfo_find_files");
  DBUG_PRINT("enter", ("db: '%s', dir: %d, path: '%s'", db, dir, path));

  const bool show_hidden = THDVAR(thd, show_hidden);

  if(show_hidden)
    DBUG_RETURN(0); // Don't filter out anything

  if (dir)
  {
    if (!ndbcluster_is_disabled())
      DBUG_RETURN(0);

    // Hide our database when ndbcluster is disabled
    LEX_STRING *dir_name;
    List_iterator<LEX_STRING> it(*files);
    while ((dir_name=it++))
    {
      if (strcmp(dir_name->str, opt_ndbinfo_dbname))
        continue;

      DBUG_PRINT("info", ("Hiding own databse '%s'", dir_name->str));
      it.remove();
    }

    DBUG_RETURN(0);
  }

  DBUG_ASSERT(db);
  if (strcmp(db, opt_ndbinfo_dbname))
    DBUG_RETURN(0); // Only hide files in "our" db

  /* Hide all files that start with "our" prefix */
  LEX_STRING *file_name;
  List_iterator<LEX_STRING> it(*files);
  while ((file_name=it++))
  {
    if (is_prefix(file_name->str, opt_ndbinfo_table_prefix))
    {
      DBUG_PRINT("info", ("Hiding '%s'", file_name->str));
      it.remove();
    }
  }

  DBUG_RETURN(0);
}


handlerton* ndbinfo_hton;

int ndbinfo_init(void *plugin)
{
  DBUG_ENTER("ndbinfo_init");

  handlerton *hton = (handlerton *) plugin;
  hton->create = create_handler;
  hton->flags =
    HTON_TEMPORARY_NOT_SUPPORTED |
    HTON_ALTER_NOT_SUPPORTED;
  hton->find_files = ndbinfo_find_files;

  ndbinfo_hton = hton;

  if (ndbcluster_is_disabled())
  {
    // Starting in limited mode since ndbcluster is disabled
     DBUG_RETURN(0);
  }

  char prefix[FN_REFLEN];
  build_table_filename(prefix, sizeof(prefix) - 1,
                       opt_ndbinfo_dbname, opt_ndbinfo_table_prefix, "", 0);
  DBUG_PRINT("info", ("prefix: '%s'", prefix));
  assert(g_ndb_cluster_connection);
  g_ndbinfo = new NdbInfo(g_ndb_cluster_connection, prefix,
                          opt_ndbinfo_dbname, opt_ndbinfo_table_prefix);
  if (!g_ndbinfo)
  {
    sql_print_error("Failed to create NdbInfo");
    DBUG_RETURN(1);
  }

  if (!g_ndbinfo->init())
  {
    sql_print_error("Failed to init NdbInfo");

    delete g_ndbinfo;
    g_ndbinfo = NULL;

    DBUG_RETURN(1);
  }

  DBUG_RETURN(0);
}

int ndbinfo_deinit(void *plugin)
{
  DBUG_ENTER("ndbinfo_deinit");

  if (g_ndbinfo)
  {
    delete g_ndbinfo;
    g_ndbinfo = NULL;
  }

  DBUG_RETURN(0);
}

struct st_mysql_sys_var* ndbinfo_system_variables[]= {
  MYSQL_SYSVAR(max_rows),
  MYSQL_SYSVAR(max_bytes),
  MYSQL_SYSVAR(show_hidden),
  MYSQL_SYSVAR(database),
  MYSQL_SYSVAR(table_prefix),
  MYSQL_SYSVAR(version),
  MYSQL_SYSVAR(offline),

  NULL
};

template class Vector<const NdbInfoRecAttr*>;

#endif