mirror of
https://github.com/MariaDB/server.git
synced 2025-01-18 04:53:01 +01:00
606073084e
mkdir in Makefile, not in configure chmod a-x
2943 lines
74 KiB
C++
2943 lines
74 KiB
C++
/* Copyright (C) 2000-2003 MySQL AB
|
|
|
|
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; either version 2 of the License, or
|
|
(at your option) any later version.
|
|
|
|
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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
*/
|
|
|
|
/*
|
|
This file defines the NDB Cluster handler: the interface between MySQL and
|
|
NDB Cluster
|
|
*/
|
|
|
|
/*
|
|
TODO
|
|
After CREATE DATABASE gör discover på alla tabeller i den databasen
|
|
|
|
*/
|
|
|
|
|
|
#ifdef __GNUC__
|
|
#pragma implementation // gcc: Class implementation
|
|
#endif
|
|
|
|
#include "mysql_priv.h"
|
|
|
|
#ifdef HAVE_NDBCLUSTER_DB
|
|
#include <my_dir.h>
|
|
#include "ha_ndbcluster.h"
|
|
#include <ndbapi/NdbApi.hpp>
|
|
#include <ndbapi/NdbScanFilter.hpp>
|
|
|
|
#define USE_DISCOVER_ON_STARTUP
|
|
//#define USE_NDB_POOL
|
|
|
|
// Default value for parallelism
|
|
static const int parallelism= 240;
|
|
|
|
#define NDB_HIDDEN_PRIMARY_KEY_LENGTH 8
|
|
|
|
#define ERR_PRINT(err) \
|
|
DBUG_PRINT("error", ("Error: %d message: %s", err.code, err.message))
|
|
|
|
#define ERR_RETURN(err) \
|
|
{ \
|
|
ERR_PRINT(err); \
|
|
DBUG_RETURN(ndb_to_mysql_error(&err)); \
|
|
}
|
|
|
|
// Typedefs for long names
|
|
typedef NdbDictionary::Column NDBCOL;
|
|
typedef NdbDictionary::Table NDBTAB;
|
|
typedef NdbDictionary::Index NDBINDEX;
|
|
typedef NdbDictionary::Dictionary NDBDICT;
|
|
|
|
bool ndbcluster_inited= false;
|
|
|
|
// Handler synchronization
|
|
pthread_mutex_t ndbcluster_mutex;
|
|
|
|
// Table lock handling
|
|
static HASH ndbcluster_open_tables;
|
|
|
|
static byte *ndbcluster_get_key(NDB_SHARE *share,uint *length,
|
|
my_bool not_used __attribute__((unused)));
|
|
static NDB_SHARE *get_share(const char *table_name);
|
|
static void free_share(NDB_SHARE *share);
|
|
|
|
static int packfrm(const void *data, uint len, const void **pack_data, uint *pack_len);
|
|
static int unpackfrm(const void **data, uint *len,
|
|
const void* pack_data);
|
|
|
|
/*
|
|
Error handling functions
|
|
*/
|
|
|
|
struct err_code_mapping
|
|
{
|
|
int ndb_err;
|
|
int my_err;
|
|
};
|
|
|
|
static const err_code_mapping err_map[]=
|
|
{
|
|
{ 626, HA_ERR_KEY_NOT_FOUND },
|
|
{ 630, HA_ERR_FOUND_DUPP_KEY },
|
|
{ 893, HA_ERR_FOUND_DUPP_UNIQUE },
|
|
{ 721, HA_ERR_TABLE_EXIST },
|
|
{ 241, HA_ERR_OLD_METADATA },
|
|
{ -1, -1 }
|
|
};
|
|
|
|
|
|
static int ndb_to_mysql_error(const NdbError *err)
|
|
{
|
|
uint i;
|
|
for (i=0 ; err_map[i].ndb_err != err->code ; i++)
|
|
{
|
|
if (err_map[i].my_err == -1)
|
|
return err->code;
|
|
}
|
|
return err_map[i].my_err;
|
|
}
|
|
|
|
|
|
/*
|
|
Take care of the error that occured in NDB
|
|
|
|
RETURN
|
|
0 No error
|
|
# The mapped error code
|
|
*/
|
|
|
|
int ha_ndbcluster::ndb_err(NdbConnection *trans)
|
|
{
|
|
const NdbError err= trans->getNdbError();
|
|
if (!err.code)
|
|
return 0; // Don't log things to DBUG log if no error
|
|
DBUG_ENTER("ndb_err");
|
|
|
|
ERR_PRINT(err);
|
|
switch (err.classification) {
|
|
case NdbError::SchemaError:
|
|
{
|
|
NDBDICT *dict= m_ndb->getDictionary();
|
|
DBUG_PRINT("info", ("invalidateTable %s", m_tabname));
|
|
dict->invalidateTable(m_tabname);
|
|
break;
|
|
}
|
|
default:
|
|
break;
|
|
}
|
|
DBUG_RETURN(ndb_to_mysql_error(&err));
|
|
}
|
|
|
|
|
|
/*
|
|
Instruct NDB to set the value of the hidden primary key
|
|
*/
|
|
|
|
bool ha_ndbcluster::set_hidden_key(NdbOperation *ndb_op,
|
|
uint fieldnr, const byte *field_ptr)
|
|
{
|
|
DBUG_ENTER("set_hidden_key");
|
|
DBUG_RETURN(ndb_op->equal(fieldnr, (char*)field_ptr,
|
|
NDB_HIDDEN_PRIMARY_KEY_LENGTH) != 0);
|
|
}
|
|
|
|
|
|
/*
|
|
Instruct NDB to set the value of one primary key attribute
|
|
*/
|
|
|
|
int ha_ndbcluster::set_ndb_key(NdbOperation *ndb_op, Field *field,
|
|
uint fieldnr, const byte *field_ptr)
|
|
{
|
|
uint32 pack_len= field->pack_length();
|
|
DBUG_ENTER("set_ndb_key");
|
|
DBUG_PRINT("enter", ("%d: %s, ndb_type: %u, len=%d",
|
|
fieldnr, field->field_name, field->type(),
|
|
pack_len));
|
|
DBUG_DUMP("key", (char*)field_ptr, pack_len);
|
|
|
|
switch (field->type()) {
|
|
case MYSQL_TYPE_DECIMAL:
|
|
case MYSQL_TYPE_TINY:
|
|
case MYSQL_TYPE_SHORT:
|
|
case MYSQL_TYPE_LONG:
|
|
case MYSQL_TYPE_FLOAT:
|
|
case MYSQL_TYPE_DOUBLE:
|
|
case MYSQL_TYPE_TIMESTAMP:
|
|
case MYSQL_TYPE_LONGLONG:
|
|
case MYSQL_TYPE_INT24:
|
|
case MYSQL_TYPE_DATE:
|
|
case MYSQL_TYPE_TIME:
|
|
case MYSQL_TYPE_DATETIME:
|
|
case MYSQL_TYPE_YEAR:
|
|
case MYSQL_TYPE_NEWDATE:
|
|
case MYSQL_TYPE_ENUM:
|
|
case MYSQL_TYPE_SET:
|
|
case MYSQL_TYPE_VAR_STRING:
|
|
case MYSQL_TYPE_STRING:
|
|
// Common implementation for most field types
|
|
DBUG_RETURN(ndb_op->equal(fieldnr, (char*) field_ptr, pack_len) != 0);
|
|
|
|
case MYSQL_TYPE_TINY_BLOB:
|
|
case MYSQL_TYPE_MEDIUM_BLOB:
|
|
case MYSQL_TYPE_LONG_BLOB:
|
|
case MYSQL_TYPE_BLOB:
|
|
case MYSQL_TYPE_NULL:
|
|
case MYSQL_TYPE_GEOMETRY:
|
|
default:
|
|
// Unhandled field types
|
|
DBUG_PRINT("error", ("Field type %d not supported", field->type()));
|
|
DBUG_RETURN(2);
|
|
}
|
|
DBUG_RETURN(3);
|
|
}
|
|
|
|
|
|
/*
|
|
Instruct NDB to set the value of one attribute
|
|
*/
|
|
|
|
int ha_ndbcluster::set_ndb_value(NdbOperation *ndb_op, Field *field,
|
|
uint fieldnr)
|
|
{
|
|
const byte* field_ptr= field->ptr;
|
|
uint32 pack_len= field->pack_length();
|
|
DBUG_ENTER("set_ndb_value");
|
|
DBUG_PRINT("enter", ("%d: %s, type: %u, len=%d, is_null=%s",
|
|
fieldnr, field->field_name, field->type(),
|
|
pack_len, field->is_null()?"Y":"N"));
|
|
DBUG_DUMP("value", (char*) field_ptr, pack_len);
|
|
|
|
if (field->is_null())
|
|
{
|
|
// Set value to NULL
|
|
DBUG_RETURN((ndb_op->setValue(fieldnr, (char*)NULL, pack_len) != 0));
|
|
}
|
|
|
|
switch (field->type()) {
|
|
case MYSQL_TYPE_DECIMAL:
|
|
case MYSQL_TYPE_TINY:
|
|
case MYSQL_TYPE_SHORT:
|
|
case MYSQL_TYPE_LONG:
|
|
case MYSQL_TYPE_FLOAT:
|
|
case MYSQL_TYPE_DOUBLE:
|
|
case MYSQL_TYPE_TIMESTAMP:
|
|
case MYSQL_TYPE_LONGLONG:
|
|
case MYSQL_TYPE_INT24:
|
|
case MYSQL_TYPE_DATE:
|
|
case MYSQL_TYPE_TIME:
|
|
case MYSQL_TYPE_DATETIME:
|
|
case MYSQL_TYPE_YEAR:
|
|
case MYSQL_TYPE_NEWDATE:
|
|
case MYSQL_TYPE_ENUM:
|
|
case MYSQL_TYPE_SET:
|
|
case MYSQL_TYPE_VAR_STRING:
|
|
case MYSQL_TYPE_STRING:
|
|
// Common implementation for most field types
|
|
DBUG_RETURN(ndb_op->setValue(fieldnr, (char*)field_ptr, pack_len) != 0);
|
|
|
|
case MYSQL_TYPE_TINY_BLOB:
|
|
case MYSQL_TYPE_MEDIUM_BLOB:
|
|
case MYSQL_TYPE_LONG_BLOB:
|
|
case MYSQL_TYPE_BLOB:
|
|
case MYSQL_TYPE_NULL:
|
|
case MYSQL_TYPE_GEOMETRY:
|
|
default:
|
|
// Unhandled field types
|
|
DBUG_PRINT("error", ("Field type %d not supported", field->type()));
|
|
DBUG_RETURN(2);
|
|
}
|
|
DBUG_RETURN(3);
|
|
}
|
|
|
|
|
|
/*
|
|
Instruct NDB to fetch one field
|
|
- data is read directly into buffer provided by field_ptr
|
|
if it's NULL, data is read into memory provided by NDBAPI
|
|
*/
|
|
|
|
int ha_ndbcluster::get_ndb_value(NdbOperation *op,
|
|
uint field_no, byte *field_ptr)
|
|
{
|
|
DBUG_ENTER("get_ndb_value");
|
|
DBUG_PRINT("enter", ("field_no: %d", field_no));
|
|
m_value[field_no]= op->getValue(field_no, field_ptr);
|
|
DBUG_RETURN(m_value == NULL);
|
|
}
|
|
|
|
|
|
/*
|
|
Get metadata for this table from NDB
|
|
|
|
IMPLEMENTATION
|
|
- save the NdbDictionary::Table for easy access
|
|
- check that frm-file on disk is equal to frm-file
|
|
of table accessed in NDB
|
|
- build a list of the indexes for the table
|
|
*/
|
|
|
|
int ha_ndbcluster::get_metadata(const char *path)
|
|
{
|
|
NDBDICT *dict= m_ndb->getDictionary();
|
|
const NDBTAB *tab;
|
|
const void *data, *pack_data;
|
|
const char **key_name;
|
|
uint ndb_columns, mysql_columns, length, pack_length, i;
|
|
int error;
|
|
DBUG_ENTER("get_metadata");
|
|
DBUG_PRINT("enter", ("m_tabname: %s, path: %s", m_tabname, path));
|
|
|
|
if (!(tab= dict->getTable(m_tabname)))
|
|
ERR_RETURN(dict->getNdbError());
|
|
DBUG_PRINT("info", ("Table schema version: %d", tab->getObjectVersion()));
|
|
|
|
/*
|
|
This is the place to check that the table we got from NDB
|
|
is equal to the one on local disk
|
|
*/
|
|
ndb_columns= (uint) tab->getNoOfColumns();
|
|
mysql_columns= table->fields;
|
|
if (table->primary_key == MAX_KEY)
|
|
ndb_columns--;
|
|
if (ndb_columns != mysql_columns)
|
|
{
|
|
DBUG_PRINT("error",
|
|
("Wrong number of columns, ndb: %d mysql: %d",
|
|
ndb_columns, mysql_columns));
|
|
DBUG_RETURN(HA_ERR_OLD_METADATA);
|
|
}
|
|
|
|
/*
|
|
Compare FrmData in NDB with frm file from disk.
|
|
*/
|
|
error= 0;
|
|
if (readfrm(path, &data, &length) ||
|
|
packfrm(data, length, &pack_data, &pack_length))
|
|
{
|
|
my_free((char*)data, MYF(MY_ALLOW_ZERO_PTR));
|
|
my_free((char*)pack_data, MYF(MY_ALLOW_ZERO_PTR));
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
if ((pack_length != tab->getFrmLength()) ||
|
|
(memcmp(pack_data, tab->getFrmData(), pack_length)))
|
|
{
|
|
DBUG_PRINT("error",
|
|
("metadata, pack_length: %d getFrmLength: %d memcmp: %d",
|
|
pack_length, tab->getFrmLength(),
|
|
memcmp(pack_data, tab->getFrmData(), pack_length)));
|
|
DBUG_DUMP("pack_data", (char*)pack_data, pack_length);
|
|
DBUG_DUMP("frm", (char*)tab->getFrmData(), tab->getFrmLength());
|
|
error= HA_ERR_OLD_METADATA;
|
|
}
|
|
my_free((char*)data, MYF(0));
|
|
my_free((char*)pack_data, MYF(0));
|
|
if (error)
|
|
DBUG_RETURN(error);
|
|
|
|
// All checks OK, lets use the table
|
|
m_table= (void*)tab;
|
|
|
|
for (i= 0; i < MAX_KEY; i++)
|
|
m_indextype[i]= UNDEFINED_INDEX;
|
|
|
|
// Save information about all known indexes
|
|
for (i= 0; i < table->keys; i++)
|
|
m_indextype[i] = get_index_type_from_table(i);
|
|
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
/*
|
|
Decode the type of an index from information
|
|
provided in table object
|
|
*/
|
|
NDB_INDEX_TYPE ha_ndbcluster::get_index_type_from_table(uint index_no) const
|
|
{
|
|
if (index_no == table->primary_key)
|
|
return PRIMARY_KEY_INDEX;
|
|
else
|
|
return ((table->key_info[index_no].flags & HA_NOSAME) ?
|
|
UNIQUE_INDEX :
|
|
ORDERED_INDEX);
|
|
}
|
|
|
|
|
|
void ha_ndbcluster::release_metadata()
|
|
{
|
|
DBUG_ENTER("release_metadata");
|
|
DBUG_PRINT("enter", ("m_tabname: %s", m_tabname));
|
|
|
|
m_table= NULL;
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
static const ulong index_type_flags[]=
|
|
{
|
|
/* UNDEFINED_INDEX */
|
|
0,
|
|
|
|
/* PRIMARY_KEY_INDEX */
|
|
HA_ONLY_WHOLE_INDEX |
|
|
HA_WRONG_ASCII_ORDER |
|
|
HA_NOT_READ_PREFIX_LAST,
|
|
|
|
/* UNIQUE_INDEX */
|
|
HA_ONLY_WHOLE_INDEX |
|
|
HA_WRONG_ASCII_ORDER |
|
|
HA_NOT_READ_PREFIX_LAST,
|
|
|
|
/* ORDERED_INDEX */
|
|
HA_READ_NEXT |
|
|
HA_READ_PREV |
|
|
HA_NOT_READ_AFTER_KEY
|
|
};
|
|
|
|
static const int index_flags_size= sizeof(index_type_flags)/sizeof(ulong);
|
|
|
|
inline const char* ha_ndbcluster::get_index_name(uint idx_no) const
|
|
{
|
|
return table->keynames.type_names[idx_no];
|
|
}
|
|
|
|
inline NDB_INDEX_TYPE ha_ndbcluster::get_index_type(uint idx_no) const
|
|
{
|
|
DBUG_ASSERT(idx_no < MAX_KEY);
|
|
return m_indextype[idx_no];
|
|
}
|
|
|
|
|
|
/*
|
|
Get the flags for an index
|
|
|
|
RETURN
|
|
flags depending on the type of the index.
|
|
*/
|
|
|
|
inline ulong ha_ndbcluster::index_flags(uint idx_no) const
|
|
{
|
|
DBUG_ENTER("index_flags");
|
|
DBUG_ASSERT(get_index_type_from_table(idx_no) < index_flags_size);
|
|
DBUG_RETURN(index_type_flags[get_index_type_from_table(idx_no)]);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::set_primary_key(NdbOperation *op, const byte *key)
|
|
{
|
|
KEY* key_info= table->key_info + table->primary_key;
|
|
KEY_PART_INFO* key_part= key_info->key_part;
|
|
KEY_PART_INFO* end= key_part+key_info->key_parts;
|
|
DBUG_ENTER("set_primary_key");
|
|
|
|
for (; key_part != end; key_part++)
|
|
{
|
|
Field* field= key_part->field;
|
|
if (set_ndb_key(op, field,
|
|
key_part->fieldnr-1, key))
|
|
ERR_RETURN(op->getNdbError());
|
|
key += key_part->length;
|
|
}
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::set_primary_key(NdbOperation *op)
|
|
{
|
|
DBUG_ENTER("set_primary_key");
|
|
KEY* key_info= table->key_info + table->primary_key;
|
|
KEY_PART_INFO* key_part= key_info->key_part;
|
|
KEY_PART_INFO* end= key_part+key_info->key_parts;
|
|
|
|
for (; key_part != end; key_part++)
|
|
{
|
|
Field* field= key_part->field;
|
|
if (set_ndb_key(op, field,
|
|
key_part->fieldnr-1, field->ptr))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Read one record from NDB using primary key
|
|
*/
|
|
|
|
int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf)
|
|
{
|
|
uint no_fields= table->fields, i;
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbOperation *op;
|
|
THD *thd= current_thd;
|
|
DBUG_ENTER("pk_read");
|
|
DBUG_PRINT("enter", ("key_len: %u", key_len));
|
|
DBUG_DUMP("key", (char*)key, key_len);
|
|
|
|
if (!(op= trans->getNdbOperation(m_tabname)) || op->readTuple() != 0)
|
|
goto err;
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
// This table has no primary key, use "hidden" primary key
|
|
DBUG_PRINT("info", ("Using hidden key"));
|
|
DBUG_DUMP("key", (char*)key, 8);
|
|
if (set_hidden_key(op, no_fields, key))
|
|
goto err;
|
|
// Read key at the same time, for future reference
|
|
if (get_ndb_value(op, no_fields, NULL))
|
|
goto err;
|
|
}
|
|
else
|
|
{
|
|
int res;
|
|
if ((res= set_primary_key(op, key)))
|
|
return res;
|
|
}
|
|
|
|
// Read non-key field(s)
|
|
for (i= 0; i < no_fields; i++)
|
|
{
|
|
Field *field= table->field[i];
|
|
if (thd->query_id == field->query_id)
|
|
{
|
|
if (get_ndb_value(op, i, field->ptr))
|
|
goto err;
|
|
}
|
|
else
|
|
{
|
|
// Attribute was not to be read
|
|
m_value[i]= NULL;
|
|
}
|
|
}
|
|
|
|
if (trans->execute(NoCommit, IgnoreError) != 0)
|
|
{
|
|
table->status= STATUS_NOT_FOUND;
|
|
DBUG_RETURN(ndb_err(trans));
|
|
}
|
|
|
|
// The value have now been fetched from NDB
|
|
unpack_record(buf);
|
|
table->status= 0;
|
|
DBUG_RETURN(0);
|
|
|
|
err:
|
|
ERR_RETURN(trans->getNdbError());
|
|
}
|
|
|
|
|
|
/*
|
|
Read one record from NDB using unique secondary index
|
|
*/
|
|
|
|
int ha_ndbcluster::unique_index_read(const byte *key,
|
|
uint key_len, byte *buf)
|
|
{
|
|
NdbConnection *trans= m_active_trans;
|
|
const char *index_name;
|
|
NdbIndexOperation *op;
|
|
THD *thd= current_thd;
|
|
byte *key_ptr;
|
|
KEY* key_info;
|
|
KEY_PART_INFO *key_part, *end;
|
|
uint i;
|
|
DBUG_ENTER("unique_index_read");
|
|
DBUG_PRINT("enter", ("key_len: %u, index: %u", key_len, active_index));
|
|
DBUG_DUMP("key", (char*)key, key_len);
|
|
|
|
index_name= get_index_name(active_index);
|
|
if (!(op= trans->getNdbIndexOperation(index_name, m_tabname)) ||
|
|
op->readTuple() != 0)
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
// Set secondary index key(s)
|
|
key_ptr= (byte *) key;
|
|
key_info= table->key_info + active_index;
|
|
DBUG_ASSERT(key_info->key_length == key_len);
|
|
end= (key_part= key_info->key_part) + key_info->key_parts;
|
|
|
|
for (i= 0; key_part != end; key_part++, i++)
|
|
{
|
|
if (set_ndb_key(op, key_part->field, i, key_ptr))
|
|
ERR_RETURN(trans->getNdbError());
|
|
key_ptr+= key_part->length;
|
|
}
|
|
|
|
// Get non-index attribute(s)
|
|
for (i= 0; i < table->fields; i++)
|
|
{
|
|
Field *field= table->field[i];
|
|
if ((thd->query_id == field->query_id) ||
|
|
(field->flags & PRI_KEY_FLAG))
|
|
{
|
|
if (get_ndb_value(op, i, field->ptr))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
else
|
|
{
|
|
// Attribute was not to be read
|
|
m_value[i]= NULL;
|
|
}
|
|
}
|
|
|
|
if (trans->execute(NoCommit, IgnoreError) != 0)
|
|
{
|
|
table->status= STATUS_NOT_FOUND;
|
|
DBUG_RETURN(ndb_err(trans));
|
|
}
|
|
// The value have now been fetched from NDB
|
|
unpack_record(buf);
|
|
table->status= 0;
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
/*
|
|
Get the next record of a started scan
|
|
*/
|
|
|
|
inline int ha_ndbcluster::next_result(byte *buf)
|
|
{
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbResultSet *cursor= m_active_cursor;
|
|
DBUG_ENTER("next_result");
|
|
|
|
if (cursor->nextResult() == 0)
|
|
{
|
|
// One more record found
|
|
unpack_record(buf);
|
|
table->status= 0;
|
|
DBUG_RETURN(0);
|
|
}
|
|
table->status= STATUS_NOT_FOUND;
|
|
if (ndb_err(trans))
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
// No more records
|
|
DBUG_PRINT("info", ("No more records"));
|
|
DBUG_RETURN(HA_ERR_END_OF_FILE);
|
|
}
|
|
|
|
|
|
/*
|
|
Read record(s) from NDB using ordered index scan
|
|
*/
|
|
|
|
int ha_ndbcluster::ordered_index_scan(const byte *key, uint key_len,
|
|
byte *buf,
|
|
enum ha_rkey_function find_flag)
|
|
{
|
|
uint no_fields= table->fields;
|
|
uint tot_len, i;
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbResultSet *cursor= m_active_cursor;
|
|
NdbScanOperation *op;
|
|
const char *bound_str= NULL;
|
|
const char *index_name;
|
|
NdbOperation::BoundType bound_type = NdbOperation::BoundEQ;
|
|
bool can_be_handled_by_ndb= FALSE;
|
|
byte *key_ptr;
|
|
KEY *key_info;
|
|
THD* thd = current_thd;
|
|
DBUG_ENTER("ordered_index_scan");
|
|
DBUG_PRINT("enter", ("index: %u", active_index));
|
|
DBUG_PRINT("enter", ("Starting new ordered scan on %s", m_tabname));
|
|
|
|
index_name= get_index_name(active_index);
|
|
if (!(op= trans->getNdbScanOperation(index_name, m_tabname)))
|
|
ERR_RETURN(trans->getNdbError());
|
|
if (!(cursor= op->readTuples(parallelism)))
|
|
ERR_RETURN(trans->getNdbError());
|
|
m_active_cursor= cursor;
|
|
|
|
switch (find_flag) {
|
|
case HA_READ_KEY_EXACT: /* Find first record else error */
|
|
bound_str= "HA_READ_KEY_EXACT";
|
|
bound_type= NdbOperation::BoundEQ;
|
|
can_be_handled_by_ndb= TRUE;
|
|
break;
|
|
case HA_READ_KEY_OR_NEXT: /* Record or next record */
|
|
bound_str= "HA_READ_KEY_OR_NEXT";
|
|
bound_type= NdbOperation::BoundLE;
|
|
can_be_handled_by_ndb= TRUE;
|
|
break;
|
|
case HA_READ_KEY_OR_PREV: /* Record or previous */
|
|
bound_str= "HA_READ_KEY_OR_PREV";
|
|
bound_type= NdbOperation::BoundGE;
|
|
can_be_handled_by_ndb= TRUE;
|
|
break;
|
|
case HA_READ_AFTER_KEY: /* Find next rec. after key-record */
|
|
bound_str= "HA_READ_AFTER_KEY";
|
|
bound_type= NdbOperation::BoundLT;
|
|
can_be_handled_by_ndb= TRUE;
|
|
break;
|
|
case HA_READ_BEFORE_KEY: /* Find next rec. before key-record */
|
|
bound_str= "HA_READ_BEFORE_KEY";
|
|
bound_type= NdbOperation::BoundGT;
|
|
can_be_handled_by_ndb= TRUE;
|
|
break;
|
|
case HA_READ_PREFIX: /* Key which as same prefix */
|
|
bound_str= "HA_READ_PREFIX";
|
|
break;
|
|
case HA_READ_PREFIX_LAST: /* Last key with the same prefix */
|
|
bound_str= "HA_READ_PREFIX_LAST";
|
|
break;
|
|
case HA_READ_PREFIX_LAST_OR_PREV:
|
|
/* Last or prev key with the same prefix */
|
|
bound_str= "HA_READ_PREFIX_LAST_OR_PREV";
|
|
break;
|
|
default:
|
|
bound_str= "UNKNOWN";
|
|
break;
|
|
}
|
|
DBUG_PRINT("info", ("find_flag: %s, bound_type: %d,"
|
|
"can_be_handled_by_ndb: %d",
|
|
bound_str, bound_type, can_be_handled_by_ndb));
|
|
if (!can_be_handled_by_ndb)
|
|
DBUG_RETURN(1);
|
|
|
|
// Set bounds using key data
|
|
tot_len= 0;
|
|
key_ptr= (byte *) key;
|
|
key_info= table->key_info + active_index;
|
|
for (i= 0; i < key_info->key_parts; i++)
|
|
{
|
|
Field* field= key_info->key_part[i].field;
|
|
uint32 field_len= field->pack_length();
|
|
DBUG_PRINT("info", ("Set index bound on %s",
|
|
field->field_name));
|
|
DBUG_DUMP("key", (char*)key_ptr, field_len);
|
|
|
|
if (op->setBound(field->field_name,
|
|
bound_type,
|
|
key_ptr,
|
|
field_len) != 0)
|
|
ERR_RETURN(op->getNdbError());
|
|
|
|
key_ptr+= field_len;
|
|
tot_len+= field_len;
|
|
if (tot_len >= key_len)
|
|
break;
|
|
}
|
|
|
|
// Define attributes to read
|
|
for (i= 0; i < no_fields; i++)
|
|
{
|
|
Field *field= table->field[i];
|
|
if ((thd->query_id == field->query_id) ||
|
|
(field->flags & PRI_KEY_FLAG))
|
|
{
|
|
if (get_ndb_value(op, i, field->ptr))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
else
|
|
{
|
|
m_value[i]= NULL;
|
|
}
|
|
}
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
DBUG_PRINT("info", ("Getting hidden key"));
|
|
// Scanning table with no primary key
|
|
int hidden_no= no_fields;
|
|
#ifndef DBUG_OFF
|
|
const NDBTAB *tab= (NDBTAB *) m_table;
|
|
if (!tab->getColumn(hidden_no))
|
|
DBUG_RETURN(1);
|
|
#endif
|
|
if (get_ndb_value(op, hidden_no, NULL))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
|
|
if (trans->execute(NoCommit) != 0)
|
|
DBUG_RETURN(ndb_err(trans));
|
|
DBUG_PRINT("exit", ("Scan started successfully"));
|
|
DBUG_RETURN(next_result(buf));
|
|
}
|
|
|
|
|
|
#if 0
|
|
/*
|
|
Read record(s) from NDB using full table scan with filter
|
|
*/
|
|
|
|
int ha_ndbcluster::filtered_scan(const byte *key, uint key_len,
|
|
byte *buf,
|
|
enum ha_rkey_function find_flag)
|
|
{
|
|
uint no_fields= table->fields;
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbResultSet *cursor= m_active_cursor;
|
|
|
|
DBUG_ENTER("filtered_scan");
|
|
DBUG_PRINT("enter", ("key_len: %u, index: %u",
|
|
key_len, active_index));
|
|
DBUG_DUMP("key", (char*)key, key_len);
|
|
DBUG_PRINT("info", ("Starting a new filtered scan on %s",
|
|
m_tabname));
|
|
NdbScanOperation *op= trans->getNdbScanOperation(m_tabname);
|
|
if (!op)
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
cursor= op->readTuples(parallelism);
|
|
if (!cursor)
|
|
ERR_RETURN(trans->getNdbError());
|
|
m_active_cursor= cursor;
|
|
|
|
{
|
|
// Start scan filter
|
|
NdbScanFilter sf(op);
|
|
sf.begin();
|
|
|
|
// Set filter using the supplied key data
|
|
byte *key_ptr= (byte *) key;
|
|
uint tot_len= 0;
|
|
KEY* key_info= table->key_info + active_index;
|
|
for (uint k= 0; k < key_info->key_parts; k++)
|
|
{
|
|
KEY_PART_INFO* key_part= key_info->key_part+k;
|
|
Field* field= key_part->field;
|
|
uint ndb_fieldnr= key_part->fieldnr-1;
|
|
DBUG_PRINT("key_part", ("fieldnr: %d", ndb_fieldnr));
|
|
// const NDBCOL *col= tab->getColumn(ndb_fieldnr);
|
|
uint32 field_len= field->pack_length();
|
|
DBUG_DUMP("key", (char*)key, field_len);
|
|
|
|
DBUG_PRINT("info", ("Column %s, type: %d, len: %d",
|
|
field->field_name, field->real_type(), field_len));
|
|
|
|
// Define scan filter
|
|
if (field->real_type() == MYSQL_TYPE_STRING)
|
|
sf.eq(ndb_fieldnr, key_ptr, field_len);
|
|
else
|
|
{
|
|
if (field_len == 8)
|
|
sf.eq(ndb_fieldnr, (Uint64)*key_ptr);
|
|
else if (field_len <= 4)
|
|
sf.eq(ndb_fieldnr, (Uint32)*key_ptr);
|
|
else
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
key_ptr += field_len;
|
|
tot_len += field_len;
|
|
|
|
if (tot_len >= key_len)
|
|
break;
|
|
}
|
|
// End scan filter
|
|
sf.end();
|
|
}
|
|
|
|
// Define attributes to read
|
|
for (uint field_no= 0; field_no < no_fields; field_no++)
|
|
{
|
|
Field *field= table->field[field_no];
|
|
|
|
// Read attribute
|
|
DBUG_PRINT("get", ("%d: %s", field_no, field->field_name));
|
|
if (get_ndb_value(op, field_no, field->ptr))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
DBUG_PRINT("info", ("Getting hidden key"));
|
|
// Scanning table with no primary key
|
|
int hidden_no= no_fields;
|
|
#ifndef DBUG_OFF
|
|
const NDBTAB *tab= (NDBTAB *) m_table;
|
|
if (!tab->getColumn(hidden_no))
|
|
DBUG_RETURN(1);
|
|
#endif
|
|
if (get_ndb_value(op, hidden_no, NULL))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
|
|
if (trans->execute(NoCommit) != 0)
|
|
DBUG_RETURN(ndb_err(trans));
|
|
DBUG_PRINT("exit", ("Scan started successfully"));
|
|
DBUG_RETURN(next_result(buf));
|
|
}
|
|
#endif
|
|
|
|
|
|
/*
|
|
Read records from NDB using full table scan
|
|
*/
|
|
|
|
int ha_ndbcluster::full_table_scan(byte *buf)
|
|
{
|
|
uint i;
|
|
THD *thd= current_thd;
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbResultSet *cursor;
|
|
NdbScanOperation *op;
|
|
|
|
DBUG_ENTER("full_table_scan");
|
|
DBUG_PRINT("enter", ("Starting new scan on %s", m_tabname));
|
|
|
|
if (!(op=trans->getNdbScanOperation(m_tabname)))
|
|
ERR_RETURN(trans->getNdbError());
|
|
if (!(cursor= op->readTuples(parallelism)))
|
|
ERR_RETURN(trans->getNdbError());
|
|
m_active_cursor= cursor;
|
|
|
|
// Define attributes to read
|
|
for (i= 0; i < table->fields; i++)
|
|
{
|
|
Field *field= table->field[i];
|
|
if ((thd->query_id == field->query_id) ||
|
|
(field->flags & PRI_KEY_FLAG))
|
|
{
|
|
if (get_ndb_value(op, i, field->ptr))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
else
|
|
{
|
|
m_value[i]= NULL;
|
|
}
|
|
}
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
DBUG_PRINT("info", ("Getting hidden key"));
|
|
// Scanning table with no primary key
|
|
int hidden_no= table->fields;
|
|
#ifndef DBUG_OFF
|
|
const NDBTAB *tab= (NDBTAB *) m_table;
|
|
if (!tab->getColumn(hidden_no))
|
|
DBUG_RETURN(1);
|
|
#endif
|
|
if (get_ndb_value(op, hidden_no, NULL))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
|
|
if (trans->execute(NoCommit) != 0)
|
|
DBUG_RETURN(ndb_err(trans));
|
|
DBUG_PRINT("exit", ("Scan started successfully"));
|
|
DBUG_RETURN(next_result(buf));
|
|
}
|
|
|
|
|
|
/*
|
|
Insert one record into NDB
|
|
*/
|
|
|
|
int ha_ndbcluster::write_row(byte *record)
|
|
{
|
|
uint i;
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbOperation *op;
|
|
int res;
|
|
DBUG_ENTER("write_row");
|
|
|
|
statistic_increment(ha_write_count,&LOCK_status);
|
|
if (table->timestamp_default_now)
|
|
update_timestamp(record+table->timestamp_default_now-1);
|
|
if (table->next_number_field && record == table->record[0])
|
|
update_auto_increment();
|
|
|
|
if (!(op= trans->getNdbOperation(m_tabname)))
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
res= (m_use_write) ? op->writeTuple() :op->insertTuple();
|
|
if (res != 0)
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
// Table has hidden primary key
|
|
Uint64 auto_value= m_ndb->getAutoIncrementValue(m_tabname);
|
|
if (set_hidden_key(op, table->fields, (const byte*)&auto_value))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
else
|
|
{
|
|
int res;
|
|
if ((res= set_primary_key(op)))
|
|
return res;
|
|
}
|
|
|
|
// Set non-key attribute(s)
|
|
for (i= 0; i < table->fields; i++)
|
|
{
|
|
Field *field= table->field[i];
|
|
if (!(field->flags & PRI_KEY_FLAG) &&
|
|
set_ndb_value(op, field, i))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
|
|
/*
|
|
Execute write operation
|
|
NOTE When doing inserts with many values in
|
|
each INSERT statement it should not be necessary
|
|
to NoCommit the transaction between each row.
|
|
Find out how this is detected!
|
|
*/
|
|
if (trans->execute(NoCommit) != 0)
|
|
DBUG_RETURN(ndb_err(trans));
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/* Compare if a key in a row has changed */
|
|
|
|
int ha_ndbcluster::key_cmp(uint keynr, const byte * old_row,
|
|
const byte * new_row)
|
|
{
|
|
KEY_PART_INFO *key_part=table->key_info[keynr].key_part;
|
|
KEY_PART_INFO *end=key_part+table->key_info[keynr].key_parts;
|
|
|
|
for (; key_part != end ; key_part++)
|
|
{
|
|
if (key_part->null_bit)
|
|
{
|
|
if ((old_row[key_part->null_offset] & key_part->null_bit) !=
|
|
(new_row[key_part->null_offset] & key_part->null_bit))
|
|
return 1;
|
|
}
|
|
if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH))
|
|
{
|
|
|
|
if (key_part->field->cmp_binary((char*) (old_row + key_part->offset),
|
|
(char*) (new_row + key_part->offset),
|
|
(ulong) key_part->length))
|
|
return 1;
|
|
}
|
|
else
|
|
{
|
|
if (memcmp(old_row+key_part->offset, new_row+key_part->offset,
|
|
key_part->length))
|
|
return 1;
|
|
}
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
/*
|
|
Update one record in NDB using primary key
|
|
*/
|
|
|
|
int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
|
|
{
|
|
THD *thd= current_thd;
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbOperation *op;
|
|
uint i;
|
|
DBUG_ENTER("update_row");
|
|
|
|
statistic_increment(ha_update_count,&LOCK_status);
|
|
if (table->timestamp_on_update_now)
|
|
update_timestamp(new_data+table->timestamp_on_update_now-1);
|
|
|
|
if (!(op= trans->getNdbOperation(m_tabname)) ||
|
|
op->updateTuple() != 0)
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
// This table has no primary key, use "hidden" primary key
|
|
DBUG_PRINT("info", ("Using hidden key"));
|
|
|
|
// Require that the PK for this record has previously been
|
|
// read into m_value
|
|
uint no_fields= table->fields;
|
|
NdbRecAttr* rec= m_value[no_fields];
|
|
DBUG_ASSERT(rec);
|
|
DBUG_DUMP("key", (char*)rec->aRef(), NDB_HIDDEN_PRIMARY_KEY_LENGTH);
|
|
|
|
if (set_hidden_key(op, no_fields, rec->aRef()))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
else
|
|
{
|
|
/* Check for update of primary key and return error */
|
|
if (key_cmp(table->primary_key, old_data, new_data))
|
|
DBUG_RETURN(HA_ERR_UNSUPPORTED);
|
|
|
|
int res;
|
|
if ((res= set_primary_key(op, old_data + table->null_bytes)))
|
|
DBUG_RETURN(res);
|
|
}
|
|
|
|
// Set non-key attribute(s)
|
|
for (i= 0; i < table->fields; i++)
|
|
{
|
|
|
|
Field *field= table->field[i];
|
|
if ((thd->query_id == field->query_id) &&
|
|
(!(field->flags & PRI_KEY_FLAG)) &&
|
|
set_ndb_value(op, field, i))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
|
|
// Execute update operation
|
|
if (trans->execute(NoCommit) != 0)
|
|
DBUG_RETURN(ndb_err(trans));
|
|
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Delete one record from NDB, using primary key
|
|
*/
|
|
|
|
int ha_ndbcluster::delete_row(const byte *record)
|
|
{
|
|
NdbConnection *trans= m_active_trans;
|
|
NdbOperation *op;
|
|
DBUG_ENTER("delete_row");
|
|
|
|
statistic_increment(ha_delete_count,&LOCK_status);
|
|
|
|
if (!(op=trans->getNdbOperation(m_tabname)) ||
|
|
op->deleteTuple() != 0)
|
|
ERR_RETURN(trans->getNdbError());
|
|
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
// This table has no primary key, use "hidden" primary key
|
|
DBUG_PRINT("info", ("Using hidden key"));
|
|
uint no_fields= table->fields;
|
|
NdbRecAttr* rec= m_value[no_fields];
|
|
DBUG_ASSERT(rec != NULL);
|
|
|
|
if (set_hidden_key(op, no_fields, rec->aRef()))
|
|
ERR_RETURN(op->getNdbError());
|
|
}
|
|
else
|
|
{
|
|
int res;
|
|
if ((res= set_primary_key(op)))
|
|
return res;
|
|
}
|
|
|
|
// Execute delete operation
|
|
if (trans->execute(NoCommit) != 0)
|
|
DBUG_RETURN(ndb_err(trans));
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
/*
|
|
Unpack a record read from NDB
|
|
|
|
SYNOPSIS
|
|
unpack_record()
|
|
buf Buffer to store read row
|
|
|
|
NOTE
|
|
The data for each row is read directly into the
|
|
destination buffer. This function is primarily
|
|
called in order to check if any fields should be
|
|
set to null.
|
|
*/
|
|
|
|
void ha_ndbcluster::unpack_record(byte* buf)
|
|
{
|
|
uint row_offset= (uint) (buf - table->record[0]);
|
|
Field **field, **end;
|
|
NdbRecAttr **value= m_value;
|
|
DBUG_ENTER("unpack_record");
|
|
|
|
// Set null flag(s)
|
|
bzero(buf, table->null_bytes);
|
|
for (field= table->field, end= field+table->fields;
|
|
field < end;
|
|
field++, value++)
|
|
{
|
|
if (*value && (*value)->isNULL())
|
|
(*field)->set_null(row_offset);
|
|
}
|
|
|
|
#ifndef DBUG_OFF
|
|
// Read and print all values that was fetched
|
|
if (table->primary_key == MAX_KEY)
|
|
{
|
|
// Table with hidden primary key
|
|
int hidden_no= table->fields;
|
|
const NDBTAB *tab= (NDBTAB *) m_table;
|
|
const NDBCOL *hidden_col= tab->getColumn(hidden_no);
|
|
NdbRecAttr* rec= m_value[hidden_no];
|
|
DBUG_ASSERT(rec);
|
|
DBUG_PRINT("hidden", ("%d: %s \"%llu\"", hidden_no,
|
|
hidden_col->getName(), rec->u_64_value()));
|
|
}
|
|
print_results();
|
|
#endif
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Utility function to print/dump the fetched field
|
|
*/
|
|
|
|
void ha_ndbcluster::print_results()
|
|
{
|
|
const NDBTAB *tab= (NDBTAB*) m_table;
|
|
DBUG_ENTER("print_results");
|
|
|
|
#ifndef DBUG_OFF
|
|
if (!_db_on_)
|
|
DBUG_VOID_RETURN;
|
|
|
|
for (uint f=0; f<table->fields;f++)
|
|
{
|
|
Field *field;
|
|
const NDBCOL *col;
|
|
NdbRecAttr *value;
|
|
|
|
if (!(value= m_value[f]))
|
|
{
|
|
fprintf(DBUG_FILE, "Field %d was not read\n", f);
|
|
continue;
|
|
}
|
|
field= table->field[f];
|
|
DBUG_DUMP("field->ptr", (char*)field->ptr, field->pack_length());
|
|
col= tab->getColumn(f);
|
|
fprintf(DBUG_FILE, "%d: %s\t", f, col->getName());
|
|
|
|
if (value->isNULL())
|
|
{
|
|
fprintf(DBUG_FILE, "NULL\n");
|
|
continue;
|
|
}
|
|
|
|
switch (col->getType()) {
|
|
case NdbDictionary::Column::Blob:
|
|
case NdbDictionary::Column::Undefined:
|
|
fprintf(DBUG_FILE, "Unknown type: %d", col->getType());
|
|
break;
|
|
case NdbDictionary::Column::Tinyint: {
|
|
char value= *field->ptr;
|
|
fprintf(DBUG_FILE, "Tinyint\t%d", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Tinyunsigned: {
|
|
unsigned char value= *field->ptr;
|
|
fprintf(DBUG_FILE, "Tinyunsigned\t%u", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Smallint: {
|
|
short value= *field->ptr;
|
|
fprintf(DBUG_FILE, "Smallint\t%d", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Smallunsigned: {
|
|
unsigned short value= *field->ptr;
|
|
fprintf(DBUG_FILE, "Smallunsigned\t%u", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Mediumint: {
|
|
byte value[3];
|
|
memcpy(value, field->ptr, 3);
|
|
fprintf(DBUG_FILE, "Mediumint\t%d,%d,%d", value[0], value[1], value[2]);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Mediumunsigned: {
|
|
byte value[3];
|
|
memcpy(value, field->ptr, 3);
|
|
fprintf(DBUG_FILE, "Mediumunsigned\t%u,%u,%u", value[0], value[1], value[2]);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Int: {
|
|
fprintf(DBUG_FILE, "Int\t%lld", field->val_int());
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Unsigned: {
|
|
Uint32 value= (Uint32) *field->ptr;
|
|
fprintf(DBUG_FILE, "Unsigned\t%u", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Bigint: {
|
|
Int64 value= (Int64) *field->ptr;
|
|
fprintf(DBUG_FILE, "Bigint\t%lld", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Bigunsigned: {
|
|
Uint64 value= (Uint64) *field->ptr;
|
|
fprintf(DBUG_FILE, "Bigunsigned\t%llu", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Float: {
|
|
float value= (float) *field->ptr;
|
|
fprintf(DBUG_FILE, "Float\t%f", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Double: {
|
|
double value= (double) *field->ptr;
|
|
fprintf(DBUG_FILE, "Double\t%f", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Decimal: {
|
|
char *value= field->ptr;
|
|
|
|
fprintf(DBUG_FILE, "Decimal\t'%-*s'", field->pack_length(), value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Char:{
|
|
char buf[field->pack_length()+1];
|
|
char *value= (char *) field->ptr;
|
|
snprintf(buf, field->pack_length(), "%s", value);
|
|
fprintf(DBUG_FILE, "Char\t'%s'", buf);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Varchar:
|
|
case NdbDictionary::Column::Binary:
|
|
case NdbDictionary::Column::Varbinary: {
|
|
char *value= (char *) field->ptr;
|
|
fprintf(DBUG_FILE, "'%s'", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Datetime: {
|
|
Uint64 value= (Uint64) *field->ptr;
|
|
fprintf(DBUG_FILE, "Datetime\t%llu", value);
|
|
break;
|
|
}
|
|
case NdbDictionary::Column::Timespec: {
|
|
Uint64 value= (Uint64) *field->ptr;
|
|
fprintf(DBUG_FILE, "Timespec\t%llu", value);
|
|
break;
|
|
}
|
|
}
|
|
fprintf(DBUG_FILE, "\n");
|
|
|
|
}
|
|
#endif
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_init(uint index)
|
|
{
|
|
DBUG_ENTER("index_init");
|
|
DBUG_PRINT("enter", ("index: %u", index));
|
|
DBUG_RETURN(handler::index_init(index));
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_end()
|
|
{
|
|
DBUG_ENTER("index_end");
|
|
DBUG_RETURN(rnd_end());
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_read(byte *buf,
|
|
const byte *key, uint key_len,
|
|
enum ha_rkey_function find_flag)
|
|
{
|
|
DBUG_ENTER("index_read");
|
|
DBUG_PRINT("enter", ("active_index: %u, key_len: %u, find_flag: %d",
|
|
active_index, key_len, find_flag));
|
|
|
|
int error= 1;
|
|
statistic_increment(ha_read_key_count, &LOCK_status);
|
|
|
|
switch (get_index_type(active_index)){
|
|
case PRIMARY_KEY_INDEX:
|
|
error= pk_read(key, key_len, buf);
|
|
break;
|
|
|
|
case UNIQUE_INDEX:
|
|
error= unique_index_read(key, key_len, buf);
|
|
break;
|
|
|
|
case ORDERED_INDEX:
|
|
error= ordered_index_scan(key, key_len, buf, find_flag);
|
|
break;
|
|
|
|
default:
|
|
case UNDEFINED_INDEX:
|
|
break;
|
|
}
|
|
DBUG_RETURN(error);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_read_idx(byte *buf, uint index_no,
|
|
const byte *key, uint key_len,
|
|
enum ha_rkey_function find_flag)
|
|
{
|
|
statistic_increment(ha_read_key_count,&LOCK_status);
|
|
DBUG_ENTER("index_read_idx");
|
|
DBUG_PRINT("enter", ("index_no: %u, key_len: %u", index_no, key_len));
|
|
index_init(index_no);
|
|
DBUG_RETURN(index_read(buf, key, key_len, find_flag));
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_next(byte *buf)
|
|
{
|
|
DBUG_ENTER("index_next");
|
|
|
|
int error = 1;
|
|
statistic_increment(ha_read_next_count,&LOCK_status);
|
|
if (get_index_type(active_index) == PRIMARY_KEY_INDEX)
|
|
error= HA_ERR_END_OF_FILE;
|
|
else
|
|
error = next_result(buf);
|
|
DBUG_RETURN(error);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_prev(byte *buf)
|
|
{
|
|
DBUG_ENTER("index_prev");
|
|
statistic_increment(ha_read_prev_count,&LOCK_status);
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_first(byte *buf)
|
|
{
|
|
DBUG_ENTER("index_first");
|
|
statistic_increment(ha_read_first_count,&LOCK_status);
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::index_last(byte *buf)
|
|
{
|
|
DBUG_ENTER("index_last");
|
|
statistic_increment(ha_read_last_count,&LOCK_status);
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::rnd_init(bool scan)
|
|
{
|
|
NdbResultSet *cursor= m_active_cursor;
|
|
DBUG_ENTER("rnd_init");
|
|
DBUG_PRINT("enter", ("scan: %d", scan));
|
|
// Check that cursor is not defined
|
|
if (cursor)
|
|
DBUG_RETURN(1);
|
|
index_init(table->primary_key);
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::rnd_end()
|
|
{
|
|
NdbResultSet *cursor= m_active_cursor;
|
|
DBUG_ENTER("rnd_end");
|
|
|
|
if (cursor)
|
|
{
|
|
DBUG_PRINT("info", ("Closing the cursor"));
|
|
cursor->close();
|
|
m_active_cursor= NULL;
|
|
}
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::rnd_next(byte *buf)
|
|
{
|
|
DBUG_ENTER("rnd_next");
|
|
statistic_increment(ha_read_rnd_next_count, &LOCK_status);
|
|
int error = 1;
|
|
if (!m_active_cursor)
|
|
error = full_table_scan(buf);
|
|
else
|
|
error = next_result(buf);
|
|
DBUG_RETURN(error);
|
|
}
|
|
|
|
|
|
/*
|
|
An "interesting" record has been found and it's pk
|
|
retrieved by calling position
|
|
Now it's time to read the record from db once
|
|
again
|
|
*/
|
|
|
|
int ha_ndbcluster::rnd_pos(byte *buf, byte *pos)
|
|
{
|
|
DBUG_ENTER("rnd_pos");
|
|
statistic_increment(ha_read_rnd_count,&LOCK_status);
|
|
// The primary key for the record is stored in pos
|
|
// Perform a pk_read using primary key "index"
|
|
DBUG_RETURN(pk_read(pos, ref_length, buf));
|
|
}
|
|
|
|
|
|
/*
|
|
Store the primary key of this record in ref
|
|
variable, so that the row can be retrieved again later
|
|
using "reference" in rnd_pos
|
|
*/
|
|
|
|
void ha_ndbcluster::position(const byte *record)
|
|
{
|
|
KEY *key_info;
|
|
KEY_PART_INFO *key_part;
|
|
KEY_PART_INFO *end;
|
|
byte *buff;
|
|
DBUG_ENTER("position");
|
|
|
|
if (table->primary_key != MAX_KEY)
|
|
{
|
|
key_info= table->key_info + table->primary_key;
|
|
key_part= key_info->key_part;
|
|
end= key_part + key_info->key_parts;
|
|
buff= ref;
|
|
|
|
for (; key_part != end; key_part++)
|
|
{
|
|
if (key_part->null_bit) {
|
|
/* Store 0 if the key part is a NULL part */
|
|
if (record[key_part->null_offset]
|
|
& key_part->null_bit) {
|
|
*buff++= 1;
|
|
continue;
|
|
}
|
|
*buff++= 0;
|
|
}
|
|
memcpy(buff, record + key_part->offset, key_part->length);
|
|
buff += key_part->length;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
// No primary key, get hidden key
|
|
DBUG_PRINT("info", ("Getting hidden key"));
|
|
int hidden_no= table->fields;
|
|
NdbRecAttr* rec= m_value[hidden_no];
|
|
const NDBTAB *tab= (NDBTAB *) m_table;
|
|
const NDBCOL *hidden_col= tab->getColumn(hidden_no);
|
|
DBUG_ASSERT(hidden_col->getPrimaryKey() &&
|
|
hidden_col->getAutoIncrement() &&
|
|
rec != NULL &&
|
|
ref_length == NDB_HIDDEN_PRIMARY_KEY_LENGTH);
|
|
memcpy(ref, (const void*)rec->aRef(), ref_length);
|
|
}
|
|
|
|
DBUG_DUMP("ref", (char*)ref, ref_length);
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
void ha_ndbcluster::info(uint flag)
|
|
{
|
|
DBUG_ENTER("info");
|
|
DBUG_PRINT("enter", ("flag: %d", flag));
|
|
|
|
if (flag & HA_STATUS_POS)
|
|
DBUG_PRINT("info", ("HA_STATUS_POS"));
|
|
if (flag & HA_STATUS_NO_LOCK)
|
|
DBUG_PRINT("info", ("HA_STATUS_NO_LOCK"));
|
|
if (flag & HA_STATUS_TIME)
|
|
DBUG_PRINT("info", ("HA_STATUS_TIME"));
|
|
if (flag & HA_STATUS_CONST)
|
|
DBUG_PRINT("info", ("HA_STATUS_CONST"));
|
|
if (flag & HA_STATUS_VARIABLE)
|
|
DBUG_PRINT("info", ("HA_STATUS_VARIABLE"));
|
|
if (flag & HA_STATUS_ERRKEY)
|
|
DBUG_PRINT("info", ("HA_STATUS_ERRKEY"));
|
|
if (flag & HA_STATUS_AUTO)
|
|
DBUG_PRINT("info", ("HA_STATUS_AUTO"));
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::extra(enum ha_extra_function operation)
|
|
{
|
|
DBUG_ENTER("extra");
|
|
switch (operation) {
|
|
case HA_EXTRA_NORMAL: /* Optimize for space (def) */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NORMAL"));
|
|
break;
|
|
case HA_EXTRA_QUICK: /* Optimize for speed */
|
|
DBUG_PRINT("info", ("HA_EXTRA_QUICK"));
|
|
break;
|
|
case HA_EXTRA_RESET: /* Reset database to after open */
|
|
DBUG_PRINT("info", ("HA_EXTRA_RESET"));
|
|
break;
|
|
case HA_EXTRA_CACHE: /* Cash record in HA_rrnd() */
|
|
DBUG_PRINT("info", ("HA_EXTRA_CACHE"));
|
|
break;
|
|
case HA_EXTRA_NO_CACHE: /* End cacheing of records (def) */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_CACHE"));
|
|
break;
|
|
case HA_EXTRA_NO_READCHECK: /* No readcheck on update */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_READCHECK"));
|
|
break;
|
|
case HA_EXTRA_READCHECK: /* Use readcheck (def) */
|
|
DBUG_PRINT("info", ("HA_EXTRA_READCHECK"));
|
|
break;
|
|
case HA_EXTRA_KEYREAD: /* Read only key to database */
|
|
DBUG_PRINT("info", ("HA_EXTRA_KEYREAD"));
|
|
break;
|
|
case HA_EXTRA_NO_KEYREAD: /* Normal read of records (def) */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_KEYREAD"));
|
|
break;
|
|
case HA_EXTRA_NO_USER_CHANGE: /* No user is allowed to write */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_USER_CHANGE"));
|
|
break;
|
|
case HA_EXTRA_KEY_CACHE:
|
|
DBUG_PRINT("info", ("HA_EXTRA_KEY_CACHE"));
|
|
break;
|
|
case HA_EXTRA_NO_KEY_CACHE:
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_KEY_CACHE"));
|
|
break;
|
|
case HA_EXTRA_WAIT_LOCK: /* Wait until file is avalably (def) */
|
|
DBUG_PRINT("info", ("HA_EXTRA_WAIT_LOCK"));
|
|
break;
|
|
case HA_EXTRA_NO_WAIT_LOCK: /* If file is locked, return quickly */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_WAIT_LOCK"));
|
|
break;
|
|
case HA_EXTRA_WRITE_CACHE: /* Use write cache in ha_write() */
|
|
DBUG_PRINT("info", ("HA_EXTRA_WRITE_CACHE"));
|
|
break;
|
|
case HA_EXTRA_FLUSH_CACHE: /* flush write_record_cache */
|
|
DBUG_PRINT("info", ("HA_EXTRA_FLUSH_CACHE"));
|
|
break;
|
|
case HA_EXTRA_NO_KEYS: /* Remove all update of keys */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_KEYS"));
|
|
break;
|
|
case HA_EXTRA_KEYREAD_CHANGE_POS: /* Keyread, but change pos */
|
|
DBUG_PRINT("info", ("HA_EXTRA_KEYREAD_CHANGE_POS")); /* xxxxchk -r must be used */
|
|
break;
|
|
case HA_EXTRA_REMEMBER_POS: /* Remember pos for next/prev */
|
|
DBUG_PRINT("info", ("HA_EXTRA_REMEMBER_POS"));
|
|
break;
|
|
case HA_EXTRA_RESTORE_POS:
|
|
DBUG_PRINT("info", ("HA_EXTRA_RESTORE_POS"));
|
|
break;
|
|
case HA_EXTRA_REINIT_CACHE: /* init cache from current record */
|
|
DBUG_PRINT("info", ("HA_EXTRA_REINIT_CACHE"));
|
|
break;
|
|
case HA_EXTRA_FORCE_REOPEN: /* Datafile have changed on disk */
|
|
DBUG_PRINT("info", ("HA_EXTRA_FORCE_REOPEN"));
|
|
break;
|
|
case HA_EXTRA_FLUSH: /* Flush tables to disk */
|
|
DBUG_PRINT("info", ("HA_EXTRA_FLUSH"));
|
|
break;
|
|
case HA_EXTRA_NO_ROWS: /* Don't write rows */
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_ROWS"));
|
|
break;
|
|
case HA_EXTRA_RESET_STATE: /* Reset positions */
|
|
DBUG_PRINT("info", ("HA_EXTRA_RESET_STATE"));
|
|
break;
|
|
case HA_EXTRA_IGNORE_DUP_KEY: /* Dup keys don't rollback everything*/
|
|
DBUG_PRINT("info", ("HA_EXTRA_IGNORE_DUP_KEY"));
|
|
|
|
DBUG_PRINT("info", ("Turning ON use of write instead of insert"));
|
|
m_use_write= TRUE;
|
|
break;
|
|
case HA_EXTRA_NO_IGNORE_DUP_KEY:
|
|
DBUG_PRINT("info", ("HA_EXTRA_NO_IGNORE_DUP_KEY"));
|
|
DBUG_PRINT("info", ("Turning OFF use of write instead of insert"));
|
|
m_use_write= false;
|
|
break;
|
|
case HA_EXTRA_RETRIEVE_ALL_COLS: /* Retrieve all columns, not just those
|
|
where field->query_id is the same as
|
|
the current query id */
|
|
DBUG_PRINT("info", ("HA_EXTRA_RETRIEVE_ALL_COLS"));
|
|
break;
|
|
case HA_EXTRA_PREPARE_FOR_DELETE:
|
|
DBUG_PRINT("info", ("HA_EXTRA_PREPARE_FOR_DELETE"));
|
|
break;
|
|
case HA_EXTRA_PREPARE_FOR_UPDATE: /* Remove read cache if problems */
|
|
DBUG_PRINT("info", ("HA_EXTRA_PREPARE_FOR_UPDATE"));
|
|
break;
|
|
case HA_EXTRA_PRELOAD_BUFFER_SIZE:
|
|
DBUG_PRINT("info", ("HA_EXTRA_PRELOAD_BUFFER_SIZE"));
|
|
break;
|
|
case HA_EXTRA_RETRIEVE_PRIMARY_KEY:
|
|
DBUG_PRINT("info", ("HA_EXTRA_RETRIEVE_PRIMARY_KEY"));
|
|
break;
|
|
case HA_EXTRA_CHANGE_KEY_TO_UNIQUE:
|
|
DBUG_PRINT("info", ("HA_EXTRA_CHANGE_KEY_TO_UNIQUE"));
|
|
break;
|
|
case HA_EXTRA_CHANGE_KEY_TO_DUP:
|
|
DBUG_PRINT("info", ("HA_EXTRA_CHANGE_KEY_TO_DUP"));
|
|
break;
|
|
|
|
}
|
|
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::extra_opt(enum ha_extra_function operation, ulong cache_size)
|
|
{
|
|
DBUG_ENTER("extra_opt");
|
|
DBUG_PRINT("enter", ("cache_size: %d", cache_size));
|
|
DBUG_RETURN(extra(operation));
|
|
}
|
|
|
|
|
|
int ha_ndbcluster::reset()
|
|
{
|
|
DBUG_ENTER("reset");
|
|
// Reset what?
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
|
|
const char **ha_ndbcluster::bas_ext() const
|
|
{ static const char *ext[1] = { NullS }; return ext; }
|
|
|
|
|
|
/*
|
|
How many seeks it will take to read through the table
|
|
This is to be comparable to the number returned by records_in_range so
|
|
that we can decide if we should scan the table or use keys.
|
|
*/
|
|
|
|
double ha_ndbcluster::scan_time()
|
|
{
|
|
return rows2double(records/3);
|
|
}
|
|
|
|
|
|
THR_LOCK_DATA **ha_ndbcluster::store_lock(THD *thd,
|
|
THR_LOCK_DATA **to,
|
|
enum thr_lock_type lock_type)
|
|
{
|
|
DBUG_ENTER("store_lock");
|
|
|
|
if (lock_type != TL_IGNORE && m_lock.type == TL_UNLOCK)
|
|
{
|
|
|
|
/* If we are not doing a LOCK TABLE, then allow multiple
|
|
writers */
|
|
|
|
if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
|
|
lock_type <= TL_WRITE) && !thd->in_lock_tables)
|
|
lock_type= TL_WRITE_ALLOW_WRITE;
|
|
|
|
/* In queries of type INSERT INTO t1 SELECT ... FROM t2 ...
|
|
MySQL would use the lock TL_READ_NO_INSERT on t2, and that
|
|
would conflict with TL_WRITE_ALLOW_WRITE, blocking all inserts
|
|
to t2. Convert the lock to a normal read lock to allow
|
|
concurrent inserts to t2. */
|
|
|
|
if (lock_type == TL_READ_NO_INSERT && !thd->in_lock_tables)
|
|
lock_type= TL_READ;
|
|
|
|
m_lock.type=lock_type;
|
|
}
|
|
*to++= &m_lock;
|
|
|
|
DBUG_RETURN(to);
|
|
}
|
|
|
|
#ifndef DBUG_OFF
|
|
#define PRINT_OPTION_FLAGS(t) { \
|
|
if (t->options & OPTION_NOT_AUTOCOMMIT) \
|
|
DBUG_PRINT("thd->options", ("OPTION_NOT_AUTOCOMMIT")); \
|
|
if (t->options & OPTION_BEGIN) \
|
|
DBUG_PRINT("thd->options", ("OPTION_BEGIN")); \
|
|
if (t->options & OPTION_TABLE_LOCK) \
|
|
DBUG_PRINT("thd->options", ("OPTION_TABLE_LOCK")); \
|
|
}
|
|
#else
|
|
#define PRINT_OPTION_FLAGS(t)
|
|
#endif
|
|
|
|
|
|
/*
|
|
As MySQL will execute an external lock for every new table it uses
|
|
we can use this to start the transactions.
|
|
If we are in auto_commit mode we just need to start a transaction
|
|
for the statement, this will be stored in transaction.stmt.
|
|
If not, we have to start a master transaction if there doesn't exist
|
|
one from before, this will be stored in transaction.all
|
|
|
|
When a table lock is held one transaction will be started which holds
|
|
the table lock and for each statement a hupp transaction will be started
|
|
*/
|
|
|
|
int ha_ndbcluster::external_lock(THD *thd, int lock_type)
|
|
{
|
|
int error=0;
|
|
NdbConnection* trans= NULL;
|
|
|
|
DBUG_ENTER("external_lock");
|
|
DBUG_PRINT("enter", ("transaction.ndb_lock_count: %d",
|
|
thd->transaction.ndb_lock_count));
|
|
|
|
/*
|
|
Check that this handler instance has a connection
|
|
set up to the Ndb object of thd
|
|
*/
|
|
if (check_ndb_connection())
|
|
DBUG_RETURN(1);
|
|
|
|
if (lock_type != F_UNLCK)
|
|
{
|
|
if (!thd->transaction.ndb_lock_count++)
|
|
{
|
|
PRINT_OPTION_FLAGS(thd);
|
|
|
|
if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN | OPTION_TABLE_LOCK)))
|
|
{
|
|
// Autocommit transaction
|
|
DBUG_ASSERT(!thd->transaction.stmt.ndb_tid);
|
|
DBUG_PRINT("trans",("Starting transaction stmt"));
|
|
|
|
trans= m_ndb->startTransaction();
|
|
if (trans == NULL)
|
|
{
|
|
thd->transaction.ndb_lock_count--; // We didn't get the lock
|
|
ERR_RETURN(m_ndb->getNdbError());
|
|
}
|
|
thd->transaction.stmt.ndb_tid= trans;
|
|
}
|
|
else
|
|
{
|
|
if (!thd->transaction.all.ndb_tid)
|
|
{
|
|
// Not autocommit transaction
|
|
// A "master" transaction ha not been started yet
|
|
DBUG_PRINT("trans",("starting transaction, all"));
|
|
|
|
trans= m_ndb->startTransaction();
|
|
if (trans == NULL)
|
|
{
|
|
thd->transaction.ndb_lock_count--; // We didn't get the lock
|
|
ERR_RETURN(m_ndb->getNdbError());
|
|
}
|
|
|
|
/*
|
|
If this is the start of a LOCK TABLE, a table look
|
|
should be taken on the table in NDB
|
|
|
|
Check if it should be read or write lock
|
|
*/
|
|
if (thd->options & (OPTION_TABLE_LOCK))
|
|
{
|
|
//lockThisTable();
|
|
DBUG_PRINT("info", ("Locking the table..." ));
|
|
}
|
|
|
|
thd->transaction.all.ndb_tid= trans;
|
|
}
|
|
}
|
|
}
|
|
/*
|
|
This is the place to make sure this handler instance
|
|
has a started transaction.
|
|
|
|
The transaction is started by the first handler on which
|
|
MySQL Server calls external lock
|
|
|
|
Other handlers in the same stmt or transaction should use
|
|
the same NDB transaction. This is done by setting up the m_active_trans
|
|
pointer to point to the NDB transaction.
|
|
*/
|
|
|
|
m_active_trans= thd->transaction.all.ndb_tid ?
|
|
(NdbConnection*)thd->transaction.all.ndb_tid:
|
|
(NdbConnection*)thd->transaction.stmt.ndb_tid;
|
|
DBUG_ASSERT(m_active_trans);
|
|
|
|
}
|
|
else
|
|
{
|
|
if (!--thd->transaction.ndb_lock_count)
|
|
{
|
|
DBUG_PRINT("trans", ("Last external_lock"));
|
|
PRINT_OPTION_FLAGS(thd);
|
|
|
|
if (thd->transaction.stmt.ndb_tid)
|
|
{
|
|
/*
|
|
Unlock is done without a transaction commit / rollback.
|
|
This happens if the thread didn't update any rows
|
|
We must in this case close the transaction to release resources
|
|
*/
|
|
DBUG_PRINT("trans",("ending non-updating transaction"));
|
|
m_ndb->closeTransaction(m_active_trans);
|
|
thd->transaction.stmt.ndb_tid= 0;
|
|
}
|
|
}
|
|
m_active_trans= NULL;
|
|
}
|
|
DBUG_RETURN(error);
|
|
}
|
|
|
|
/*
|
|
When using LOCK TABLE's external_lock is only called when the actual
|
|
TABLE LOCK is done.
|
|
Under LOCK TABLES, each used tables will force a call to start_stmt.
|
|
*/
|
|
|
|
int ha_ndbcluster::start_stmt(THD *thd)
|
|
{
|
|
int error=0;
|
|
DBUG_ENTER("start_stmt");
|
|
PRINT_OPTION_FLAGS(thd);
|
|
|
|
NdbConnection *trans= (NdbConnection*)thd->transaction.stmt.ndb_tid;
|
|
if (!trans){
|
|
DBUG_PRINT("trans",("Starting transaction stmt"));
|
|
|
|
NdbConnection *tablock_trans=
|
|
(NdbConnection*)thd->transaction.all.ndb_tid;
|
|
DBUG_PRINT("info", ("tablock_trans: %x", tablock_trans));
|
|
DBUG_ASSERT(tablock_trans); trans= m_ndb->hupp(tablock_trans);
|
|
if (trans == NULL)
|
|
ERR_RETURN(m_ndb->getNdbError());
|
|
thd->transaction.stmt.ndb_tid= trans;
|
|
}
|
|
m_active_trans= trans;
|
|
|
|
DBUG_RETURN(error);
|
|
}
|
|
|
|
|
|
/*
|
|
Commit a transaction started in NDB
|
|
*/
|
|
|
|
int ndbcluster_commit(THD *thd, void *ndb_transaction)
|
|
{
|
|
int res= 0;
|
|
Ndb *ndb= (Ndb*)thd->transaction.ndb;
|
|
NdbConnection *trans= (NdbConnection*)ndb_transaction;
|
|
|
|
DBUG_ENTER("ndbcluster_commit");
|
|
DBUG_PRINT("transaction",("%s",
|
|
trans == thd->transaction.stmt.ndb_tid ?
|
|
"stmt" : "all"));
|
|
DBUG_ASSERT(ndb && trans);
|
|
|
|
if (trans->execute(Commit) != 0)
|
|
{
|
|
const NdbError err= trans->getNdbError();
|
|
ERR_PRINT(err);
|
|
res= ndb_to_mysql_error(&err);
|
|
}
|
|
ndb->closeTransaction(trans);
|
|
DBUG_RETURN(res);
|
|
}
|
|
|
|
|
|
/*
|
|
Rollback a transaction started in NDB
|
|
*/
|
|
|
|
int ndbcluster_rollback(THD *thd, void *ndb_transaction)
|
|
{
|
|
int res= 0;
|
|
Ndb *ndb= (Ndb*)thd->transaction.ndb;
|
|
NdbConnection *trans= (NdbConnection*)ndb_transaction;
|
|
|
|
DBUG_ENTER("ndbcluster_rollback");
|
|
DBUG_PRINT("transaction",("%s",
|
|
trans == thd->transaction.stmt.ndb_tid ?
|
|
"stmt" : "all"));
|
|
DBUG_ASSERT(ndb && trans);
|
|
|
|
if (trans->execute(Rollback) != 0)
|
|
{
|
|
const NdbError err= trans->getNdbError();
|
|
ERR_PRINT(err);
|
|
res= ndb_to_mysql_error(&err);
|
|
}
|
|
ndb->closeTransaction(trans);
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Map MySQL type to the corresponding NDB type
|
|
*/
|
|
|
|
inline NdbDictionary::Column::Type
|
|
mysql_to_ndb_type(enum enum_field_types mysql_type, bool unsigned_flg)
|
|
{
|
|
switch(mysql_type) {
|
|
case MYSQL_TYPE_DECIMAL:
|
|
return NdbDictionary::Column::Char;
|
|
case MYSQL_TYPE_TINY:
|
|
return (unsigned_flg) ?
|
|
NdbDictionary::Column::Tinyunsigned :
|
|
NdbDictionary::Column::Tinyint;
|
|
case MYSQL_TYPE_SHORT:
|
|
return (unsigned_flg) ?
|
|
NdbDictionary::Column::Smallunsigned :
|
|
NdbDictionary::Column::Smallint;
|
|
case MYSQL_TYPE_LONG:
|
|
return (unsigned_flg) ?
|
|
NdbDictionary::Column::Unsigned :
|
|
NdbDictionary::Column::Int;
|
|
case MYSQL_TYPE_TIMESTAMP:
|
|
return NdbDictionary::Column::Unsigned;
|
|
case MYSQL_TYPE_LONGLONG:
|
|
return (unsigned_flg) ?
|
|
NdbDictionary::Column::Bigunsigned :
|
|
NdbDictionary::Column::Bigint;
|
|
case MYSQL_TYPE_INT24:
|
|
return (unsigned_flg) ?
|
|
NdbDictionary::Column::Mediumunsigned :
|
|
NdbDictionary::Column::Mediumint;
|
|
break;
|
|
case MYSQL_TYPE_FLOAT:
|
|
return NdbDictionary::Column::Float;
|
|
case MYSQL_TYPE_DOUBLE:
|
|
return NdbDictionary::Column::Double;
|
|
case MYSQL_TYPE_DATETIME :
|
|
return NdbDictionary::Column::Datetime;
|
|
case MYSQL_TYPE_DATE :
|
|
case MYSQL_TYPE_NEWDATE :
|
|
case MYSQL_TYPE_TIME :
|
|
case MYSQL_TYPE_YEAR :
|
|
// Missing NDB data types, mapped to char
|
|
return NdbDictionary::Column::Char;
|
|
case MYSQL_TYPE_ENUM :
|
|
return NdbDictionary::Column::Char;
|
|
case MYSQL_TYPE_SET :
|
|
return NdbDictionary::Column::Char;
|
|
case MYSQL_TYPE_TINY_BLOB :
|
|
case MYSQL_TYPE_MEDIUM_BLOB :
|
|
case MYSQL_TYPE_LONG_BLOB :
|
|
case MYSQL_TYPE_BLOB :
|
|
return NdbDictionary::Column::Blob;
|
|
case MYSQL_TYPE_VAR_STRING :
|
|
return NdbDictionary::Column::Varchar;
|
|
case MYSQL_TYPE_STRING :
|
|
return NdbDictionary::Column::Char;
|
|
case MYSQL_TYPE_NULL :
|
|
case MYSQL_TYPE_GEOMETRY :
|
|
return NdbDictionary::Column::Undefined;
|
|
}
|
|
return NdbDictionary::Column::Undefined;
|
|
}
|
|
|
|
|
|
/*
|
|
Create a table in NDB Cluster
|
|
*/
|
|
|
|
int ha_ndbcluster::create(const char *name,
|
|
TABLE *form,
|
|
HA_CREATE_INFO *info)
|
|
{
|
|
NDBTAB tab;
|
|
NdbDictionary::Column::Type ndb_type;
|
|
NDBCOL col;
|
|
uint pack_length, length, i;
|
|
int res;
|
|
const void *data, *pack_data;
|
|
const char **key_name= form->keynames.type_names;
|
|
char name2[FN_HEADLEN];
|
|
|
|
DBUG_ENTER("create");
|
|
DBUG_PRINT("enter", ("name: %s", name));
|
|
fn_format(name2, name, "", "",2); // Remove the .frm extension
|
|
set_dbname(name2);
|
|
set_tabname(name2);
|
|
|
|
DBUG_PRINT("table", ("name: %s", m_tabname));
|
|
tab.setName(m_tabname);
|
|
tab.setLogging(!(info->options & HA_LEX_CREATE_TMP_TABLE));
|
|
|
|
// Save frm data for this table
|
|
if (readfrm(name, &data, &length))
|
|
DBUG_RETURN(1);
|
|
if (packfrm(data, length, &pack_data, &pack_length))
|
|
DBUG_RETURN(2);
|
|
|
|
DBUG_PRINT("info", ("setFrm data=%x, len=%d", pack_data, pack_length));
|
|
tab.setFrm(pack_data, pack_length);
|
|
my_free((char*)data, MYF(0));
|
|
my_free((char*)pack_data, MYF(0));
|
|
|
|
for (i= 0; i < form->fields; i++)
|
|
{
|
|
Field *field= form->field[i];
|
|
ndb_type= mysql_to_ndb_type(field->real_type(),
|
|
field->flags & UNSIGNED_FLAG);
|
|
DBUG_PRINT("info", ("name: %s, type: %u, pack_length: %d",
|
|
field->field_name, field->real_type(),
|
|
field->pack_length()));
|
|
col.setName(field->field_name);
|
|
col.setType(ndb_type);
|
|
if ((ndb_type == NdbDictionary::Column::Char) ||
|
|
(ndb_type == NdbDictionary::Column::Varchar))
|
|
col.setLength(field->pack_length());
|
|
else
|
|
col.setLength(1);
|
|
col.setNullable(field->maybe_null());
|
|
col.setPrimaryKey(field->flags & PRI_KEY_FLAG);
|
|
if (field->flags & AUTO_INCREMENT_FLAG)
|
|
{
|
|
DBUG_PRINT("info", ("Found auto_increment key"));
|
|
col.setAutoIncrement(TRUE);
|
|
ulonglong value = info->auto_increment_value ?
|
|
info->auto_increment_value -1 :
|
|
(ulonglong) 0;
|
|
DBUG_PRINT("info", ("initial value=%ld", value));
|
|
// col.setInitialAutIncValue(value);
|
|
}
|
|
else
|
|
col.setAutoIncrement(false);
|
|
|
|
tab.addColumn(col);
|
|
}
|
|
|
|
// No primary key, create shadow key as 64 bit, auto increment
|
|
if (form->primary_key == MAX_KEY)
|
|
{
|
|
DBUG_PRINT("info", ("Generating shadow key"));
|
|
col.setName("$PK");
|
|
col.setType(NdbDictionary::Column::Bigunsigned);
|
|
col.setLength(1);
|
|
col.setNullable(false);
|
|
col.setPrimaryKey(TRUE);
|
|
col.setAutoIncrement(TRUE);
|
|
tab.addColumn(col);
|
|
}
|
|
|
|
my_errno= 0;
|
|
if (check_ndb_connection())
|
|
{
|
|
my_errno= HA_ERR_NO_CONNECTION;
|
|
DBUG_RETURN(my_errno);
|
|
}
|
|
|
|
// Create the table in NDB
|
|
NDBDICT *dict= m_ndb->getDictionary();
|
|
if (dict->createTable(tab))
|
|
{
|
|
const NdbError err= dict->getNdbError();
|
|
ERR_PRINT(err);
|
|
my_errno= ndb_to_mysql_error(&err);
|
|
DBUG_RETURN(my_errno);
|
|
}
|
|
DBUG_PRINT("info", ("Table %s/%s created successfully",
|
|
m_dbname, m_tabname));
|
|
|
|
// Fetch table from NDB, check that it exists
|
|
const NDBTAB *tab2= dict->getTable(m_tabname);
|
|
if (tab2 == NULL)
|
|
{
|
|
const NdbError err= dict->getNdbError();
|
|
ERR_PRINT(err);
|
|
my_errno= ndb_to_mysql_error(&err);
|
|
DBUG_RETURN(my_errno);
|
|
}
|
|
|
|
// Create secondary indexes
|
|
for (i= 0; i < form->keys; i++)
|
|
{
|
|
DBUG_PRINT("info", ("Found index %u: %s", i, key_name[i]));
|
|
if (i == form->primary_key)
|
|
{
|
|
DBUG_PRINT("info", ("Skipping it, PK already created"));
|
|
continue;
|
|
}
|
|
|
|
DBUG_PRINT("info", ("Creating index %u: %s", i, key_name[i]));
|
|
res= create_index(key_name[i],
|
|
form->key_info + i);
|
|
switch(res){
|
|
case 0:
|
|
// OK
|
|
break;
|
|
default:
|
|
DBUG_PRINT("error", ("Failed to create index %u", i));
|
|
drop_table();
|
|
my_errno= res;
|
|
goto err_end;
|
|
}
|
|
}
|
|
|
|
err_end:
|
|
DBUG_RETURN(my_errno);
|
|
}
|
|
|
|
|
|
/*
|
|
Create an index in NDB Cluster
|
|
*/
|
|
|
|
int ha_ndbcluster::create_index(const char *name,
|
|
KEY *key_info){
|
|
NdbDictionary::Dictionary *dict= m_ndb->getDictionary();
|
|
KEY_PART_INFO *key_part= key_info->key_part;
|
|
KEY_PART_INFO *end= key_part + key_info->key_parts;
|
|
|
|
DBUG_ENTER("create_index");
|
|
DBUG_PRINT("enter", ("name: %s ", name));
|
|
|
|
// Check that an index with the same name do not already exist
|
|
if (dict->getIndex(name, m_tabname))
|
|
ERR_RETURN(dict->getNdbError());
|
|
|
|
NdbDictionary::Index ndb_index(name);
|
|
if (key_info->flags & HA_NOSAME)
|
|
ndb_index.setType(NdbDictionary::Index::UniqueHashIndex);
|
|
else
|
|
{
|
|
ndb_index.setType(NdbDictionary::Index::OrderedIndex);
|
|
// TODO Only temporary ordered indexes supported
|
|
ndb_index.setLogging(false);
|
|
}
|
|
ndb_index.setTable(m_tabname);
|
|
|
|
for (; key_part != end; key_part++)
|
|
{
|
|
Field *field= key_part->field;
|
|
DBUG_PRINT("info", ("attr: %s", field->field_name));
|
|
ndb_index.addColumnName(field->field_name);
|
|
}
|
|
|
|
if (dict->createIndex(ndb_index))
|
|
ERR_RETURN(dict->getNdbError());
|
|
|
|
// Success
|
|
DBUG_PRINT("info", ("Created index %s", name));
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Rename a table in NDB Cluster
|
|
*/
|
|
|
|
int ha_ndbcluster::rename_table(const char *from, const char *to)
|
|
{
|
|
char new_tabname[FN_HEADLEN];
|
|
|
|
DBUG_ENTER("ha_ndbcluster::rename_table");
|
|
set_dbname(from);
|
|
set_tabname(from);
|
|
set_tabname(to, new_tabname);
|
|
|
|
if (check_ndb_connection()) {
|
|
my_errno= HA_ERR_NO_CONNECTION;
|
|
DBUG_RETURN(my_errno);
|
|
}
|
|
|
|
int result= alter_table_name(m_tabname, new_tabname);
|
|
if (result == 0)
|
|
set_tabname(to);
|
|
|
|
DBUG_RETURN(result);
|
|
}
|
|
|
|
|
|
/*
|
|
Rename a table in NDB Cluster using alter table
|
|
*/
|
|
|
|
int ha_ndbcluster::alter_table_name(const char *from, const char *to)
|
|
{
|
|
NDBDICT *dict= m_ndb->getDictionary();
|
|
const NDBTAB *orig_tab;
|
|
DBUG_ENTER("alter_table_name_table");
|
|
DBUG_PRINT("enter", ("Renaming %s to %s", from, to));
|
|
|
|
if (!(orig_tab= dict->getTable(from)))
|
|
ERR_RETURN(dict->getNdbError());
|
|
|
|
NdbDictionary::Table copy_tab= dict->getTableForAlteration(from);
|
|
copy_tab.setName(to);
|
|
if (dict->alterTable(copy_tab) != 0)
|
|
ERR_RETURN(dict->getNdbError());
|
|
|
|
m_table= NULL;
|
|
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Delete a table from NDB Cluster
|
|
*/
|
|
|
|
int ha_ndbcluster::delete_table(const char *name)
|
|
{
|
|
DBUG_ENTER("delete_table");
|
|
DBUG_PRINT("enter", ("name: %s", name));
|
|
set_dbname(name);
|
|
set_tabname(name);
|
|
|
|
if (check_ndb_connection())
|
|
DBUG_RETURN(HA_ERR_NO_CONNECTION);
|
|
DBUG_RETURN(drop_table());
|
|
}
|
|
|
|
|
|
/*
|
|
Drop a table in NDB Cluster
|
|
*/
|
|
|
|
int ha_ndbcluster::drop_table()
|
|
{
|
|
NdbDictionary::Dictionary *dict= m_ndb->getDictionary();
|
|
|
|
DBUG_ENTER("drop_table");
|
|
DBUG_PRINT("enter", ("Deleting %s", m_tabname));
|
|
|
|
if (dict->dropTable(m_tabname))
|
|
{
|
|
const NdbError err= dict->getNdbError();
|
|
if (err.code == 709)
|
|
; // 709: No such table existed
|
|
else
|
|
ERR_RETURN(dict->getNdbError());
|
|
}
|
|
release_metadata();
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Drop a database in NDB Cluster
|
|
*/
|
|
|
|
int ndbcluster_drop_database(const char *path)
|
|
{
|
|
DBUG_ENTER("ndbcluster_drop_database");
|
|
// TODO drop all tables for this database
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
|
|
longlong ha_ndbcluster::get_auto_increment()
|
|
{
|
|
// NOTE If number of values to be inserted is known
|
|
// the autoincrement cache could be used here
|
|
Uint64 auto_value= m_ndb->getAutoIncrementValue(m_tabname);
|
|
return (longlong)auto_value;
|
|
}
|
|
|
|
|
|
/*
|
|
Constructor for the NDB Cluster table handler
|
|
*/
|
|
|
|
ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
|
|
handler(table_arg),
|
|
m_active_trans(NULL),
|
|
m_active_cursor(NULL),
|
|
m_ndb(NULL),
|
|
m_table(NULL),
|
|
m_table_flags(HA_REC_NOT_IN_SEQ |
|
|
HA_KEYPOS_TO_RNDPOS |
|
|
HA_NOT_EXACT_COUNT |
|
|
HA_NO_WRITE_DELAYED |
|
|
HA_NO_PREFIX_CHAR_KEYS |
|
|
HA_NO_BLOBS |
|
|
HA_DROP_BEFORE_CREATE |
|
|
HA_NOT_READ_AFTER_KEY),
|
|
m_use_write(false)
|
|
{
|
|
|
|
DBUG_ENTER("ha_ndbcluster");
|
|
|
|
m_tabname[0]= '\0';
|
|
m_dbname[0]= '\0';
|
|
|
|
// TODO Adjust number of records and other parameters for proper
|
|
// selection of scan/pk access
|
|
records= 100;
|
|
block_size= 1024;
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Destructor for NDB Cluster table handler
|
|
*/
|
|
|
|
ha_ndbcluster::~ha_ndbcluster()
|
|
{
|
|
DBUG_ENTER("~ha_ndbcluster");
|
|
|
|
release_metadata();
|
|
|
|
// Check for open cursor/transaction
|
|
DBUG_ASSERT(m_active_cursor == NULL);
|
|
DBUG_ASSERT(m_active_trans == NULL);
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Open a table for further use
|
|
- fetch metadata for this table from NDB
|
|
- check that table exists
|
|
*/
|
|
|
|
int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
|
|
{
|
|
KEY *key;
|
|
DBUG_ENTER("open");
|
|
DBUG_PRINT("enter", ("name: %s mode: %d test_if_locked: %d",
|
|
name, mode, test_if_locked));
|
|
|
|
// Setup ref_length to make room for the whole
|
|
// primary key to be written in the ref variable
|
|
|
|
if (table->primary_key != MAX_KEY)
|
|
{
|
|
key= table->key_info+table->primary_key;
|
|
ref_length= key->key_length;
|
|
DBUG_PRINT("info", (" ref_length: %d", ref_length));
|
|
}
|
|
// Init table lock structure
|
|
if (!(m_share=get_share(name)))
|
|
DBUG_RETURN(1);
|
|
thr_lock_data_init(&m_share->lock,&m_lock,(void*) 0);
|
|
|
|
set_dbname(name);
|
|
set_tabname(name);
|
|
|
|
if (check_ndb_connection())
|
|
DBUG_RETURN(HA_ERR_NO_CONNECTION);
|
|
|
|
DBUG_RETURN(get_metadata(name));
|
|
}
|
|
|
|
|
|
/*
|
|
Close the table
|
|
- release resources setup by open()
|
|
*/
|
|
|
|
int ha_ndbcluster::close(void)
|
|
{
|
|
DBUG_ENTER("close");
|
|
free_share(m_share);
|
|
release_metadata();
|
|
m_ndb= NULL;
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
Ndb* ha_ndbcluster::seize_ndb()
|
|
{
|
|
Ndb* ndb;
|
|
DBUG_ENTER("seize_ndb");
|
|
|
|
#ifdef USE_NDB_POOL
|
|
// Seize from pool
|
|
ndb= Ndb::seize();
|
|
#else
|
|
ndb= new Ndb("");
|
|
#endif
|
|
if (ndb->init(NDB_MAX_TRANSACTIONS) != 0)
|
|
{
|
|
ERR_PRINT(ndb->getNdbError());
|
|
/*
|
|
TODO
|
|
Alt.1 If init fails because to many allocated Ndb
|
|
wait on condition for a Ndb object to be released.
|
|
Alt.2 Seize/release from pool, wait until next release
|
|
*/
|
|
delete ndb;
|
|
ndb= NULL;
|
|
}
|
|
DBUG_RETURN(ndb);
|
|
}
|
|
|
|
|
|
void ha_ndbcluster::release_ndb(Ndb* ndb)
|
|
{
|
|
DBUG_ENTER("release_ndb");
|
|
#ifdef USE_NDB_POOL
|
|
// Release to pool
|
|
Ndb::release(ndb);
|
|
#else
|
|
delete ndb;
|
|
#endif
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
If this thread already has a Ndb object allocated
|
|
in current THD, reuse it. Otherwise
|
|
seize a Ndb object, assign it to current THD and use it.
|
|
|
|
Having a Ndb object also means that a connection to
|
|
NDB cluster has been opened. The connection is
|
|
checked.
|
|
|
|
*/
|
|
|
|
int ha_ndbcluster::check_ndb_connection()
|
|
{
|
|
THD* thd= current_thd;
|
|
Ndb* ndb;
|
|
DBUG_ENTER("check_ndb_connection");
|
|
|
|
if (!thd->transaction.ndb)
|
|
{
|
|
ndb= seize_ndb();
|
|
if (!ndb)
|
|
DBUG_RETURN(2);
|
|
thd->transaction.ndb= ndb;
|
|
}
|
|
m_ndb= (Ndb*)thd->transaction.ndb;
|
|
m_ndb->setDatabaseName(m_dbname);
|
|
if (m_ndb->waitUntilReady() != 0)
|
|
{
|
|
DBUG_PRINT("error", ("Ndb was not ready"));
|
|
DBUG_RETURN(3);
|
|
}
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
void ndbcluster_close_connection(THD *thd)
|
|
{
|
|
Ndb* ndb;
|
|
DBUG_ENTER("ndbcluster_close_connection");
|
|
ndb= (Ndb*)thd->transaction.ndb;
|
|
ha_ndbcluster::release_ndb(ndb);
|
|
thd->transaction.ndb= NULL;
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Try to discover one table from NDB
|
|
*/
|
|
|
|
int ndbcluster_discover(const char *dbname, const char *name,
|
|
const void** frmblob, uint* frmlen)
|
|
{
|
|
uint len;
|
|
const void* data;
|
|
const NDBTAB* tab;
|
|
DBUG_ENTER("ndbcluster_discover");
|
|
DBUG_PRINT("enter", ("db: %s, name: %s", dbname, name));
|
|
|
|
Ndb ndb(dbname);
|
|
if ((ndb.init() != 0) && (ndb.waitUntilReady() != 0))
|
|
ERR_RETURN(ndb.getNdbError());
|
|
|
|
if (!(tab= ndb.getDictionary()->getTable(name)))
|
|
{
|
|
DBUG_PRINT("info", ("Table %s not found", name));
|
|
DBUG_RETURN(1);
|
|
}
|
|
|
|
DBUG_PRINT("info", ("Found table %s", tab->getName()));
|
|
|
|
len= tab->getFrmLength();
|
|
if (len == 0 || tab->getFrmData() == NULL)
|
|
{
|
|
DBUG_PRINT("No frm data found",
|
|
("Table is probably created via NdbApi"));
|
|
DBUG_RETURN(2);
|
|
}
|
|
|
|
if (unpackfrm(&data, &len, tab->getFrmData()))
|
|
DBUG_RETURN(3);
|
|
|
|
*frmlen= len;
|
|
*frmblob= data;
|
|
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
static Ndb* g_ndb= NULL;
|
|
|
|
#ifdef USE_DISCOVER_ON_STARTUP
|
|
/*
|
|
Dicover tables from NDB Cluster
|
|
- fetch a list of tables from NDB
|
|
- store the frm file for each table on disk
|
|
- if the table has an attached frm file
|
|
- if the database of the table exists
|
|
*/
|
|
|
|
int ndb_discover_tables()
|
|
{
|
|
uint i;
|
|
NdbDictionary::Dictionary::List list;
|
|
NdbDictionary::Dictionary* dict;
|
|
char path[FN_REFLEN];
|
|
DBUG_ENTER("ndb_discover_tables");
|
|
|
|
/* List tables in NDB Cluster kernel */
|
|
dict= g_ndb->getDictionary();
|
|
if (dict->listObjects(list,
|
|
NdbDictionary::Object::UserTable) != 0)
|
|
ERR_RETURN(g_ndb->getNdbError());
|
|
|
|
for (i= 0 ; i < list.count ; i++)
|
|
{
|
|
NdbDictionary::Dictionary::List::Element& t= list.elements[i];
|
|
|
|
DBUG_PRINT("discover", ("%d: %s/%s", t.id, t.database, t.name));
|
|
if (create_table_from_handler(t.database, t.name, true))
|
|
DBUG_PRINT("info", ("Could not discover %s/%s", t.database, t.name));
|
|
}
|
|
DBUG_RETURN(0);
|
|
}
|
|
#endif
|
|
|
|
|
|
/*
|
|
Initialise all gloal variables before creating
|
|
a NDB Cluster table handler
|
|
*/
|
|
|
|
bool ndbcluster_init()
|
|
{
|
|
DBUG_ENTER("ndbcluster_init");
|
|
// Create a Ndb object to open the connection to NDB
|
|
g_ndb= new Ndb("sys");
|
|
if (g_ndb->init() != 0)
|
|
{
|
|
ERR_PRINT (g_ndb->getNdbError());
|
|
DBUG_RETURN(TRUE);
|
|
}
|
|
if (g_ndb->waitUntilReady() != 0)
|
|
{
|
|
ERR_PRINT (g_ndb->getNdbError());
|
|
DBUG_RETURN(TRUE);
|
|
}
|
|
(void) hash_init(&ndbcluster_open_tables,system_charset_info,32,0,0,
|
|
(hash_get_key) ndbcluster_get_key,0,0);
|
|
pthread_mutex_init(&ndbcluster_mutex,MY_MUTEX_INIT_FAST);
|
|
ndbcluster_inited= 1;
|
|
#ifdef USE_DISCOVER_ON_STARTUP
|
|
if (ndb_discover_tables() != 0)
|
|
DBUG_RETURN(TRUE);
|
|
#endif
|
|
DBUG_RETURN(false);
|
|
}
|
|
|
|
|
|
/*
|
|
End use of the NDB Cluster table handler
|
|
- free all global variables allocated by
|
|
ndcluster_init()
|
|
*/
|
|
|
|
bool ndbcluster_end()
|
|
{
|
|
DBUG_ENTER("ndbcluster_end");
|
|
delete g_ndb;
|
|
g_ndb= NULL;
|
|
if (!ndbcluster_inited)
|
|
DBUG_RETURN(0);
|
|
hash_free(&ndbcluster_open_tables);
|
|
#ifdef USE_NDB_POOL
|
|
ndb_pool_release();
|
|
#endif
|
|
pthread_mutex_destroy(&ndbcluster_mutex);
|
|
ndbcluster_inited= 0;
|
|
DBUG_RETURN(0);
|
|
}
|
|
|
|
|
|
/*
|
|
Set m_tabname from full pathname to table file
|
|
*/
|
|
|
|
void ha_ndbcluster::set_tabname(const char *path_name)
|
|
{
|
|
char *end, *ptr;
|
|
|
|
/* Scan name from the end */
|
|
end= strend(path_name)-1;
|
|
ptr= end;
|
|
while (ptr >= path_name && *ptr != '\\' && *ptr != '/') {
|
|
ptr--;
|
|
}
|
|
uint name_len= end - ptr;
|
|
memcpy(m_tabname, ptr + 1, end - ptr);
|
|
m_tabname[name_len]= '\0';
|
|
#ifdef __WIN__
|
|
/* Put to lower case */
|
|
ptr= m_tabname;
|
|
|
|
while (*ptr != '\0') {
|
|
*ptr = tolower(*ptr);
|
|
ptr++;
|
|
}
|
|
#endif
|
|
}
|
|
|
|
/**
|
|
* Set a given location from full pathname to table file
|
|
*
|
|
*/
|
|
void
|
|
ha_ndbcluster::set_tabname(const char *path_name, char * tabname)
|
|
{
|
|
char *end, *ptr;
|
|
|
|
/* Scan name from the end */
|
|
end = strend(path_name)-1;
|
|
ptr = end;
|
|
while (ptr >= path_name && *ptr != '\\' && *ptr != '/') {
|
|
ptr--;
|
|
}
|
|
uint name_len = end - ptr;
|
|
memcpy(tabname, ptr + 1, end - ptr);
|
|
tabname[name_len] = '\0';
|
|
#ifdef __WIN__
|
|
/* Put to lower case */
|
|
ptr = tabname;
|
|
|
|
while (*ptr != '\0') {
|
|
*ptr= tolower(*ptr);
|
|
ptr++;
|
|
}
|
|
#endif
|
|
}
|
|
|
|
|
|
/*
|
|
Set m_dbname from full pathname to table file
|
|
|
|
*/
|
|
|
|
void ha_ndbcluster::set_dbname(const char *path_name)
|
|
{
|
|
char *end, *ptr;
|
|
|
|
/* Scan name from the end */
|
|
ptr= strend(path_name)-1;
|
|
while (ptr >= path_name && *ptr != '\\' && *ptr != '/') {
|
|
ptr--;
|
|
}
|
|
ptr--;
|
|
end= ptr;
|
|
while (ptr >= path_name && *ptr != '\\' && *ptr != '/') {
|
|
ptr--;
|
|
}
|
|
uint name_len= end - ptr;
|
|
memcpy(m_dbname, ptr + 1, name_len);
|
|
m_dbname[name_len]= '\0';
|
|
#ifdef __WIN__
|
|
/* Put to lower case */
|
|
|
|
ptr= m_dbname;
|
|
|
|
while (*ptr != '\0') {
|
|
*ptr= tolower(*ptr);
|
|
ptr++;
|
|
}
|
|
#endif
|
|
}
|
|
|
|
|
|
ha_rows
|
|
ha_ndbcluster::records_in_range(int inx,
|
|
const byte *start_key,uint start_key_len,
|
|
enum ha_rkey_function start_search_flag,
|
|
const byte *end_key,uint end_key_len,
|
|
enum ha_rkey_function end_search_flag)
|
|
{
|
|
ha_rows records= 10;
|
|
KEY* key_info= table->key_info + inx;
|
|
uint key_length= key_info->key_length;
|
|
|
|
DBUG_ENTER("records_in_range");
|
|
DBUG_PRINT("enter", ("inx: %d", inx));
|
|
DBUG_PRINT("enter", ("start_key: %x, start_key_len: %d", start_key, start_key_len));
|
|
DBUG_PRINT("enter", ("start_search_flag: %d", start_search_flag));
|
|
DBUG_PRINT("enter", ("end_key: %x, end_key_len: %d", end_key, end_key_len));
|
|
DBUG_PRINT("enter", ("end_search_flag: %d", end_search_flag));
|
|
|
|
/*
|
|
Check that start_key_len is equal to
|
|
the length of the used index and
|
|
prevent partial scan/read of hash indexes by returning HA_POS_ERROR
|
|
*/
|
|
NDB_INDEX_TYPE idx_type= get_index_type(inx);
|
|
if ((idx_type == UNIQUE_INDEX || idx_type == PRIMARY_KEY_INDEX) &&
|
|
start_key_len < key_length)
|
|
{
|
|
DBUG_PRINT("warning", ("Tried to use index which required"
|
|
"full key length: %d, HA_POS_ERROR",
|
|
key_length));
|
|
records= HA_POS_ERROR;
|
|
}
|
|
DBUG_RETURN(records);
|
|
}
|
|
|
|
|
|
/*
|
|
Handling the shared NDB_SHARE structure that is needed to
|
|
provide table locking.
|
|
It's also used for sharing data with other NDB handlers
|
|
in the same MySQL Server. There is currently not much
|
|
data we want to or can share.
|
|
*/
|
|
|
|
static byte* ndbcluster_get_key(NDB_SHARE *share,uint *length,
|
|
my_bool not_used __attribute__((unused)))
|
|
{
|
|
*length=share->table_name_length;
|
|
return (byte*) share->table_name;
|
|
}
|
|
|
|
static NDB_SHARE* get_share(const char *table_name)
|
|
{
|
|
NDB_SHARE *share;
|
|
pthread_mutex_lock(&ndbcluster_mutex);
|
|
uint length=(uint) strlen(table_name);
|
|
if (!(share=(NDB_SHARE*) hash_search(&ndbcluster_open_tables,
|
|
(byte*) table_name,
|
|
length)))
|
|
{
|
|
if ((share=(NDB_SHARE *) my_malloc(sizeof(*share)+length+1,
|
|
MYF(MY_WME | MY_ZEROFILL))))
|
|
{
|
|
share->table_name_length=length;
|
|
share->table_name=(char*) (share+1);
|
|
strmov(share->table_name,table_name);
|
|
if (my_hash_insert(&ndbcluster_open_tables, (byte*) share))
|
|
{
|
|
pthread_mutex_unlock(&ndbcluster_mutex);
|
|
my_free((gptr) share,0);
|
|
return 0;
|
|
}
|
|
thr_lock_init(&share->lock);
|
|
pthread_mutex_init(&share->mutex,MY_MUTEX_INIT_FAST);
|
|
}
|
|
}
|
|
share->use_count++;
|
|
pthread_mutex_unlock(&ndbcluster_mutex);
|
|
return share;
|
|
}
|
|
|
|
|
|
static void free_share(NDB_SHARE *share)
|
|
{
|
|
pthread_mutex_lock(&ndbcluster_mutex);
|
|
if (!--share->use_count)
|
|
{
|
|
hash_delete(&ndbcluster_open_tables, (byte*) share);
|
|
thr_lock_delete(&share->lock);
|
|
pthread_mutex_destroy(&share->mutex);
|
|
my_free((gptr) share, MYF(0));
|
|
}
|
|
pthread_mutex_unlock(&ndbcluster_mutex);
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
Internal representation of the frm blob
|
|
|
|
*/
|
|
|
|
struct frm_blob_struct
|
|
{
|
|
struct frm_blob_header
|
|
{
|
|
uint ver; // Version of header
|
|
uint orglen; // Original length of compressed data
|
|
uint complen; // Compressed length of data, 0=uncompressed
|
|
} head;
|
|
char data[1];
|
|
};
|
|
|
|
|
|
|
|
static int packfrm(const void *data, uint len,
|
|
const void **pack_data, uint *pack_len)
|
|
{
|
|
int error;
|
|
ulong org_len, comp_len;
|
|
uint blob_len;
|
|
frm_blob_struct* blob;
|
|
DBUG_ENTER("packfrm");
|
|
DBUG_PRINT("enter", ("data: %x, len: %d", data, len));
|
|
|
|
error= 1;
|
|
org_len = len;
|
|
if (my_compress((byte*)data, &org_len, &comp_len))
|
|
goto err;
|
|
|
|
DBUG_PRINT("info", ("org_len: %d, comp_len: %d", org_len, comp_len));
|
|
DBUG_DUMP("compressed", (char*)data, org_len);
|
|
|
|
error= 2;
|
|
blob_len= sizeof(frm_blob_struct::frm_blob_header)+org_len;
|
|
if (!(blob= (frm_blob_struct*) my_malloc(blob_len,MYF(MY_WME))))
|
|
goto err;
|
|
|
|
// Store compressed blob in machine independent format
|
|
int4store((char*)(&blob->head.ver), 1);
|
|
int4store((char*)(&blob->head.orglen), comp_len);
|
|
int4store((char*)(&blob->head.complen), org_len);
|
|
|
|
// Copy frm data into blob, already in machine independent format
|
|
memcpy(blob->data, data, org_len);
|
|
|
|
*pack_data = blob;
|
|
*pack_len = blob_len;
|
|
error = 0;
|
|
|
|
DBUG_PRINT("exit", ("pack_data: %x, pack_len: %d", *pack_data, *pack_len));
|
|
err:
|
|
DBUG_RETURN(error);
|
|
|
|
}
|
|
|
|
|
|
static int unpackfrm(const void **unpack_data, uint *unpack_len,
|
|
const void *pack_data)
|
|
{
|
|
const frm_blob_struct *blob = (frm_blob_struct*)pack_data;
|
|
byte *data;
|
|
ulong complen, orglen, ver;
|
|
DBUG_ENTER("unpackfrm");
|
|
DBUG_PRINT("enter", ("pack_data: %x", pack_data));
|
|
|
|
complen= uint4korr((char*)&blob->head.complen);
|
|
orglen= uint4korr((char*)&blob->head.orglen);
|
|
ver= uint4korr((char*)&blob->head.ver);
|
|
|
|
DBUG_PRINT("blob",("ver: %d complen: %d orglen: %d",
|
|
ver,complen,orglen));
|
|
DBUG_DUMP("blob->data", (char*) blob->data, complen);
|
|
|
|
if (ver != 1)
|
|
DBUG_RETURN(1);
|
|
if (!(data = my_malloc(max(orglen, complen), MYF(MY_WME))))
|
|
DBUG_RETURN(2);
|
|
memcpy(data, blob->data, complen);
|
|
|
|
if (my_uncompress(data, &complen, &orglen))
|
|
{
|
|
my_free((char*)data, MYF(0));
|
|
DBUG_RETURN(3);
|
|
}
|
|
|
|
*unpack_data = data;
|
|
*unpack_len = complen;
|
|
|
|
DBUG_PRINT("exit", ("frmdata: %x, len: %d", *unpack_data, *unpack_len));
|
|
|
|
DBUG_RETURN(0);
|
|
}
|
|
#endif /* HAVE_NDBCLUSTER_DB */
|