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;

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;

{

/*

if (g_ndb_cluster_connection)

return false;

assert(g_ndbinfo == NULL);

return true;

{

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)

{

}

{

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);

{

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);

{

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'");

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;

HA_CREATE_INFO *create_info)

{

DBUG_ENTER("ha_ndbinfo::create");

DBUG_PRINT("enter", ("name: %s", name));

DBUG_RETURN(0);

{

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));

}

/*

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);

{

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);

{

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;

/*

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);

{

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 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);

{

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);

{

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;

{

DBUG_ENTER("ha_ndbinfo::info");

DBUG_PRINT("enter", ("flag: %d", flag));

DBUG_RETURN(0);

{

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;

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);

{

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);

{

DBUG_ENTER("ndbinfo_deinit");

if (g_ndbinfo)

{

delete g_ndbinfo;

g_ndbinfo = NULL;

}

DBUG_RETURN(0);