Added NDB storage engine

include/my_base.h

@@ -287,6 +287,9 @@ enum ha_base_keytype {
 #define HA_ERR_ROW_IS_REFERENCED 152     /* Cannot delete a parent row */
 #define HA_ERR_NO_SAVEPOINT	 153     /* No savepoint with that name */
 #define HA_ERR_NON_UNIQUE_BLOCK_SIZE 154 /* Non unique key block size */
+#define HA_ERR_OLD_METADATA      155  /* The frm file on disk is old */
+#define HA_ERR_TABLE_EXIST       156  /* The table existed in storage engine */
+#define HA_ERR_NO_CONNECTION     157  /* Could not connect to storage engine */
 
 	/* Other constants */
 

sql/Makefile.am

@@ -16,12 +16,11 @@
 
 #called from the top level Makefile
 
-
 MYSQLDATAdir =		$(localstatedir)
 MYSQLSHAREdir =		$(pkgdatadir)
 MYSQLBASEdir=		$(prefix)
 INCLUDES =		@MT_INCLUDES@ \
-			@bdb_includes@ @innodb_includes@ \
+			@bdb_includes@ @innodb_includes@ @ndbcluster_includes@ \
 			-I$(top_srcdir)/include -I$(top_srcdir)/regex \
 			-I$(srcdir) $(openssl_includes)
 WRAPLIBS=		@WRAPLIBS@
@@ -42,6 +41,7 @@ LDADD =			@isam_libs@ \
 mysqld_LDADD =		@MYSQLD_EXTRA_LDFLAGS@ \
 			@bdb_libs@ @innodb_libs@ @pstack_libs@ \
                         @innodb_system_libs@ \
+			@ndbcluster_libs@ @ndbcluster_system_libs@ \
 			$(LDADD)  $(CXXLDFLAGS) $(WRAPLIBS) @LIBDL@ @openssl_libs@
 noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			item_strfunc.h item_timefunc.h item_uniq.h \
@@ -52,7 +52,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			field.h handler.h \
 			ha_isammrg.h ha_isam.h ha_myisammrg.h\
 			ha_heap.h ha_myisam.h ha_berkeley.h ha_innodb.h \
-			opt_range.h protocol.h \
+			ha_ndbcluster.h opt_range.h protocol.h \
 			sql_select.h structs.h table.h sql_udf.h hash_filo.h\
 			lex.h lex_symbol.h sql_acl.h sql_crypt.h  \
 			log_event.h sql_repl.h slave.h \
@@ -75,11 +75,11 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc \
 			procedure.cc item_uniq.cc sql_test.cc \
 			log.cc log_event.cc init.cc derror.cc sql_acl.cc \
 			unireg.cc des_key_file.cc \
-			time.cc opt_range.cc opt_sum.cc \
+			discover.cc time.cc opt_range.cc opt_sum.cc \
 		   	records.cc filesort.cc handler.cc \
 		        ha_heap.cc ha_myisam.cc ha_myisammrg.cc \
 	                ha_berkeley.cc ha_innodb.cc \
-			ha_isam.cc ha_isammrg.cc \
+			ha_isam.cc ha_isammrg.cc ha_ndbcluster.cc \
 			sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
 			sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
 			sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \

sql/discover.cc

+/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+/* Functions for discover of frm file from handler */
+
+#include "mysql_priv.h"
+#include <my_dir.h>
+
+/*
+  Read the contents of a .frm file
+
+  SYNOPSIS
+    readfrm()
+
+    name           path to table-file "db/name"
+    frmdata        frm data
+    len            length of the read frmdata
+
+  RETURN VALUES
+   0	ok
+   1	Could not open file
+   2    Could not stat file
+   3    Could not allocate data for read
+        Could not read file
+
+   frmdata and len are set to 0 on error
+*/
+
+int readfrm(const char *name,
+	    const void **frmdata, uint *len)
+{
+  int    error;
+  char	 index_file[FN_REFLEN];
+  File	 file;
+  ulong read_len;
+  char *read_data;
+  MY_STAT state;  
+  DBUG_ENTER("readfrm");
+  DBUG_PRINT("enter",("name: '%s'",name));
+  
+  *frmdata= NULL;      // In case of errors
+  *len= 0;
+  error= 1;
+  if ((file=my_open(fn_format(index_file,name,"",reg_ext,4),
+		    O_RDONLY | O_SHARE,
+		    MYF(0))) < 0)  
+    goto err_end; 
+  
+  // Get length of file
+  error= 2;
+  if (my_fstat(file, &state, MYF(0)))
+    goto err;
+  read_len= state.st_size;  
+
+  // Read whole frm file
+  error= 3;
+  read_data= 0; 
+  if (read_string(file, &read_data, read_len))
+    goto err;
+
+  // Setup return data
+  *frmdata= (void*) read_data;
+  *len= read_len;
+  error= 0;
+  
+ err:
+  if (file > 0)
+    VOID(my_close(file,MYF(MY_WME)));
+  
+ err_end:		      /* Here when no file */
+  DBUG_RETURN (error);
+} /* readfrm */
+
+
+/*
+  Write the content of a frm data pointer 
+  to a frm file
+
+  SYNOPSIS
+    writefrm()
+
+    name           path to table-file "db/name"
+    frmdata        frm data
+    len            length of the frmdata
+
+  RETURN VALUES
+   0	ok
+   2    Could not write file
+*/
+
+int writefrm(const char *name, const void *frmdata, uint len)
+{
+  File file;
+  char	 index_file[FN_REFLEN];
+  int error;
+  DBUG_ENTER("writefrm");
+  DBUG_PRINT("enter",("name: '%s' len: %d ",name,len));
+  //DBUG_DUMP("frmdata", (char*)frmdata, len);
+
+  error= 0;
+  if ((file=my_create(fn_format(index_file,name,"",reg_ext,4),
+		      CREATE_MODE,O_RDWR | O_TRUNC,MYF(MY_WME))) >= 0)
+  {
+    if (my_write(file,(byte*)frmdata,len,MYF(MY_WME | MY_NABP)))
+      error= 2;
+  }
+  VOID(my_close(file,MYF(0)));
+  DBUG_RETURN(error);
+} /* writefrm */
+
+
+
+
+/*
+  Try to discover table from handler and 
+  if found, write the frm file to disk.
+  
+  RETURN VALUES:
+   0 : Table existed in handler and created 
+       on disk if so requested
+   1 : Table does not exist
+  >1 : error
+
+*/
+
+int create_table_from_handler(const char *db, 
+			      const char *name,
+			      bool create_if_found)
+{
+  int error= 0;  
+  const void* frmblob = NULL;
+  char path[FN_REFLEN];
+  uint frmlen = 0;
+  DBUG_ENTER("create_table_from_handler");
+  DBUG_PRINT("enter", ("create_if_found: %d", create_if_found));
+
+  if (ha_discover(db, name, &frmblob, &frmlen))
+    DBUG_RETURN(1); // Table does not exist
+
+  // Table exists in handler  
+  if  (create_if_found)
+  {
+    (void)strxnmov(path,FN_REFLEN,mysql_data_home,"/",db,"/",name,NullS);
+    // Save the frm file    
+    error = writefrm(path, frmblob, frmlen);
+  }
+    
+ err:
+  if (frmblob)
+    my_free((char*) frmblob,MYF(0));
+  DBUG_RETURN(error);  
+}
+
+int table_exists_in_handler(const char *db, 
+			     const char *name)
+{ 
+  return (create_table_from_handler(db, name, false) == 0);
+}

sql/ha_ndbcluster.cc

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

sql/ha_ndbcluster.h

+/* 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
+*/
+
+/* The class defining a handle to an NDB Cluster table */
+
+#ifdef __GNUC__
+#pragma interface                       /* gcc class implementation */
+#endif
+
+#include <ndbapi_limits.h>
+#include <ndb_types.h>
+
+class Ndb;             // Forward declaration
+class NdbOperation;    // Forward declaration
+class NdbConnection;   // Forward declaration
+class NdbRecAttr;      // Forward declaration
+class NdbResultSet;    // Forward declaration
+
+typedef enum ndb_index_type {
+  UNDEFINED_INDEX = 0,
+  PRIMARY_KEY_INDEX = 1,
+  UNIQUE_INDEX = 2,
+  ORDERED_INDEX = 3
+} NDB_INDEX_TYPE;
+
+
+typedef struct st_ndbcluster_share {
+  THR_LOCK lock;
+  pthread_mutex_t mutex;
+  char *table_name;
+  uint table_name_length,use_count;
+} NDB_SHARE;
+
+class ha_ndbcluster: public handler
+{
+ public:
+  ha_ndbcluster(TABLE *table);
+  ~ha_ndbcluster();
+
+  int open(const char *name, int mode, uint test_if_locked);
+  int close(void);
+
+  int write_row(byte *buf);
+  int update_row(const byte *old_data, byte *new_data);
+  int delete_row(const byte *buf);
+  int index_init(uint index);
+  int index_end();
+  int index_read(byte *buf, const byte *key, uint key_len, 
+		 enum ha_rkey_function find_flag);
+  int index_read_idx(byte *buf, uint index, const byte *key, uint key_len, 
+		     enum ha_rkey_function find_flag);
+  int index_next(byte *buf);
+  int index_prev(byte *buf);
+  int index_first(byte *buf);
+  int index_last(byte *buf);
+  int rnd_init(bool scan=1);
+  int rnd_end();
+  int rnd_next(byte *buf);
+  int rnd_pos(byte *buf, byte *pos);
+  void position(const byte *record);
+
+  void info(uint);
+  int extra(enum ha_extra_function operation);
+  int extra_opt(enum ha_extra_function operation, ulong cache_size);
+  int reset();
+  int external_lock(THD *thd, int lock_type);
+  int start_stmt(THD *thd);
+  const char * table_type() const { return("ndbcluster");}
+  const char ** bas_ext() const;
+  ulong table_flags(void) const { return m_table_flags; }
+  ulong index_flags(uint idx) const;
+  uint max_record_length() const { return NDB_MAX_TUPLE_SIZE; };
+  uint max_keys()          const { return MAX_KEY;  }
+  uint max_key_parts()     const { return NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY; };
+  uint max_key_length()    const { return NDB_MAX_KEY_SIZE;};
+
+  int rename_table(const char *from, const char *to);
+  int delete_table(const char *name);
+  int create(const char *name, TABLE *form, HA_CREATE_INFO *info);
+  THR_LOCK_DATA **store_lock(THD *thd,
+			     THR_LOCK_DATA **to,
+			     enum thr_lock_type lock_type);
+
+  bool low_byte_first() const
+    { 
+#ifdef WORDS_BIGENDIAN
+      return false;
+#else
+      return true;
+#endif
+    }
+  bool has_transactions()  { return true;}
+
+  const char* index_type(uint key_number) {
+    switch (get_index_type(key_number)) {
+    case ORDERED_INDEX:
+      return "BTREE";
+    case UNIQUE_INDEX:
+    case PRIMARY_KEY_INDEX:
+    default:
+      return "HASH";
+    }
+  }
+
+  double scan_time();
+  ha_rows 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);
+
+
+  static Ndb* seize_ndb();
+  static void release_ndb(Ndb* ndb);
+
+    
+ private:
+  int alter_table_name(const char *from, const char *to);
+  int drop_table();
+  int create_index(const char *name, KEY *key_info);
+  int initialize_autoincrement(const void* table);
+  int get_metadata(const char* path);
+  void release_metadata();
+  const char* get_index_name(uint idx_no) const;
+  NDB_INDEX_TYPE get_index_type(uint idx_no) const;
+  NDB_INDEX_TYPE get_index_type_from_table(uint index_no) const;
+  
+  int pk_read(const byte *key, uint key_len, 
+	      byte *buf);
+  int unique_index_read(const byte *key, uint key_len, 
+			byte *buf);
+  int ordered_index_scan(const byte *key, uint key_len, 
+			 byte *buf,
+			 enum ha_rkey_function find_flag);
+  int full_table_scan(byte * buf);
+  int next_result(byte *buf); 
+#if 0
+  int filtered_scan(const byte *key, uint key_len, 
+		    byte *buf,
+		    enum ha_rkey_function find_flag);
+#endif
+
+  void unpack_record(byte *buf);
+
+  void set_dbname(const char *pathname);
+  void set_tabname(const char *pathname);
+  void set_tabname(const char *pathname, char *tabname);
+
+  bool set_hidden_key(NdbOperation*,
+		      uint fieldnr, const byte* field_ptr);
+  int set_ndb_key(NdbOperation*, Field *field,
+		  uint fieldnr, const byte* field_ptr);
+  int set_ndb_value(NdbOperation*, Field *field, uint fieldnr);
+  int get_ndb_value(NdbOperation*, uint fieldnr, byte *field_ptr);
+  int set_primary_key(NdbOperation *op, const byte *key);
+  int set_primary_key(NdbOperation *op);
+  int key_cmp(uint keynr, const byte * old_row, const byte * new_row);
+  void print_results();
+
+  longlong get_auto_increment();
+
+  int ndb_err(NdbConnection*);
+
+ private:
+  int check_ndb_connection();
+
+  NdbConnection *m_active_trans;
+  NdbResultSet *m_active_cursor;
+  Ndb *m_ndb;
+  void *m_table;		
+  char m_dbname[FN_HEADLEN];
+  //char m_schemaname[FN_HEADLEN];
+  char m_tabname[FN_HEADLEN];
+  ulong m_table_flags;
+  THR_LOCK_DATA m_lock;
+  NDB_SHARE *m_share;
+  NDB_INDEX_TYPE  m_indextype[MAX_KEY];
+  NdbRecAttr *m_value[NDB_MAX_ATTRIBUTES_IN_TABLE];
+  bool m_use_write;
+};
+
+bool ndbcluster_init(void);
+bool ndbcluster_end(void);
+
+int ndbcluster_commit(THD *thd, void* ndb_transaction);
+int ndbcluster_rollback(THD *thd, void* ndb_transaction);
+
+void ndbcluster_close_connection(THD *thd);
+
+int ndbcluster_discover(const char* dbname, const char* name,
+			const void** frmblob, uint* frmlen);
+int ndbcluster_drop_database(const char* path);
+
+
+
+
+
+
+
+

sql/handler.cc

@@ -37,6 +37,9 @@
 #else
 #define innobase_query_caching_of_table_permitted(X,Y,Z) 1
 #endif
+#ifdef HAVE_NDBCLUSTER_DB
+#include "ha_ndbcluster.h"
+#endif
 #include <myisampack.h>
 #include <errno.h>
 
@@ -48,7 +51,7 @@ ulong ha_read_count, ha_write_count, ha_delete_count, ha_update_count,
       ha_read_key_count, ha_read_next_count, ha_read_prev_count,
       ha_read_first_count, ha_read_last_count,
       ha_commit_count, ha_rollback_count,
-      ha_read_rnd_count, ha_read_rnd_next_count;
+      ha_read_rnd_count, ha_read_rnd_next_count, ha_discover_count;
 
 static SHOW_COMP_OPTION have_yes= SHOW_OPTION_YES;
 
@@ -76,6 +79,10 @@ struct show_table_type_st sys_table_types[]=
    "Supports transactions and page-level locking", DB_TYPE_BERKELEY_DB},
   {"BERKELEYDB",&have_berkeley_db,
    "Alias for BDB", DB_TYPE_BERKELEY_DB},
+  {"NDBCLUSTER", &have_ndbcluster,
+   "Clustered, fault tolerant memory based tables", DB_TYPE_NDBCLUSTER},
+  {"NDB", &have_ndbcluster,
+   "Alias for NDBCLUSTER", DB_TYPE_NDBCLUSTER},
   {NullS, NULL, NullS, DB_TYPE_UNKNOWN}
 };
 
@@ -171,6 +178,10 @@ handler *get_new_handler(TABLE *table, enum db_type db_type)
 #ifdef HAVE_INNOBASE_DB
   case DB_TYPE_INNODB:
     return new ha_innobase(table);
+#endif
+#ifdef HAVE_NDBCLUSTER_DB
+  case DB_TYPE_NDBCLUSTER:
+    return new ha_ndbcluster(table);
 #endif
   case DB_TYPE_HEAP:
     return new ha_heap(table);
@@ -215,6 +226,18 @@ int ha_init()
     else
       opt_using_transactions=1;
   }
+#endif
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+  {
+    if (ndbcluster_init())
+    {
+      have_ndbcluster= SHOW_OPTION_DISABLED;
+      error= 1;
+    }
+    else
+      opt_using_transactions=1;
+  }
 #endif
   return error;
 }
@@ -242,6 +265,10 @@ int ha_panic(enum ha_panic_function flag)
 #ifdef HAVE_INNOBASE_DB
   if (have_innodb == SHOW_OPTION_YES)
     error|=innobase_end();
+#endif
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    error|=ndbcluster_end();
 #endif
   return error;
 } /* ha_panic */
@@ -252,6 +279,10 @@ void ha_drop_database(char* path)
   if (have_innodb == SHOW_OPTION_YES)
     innobase_drop_database(path);
 #endif
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    ndbcluster_drop_database(path);
+#endif
 }
 
 void ha_close_connection(THD* thd)
@@ -260,6 +291,10 @@ void ha_close_connection(THD* thd)
   if (have_innodb == SHOW_OPTION_YES)
     innobase_close_connection(thd);
 #endif
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    ndbcluster_close_connection(thd);
+#endif
 }
 
 /*
@@ -419,6 +454,19 @@ int ha_commit_trans(THD *thd, THD_TRANS* trans)
 		      WRITE_CACHE, (my_off_t) 0, 0, 1);
       thd->transaction.trans_log.end_of_file= max_binlog_cache_size;
     }
+#ifdef HAVE_NDBCLUSTER_DB
+    if (trans->ndb_tid)
+    {
+      if ((error=ndbcluster_commit(thd,trans->ndb_tid)))
+      {
+        my_error(ER_ERROR_DURING_COMMIT, MYF(0), error);
+        error=1;
+      }
+      if (trans == &thd->transaction.all)
+        operation_done= transaction_commited= 1;
+      trans->ndb_tid=0;
+    }
+#endif
 #ifdef HAVE_BERKELEY_DB
     if (trans->bdb_tid)
     {
@@ -472,6 +520,18 @@ int ha_rollback_trans(THD *thd, THD_TRANS *trans)
   if (opt_using_transactions)
   {
     bool operation_done=0;
+#ifdef HAVE_NDBCLUSTER_DB
+    if (trans->ndb_tid)
+    {
+      if ((error=ndbcluster_rollback(thd, trans->ndb_tid)))
+      {
+        my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), error);
+        error=1;
+      }
+      trans->ndb_tid = 0;
+      operation_done=1;
+    }
+#endif
 #ifdef HAVE_BERKELEY_DB
     if (trans->bdb_tid)
     {
@@ -1151,8 +1211,10 @@ bool handler::caching_allowed(THD* thd, char* table_key,
 ** Some general functions that isn't in the handler class
 ****************************************************************************/
 
-	/* Initiates table-file and calls apropriate database-creator */
-	/* Returns 1 if something got wrong */
+/*
+  Initiates table-file and calls apropriate database-creator
+  Returns 1 if something got wrong
+*/
 
 int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
 		    bool update_create_info)
@@ -1168,7 +1230,7 @@ int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
   {
     update_create_info_from_table(create_info, &table);
     if (table.file->table_flags() & HA_DROP_BEFORE_CREATE)
-      table.file->delete_table(name);		// Needed for BDB tables
+      table.file->delete_table(name);
   }
   if (lower_case_table_names == 2 &&
       !(table.file->table_flags() & HA_FILE_BASED))
@@ -1289,6 +1351,26 @@ int ha_change_key_cache(KEY_CACHE *old_key_cache,
 }
 
 
+/*
+  Try to discover one table from handler(s)
+*/
+
+int ha_discover(const char* dbname, const char* name,
+               const void** frmblob, uint* frmlen)
+{
+  int error= 1; // Table does not exist in any handler
+  DBUG_ENTER("ha_discover");
+  DBUG_PRINT("enter", ("db: %s, name: %s", dbname, name));
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    error= ndbcluster_discover(dbname, name, frmblob, frmlen);
+#endif
+  if (!error)
+    statistic_increment(ha_discover_count,&LOCK_status);
+  DBUG_RETURN(error);
+}
+
+
 /*
   Read first row between two ranges.
   Store ranges for future calls to read_range_next
@@ -1425,3 +1507,5 @@ int handler::compare_key(key_range *range)
   }
   return key_compare_result_on_equal;
 }
+
+

sql/handler.h

@@ -28,7 +28,8 @@
 #define NO_HASH				/* Not yet implemented */
 #endif
 
-#if defined(HAVE_BERKELEY_DB) || defined(HAVE_INNOBASE_DB)
+#if defined(HAVE_BERKELEY_DB) || defined(HAVE_INNOBASE_DB) || \
+    defined(HAVE_NDBCLUSTER_DB)
 #define USING_TRANSACTIONS
 #endif
 
@@ -80,7 +81,6 @@
 #define HA_FILE_BASED	       (1 << 26)
 
 
-
 /* bits in index_flags(index_number) for what you can do with index */
 #define HA_WRONG_ASCII_ORDER	1	/* Can't use sorting through key */
 #define HA_READ_NEXT		2	/* Read next record with same key */
@@ -141,12 +141,17 @@
 #define HA_CACHE_TBL_ASKTRANSACT 1
 #define HA_CACHE_TBL_TRANSACT    2
 
-enum db_type { DB_TYPE_UNKNOWN=0,DB_TYPE_DIAB_ISAM=1,
-	       DB_TYPE_HASH,DB_TYPE_MISAM,DB_TYPE_PISAM,
-	       DB_TYPE_RMS_ISAM, DB_TYPE_HEAP, DB_TYPE_ISAM,
-	       DB_TYPE_MRG_ISAM, DB_TYPE_MYISAM, DB_TYPE_MRG_MYISAM,
-	       DB_TYPE_BERKELEY_DB, DB_TYPE_INNODB, DB_TYPE_GEMINI,
-	       DB_TYPE_DEFAULT };
+enum db_type 
+{ 
+  DB_TYPE_UNKNOWN=0,DB_TYPE_DIAB_ISAM=1,
+  DB_TYPE_HASH,DB_TYPE_MISAM,DB_TYPE_PISAM,
+  DB_TYPE_RMS_ISAM, DB_TYPE_HEAP, DB_TYPE_ISAM,
+  DB_TYPE_MRG_ISAM, DB_TYPE_MYISAM, DB_TYPE_MRG_MYISAM,
+  DB_TYPE_BERKELEY_DB, DB_TYPE_INNODB, 
+  DB_TYPE_GEMINI, DB_TYPE_NDBCLUSTER,
+	       
+  DB_TYPE_DEFAULT // Must be last
+};
 
 struct show_table_type_st {
   const char *type;
@@ -176,6 +181,7 @@ typedef struct st_thd_trans {
   void *bdb_tid;
   void *innobase_tid;
   bool innodb_active_trans;
+  void *ndb_tid;
 } THD_TRANS;
 
 enum enum_tx_isolation { ISO_READ_UNCOMMITTED, ISO_READ_COMMITTED,
@@ -479,3 +485,5 @@ bool ha_flush_logs(void);
 int ha_recovery_logging(THD *thd, bool on);
 int ha_change_key_cache(KEY_CACHE *old_key_cache,
 			KEY_CACHE *new_key_cache);
+int ha_discover(const char* dbname, const char* name,
+		const void** frmblob, uint* frmlen);

sql/lex.h

@@ -48,7 +48,7 @@ SYM_GROUP sym_group_rtree= {"RTree keys", "HAVE_RTREE_KEYS"};
 */
 
 static SYMBOL symbols[] = {
-  { "&&",		SYM(AND)},
+  { "&&",		SYM(AND_SYM)},
   { "<",		SYM(LT)},
   { "<=",		SYM(LE)},
   { "<>",		SYM(NE)},
@@ -67,7 +67,7 @@ static SYMBOL symbols[] = {
   { "ALL",		SYM(ALL)},
   { "ALTER",		SYM(ALTER)},
   { "ANALYZE",		SYM(ANALYZE_SYM)},
-  { "AND",		SYM(AND)},
+  { "AND",		SYM(AND_SYM)},
   { "ANY",              SYM(ANY_SYM)},
   { "AS",		SYM(AS)},
   { "ASC",		SYM(ASC)},
@@ -295,6 +295,8 @@ static SYMBOL symbols[] = {
   { "NAMES",		SYM(NAMES_SYM)},
   { "NATIONAL",		SYM(NATIONAL_SYM)},
   { "NATURAL",		SYM(NATURAL)},
+  { "NDB",              SYM(NDBCLUSTER_SYM)},
+  { "NDBCLUSTER",       SYM(NDBCLUSTER_SYM)},
   { "NCHAR",		SYM(NCHAR_SYM)},
   { "NEW",              SYM(NEW_SYM)},
   { "NEXT",		SYM(NEXT_SYM)},
@@ -312,7 +314,7 @@ static SYMBOL symbols[] = {
   { "OPTIMIZE",		SYM(OPTIMIZE)},
   { "OPTION",		SYM(OPTION)},
   { "OPTIONALLY",	SYM(OPTIONALLY)},
-  { "OR",		SYM(OR)},
+  { "OR",		SYM(OR_SYM)},
   { "ORDER",		SYM(ORDER_SYM)},
   { "OUTER",		SYM(OUTER)},
   { "OUTFILE",		SYM(OUTFILE)},

sql/mysql_priv.h

@@ -837,7 +837,7 @@ extern ulong server_id, concurrency;
 extern ulong ha_read_count, ha_write_count, ha_delete_count, ha_update_count;
 extern ulong ha_read_key_count, ha_read_next_count, ha_read_prev_count;
 extern ulong ha_read_first_count, ha_read_last_count;
-extern ulong ha_read_rnd_count, ha_read_rnd_next_count;
+extern ulong ha_read_rnd_count, ha_read_rnd_next_count, ha_discover_count;
 extern ulong ha_commit_count, ha_rollback_count,table_cache_size;
 extern ulong max_connections,max_connect_errors, connect_timeout;
 extern ulong slave_net_timeout;
@@ -891,6 +891,7 @@ extern SHOW_VAR init_vars[],status_vars[], internal_vars[];
 extern SHOW_COMP_OPTION have_isam;
 extern SHOW_COMP_OPTION have_innodb;
 extern SHOW_COMP_OPTION have_berkeley_db;
+extern SHOW_COMP_OPTION have_ndbcluster;
 extern struct system_variables global_system_variables;
 extern struct system_variables max_system_variables;
 extern struct rand_struct sql_rand;
@@ -960,6 +961,10 @@ int format_number(uint inputflag,uint max_length,my_string pos,uint length,
 		  my_string *errpos);
 int openfrm(const char *name,const char *alias,uint filestat,uint prgflag,
 	    uint ha_open_flags, TABLE *outparam);
+int readfrm(const char *name, const void** data, uint* length);
+int writefrm(const char* name, const void* data, uint len);
+int create_table_from_handler(const char *db, const char *name,
+			      bool create_if_found);
 int closefrm(TABLE *table);
 db_type get_table_type(const char *name);
 int read_string(File file, gptr *to, uint length);

sql/mysqld.cc

@@ -32,6 +32,9 @@
 #ifdef HAVE_ISAM
 #include "ha_isam.h"
 #endif
+#ifdef HAVE_NDBCLUSTER_DB
+#include "ha_ndbcluster.h"
+#endif
 #include <nisam.h>
 #include <thr_alarm.h>
 #include <ft_global.h>
@@ -261,7 +264,7 @@ my_bool opt_local_infile, opt_external_locking, opt_slave_compressed_protocol;
 my_bool opt_safe_user_create = 0, opt_no_mix_types = 0;
 my_bool opt_show_slave_auth_info, opt_sql_bin_update = 0;
 my_bool opt_log_slave_updates= 0;
-my_bool	opt_console= 0, opt_bdb, opt_innodb, opt_isam;
+my_bool	opt_console= 0, opt_bdb, opt_innodb, opt_isam, opt_ndbcluster;
 my_bool opt_readonly, use_temp_pool, relay_log_purge;
 my_bool opt_sync_bdb_logs, opt_sync_frm;
 my_bool opt_secure_auth= 0;
@@ -370,7 +373,7 @@ KEY_CACHE *sql_key_cache;
 CHARSET_INFO *system_charset_info, *files_charset_info ;
 CHARSET_INFO *national_charset_info, *table_alias_charset;
 
-SHOW_COMP_OPTION have_berkeley_db, have_innodb, have_isam;
+SHOW_COMP_OPTION have_berkeley_db, have_innodb, have_isam, have_ndbcluster;
 SHOW_COMP_OPTION have_raid, have_openssl, have_symlink, have_query_cache;
 SHOW_COMP_OPTION have_crypt, have_compress;
 
@@ -3625,7 +3628,7 @@ enum options_mysqld
   OPT_INNODB_FAST_SHUTDOWN,
   OPT_INNODB_FILE_PER_TABLE,
   OPT_SAFE_SHOW_DB,
-  OPT_INNODB, OPT_ISAM, OPT_SKIP_SAFEMALLOC,
+  OPT_INNODB, OPT_ISAM, OPT_NDBCLUSTER, OPT_SKIP_SAFEMALLOC,
   OPT_TEMP_POOL, OPT_TX_ISOLATION,
   OPT_SKIP_STACK_TRACE, OPT_SKIP_SYMLINKS,
   OPT_MAX_BINLOG_DUMP_EVENTS, OPT_SPORADIC_BINLOG_DUMP_FAIL,
@@ -4158,6 +4161,13 @@ Disable with --skip-innodb (will save memory).",
 Disable with --skip-isam.",
    (gptr*) &opt_isam, (gptr*) &opt_isam, 0, GET_BOOL, NO_ARG, 1, 0, 0,
    0, 0, 0},
+#ifdef HAVE_NDBCLUSTER_DB
+  {"ndbcluster", OPT_NDBCLUSTER, "Enable NDB Cluster (if this version of MySQL
+supports it). \
+Disable with --skip-ndbcluster (will save memory).",
+   (gptr*) &opt_ndbcluster, (gptr*) &opt_ndbcluster, 0, GET_BOOL, NO_ARG, 1, 0, 0,
+   0, 0, 0},
+#endif
   {"skip-locking", OPT_SKIP_LOCK,
    "Deprecated option, use --skip-external-locking instead.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
@@ -4828,6 +4838,7 @@ struct show_var_st status_vars[]= {
   {"Handler_rollback",         (char*) &ha_rollback_count,      SHOW_LONG},
   {"Handler_update",           (char*) &ha_update_count,        SHOW_LONG},
   {"Handler_write",            (char*) &ha_write_count,         SHOW_LONG},
+  {"Handler_discover",         (char*) &ha_discover_count,      SHOW_LONG},
   {"Key_blocks_not_flushed",   (char*) &dflt_key_cache_var.global_blocks_changed,
    SHOW_KEY_CACHE_LONG},
   {"Key_blocks_used",          (char*) &dflt_key_cache_var.global_blocks_used,
@@ -5121,6 +5132,11 @@ static void mysql_init_variables(void)
 #else
   have_isam=SHOW_OPTION_NO;
 #endif
+#ifdef HAVE_NDBCLUSTER_DB
+  have_ndbcluster=SHOW_OPTION_YES;
+#else
+  have_ndbcluster=SHOW_OPTION_NO;
+#endif
 #ifdef USE_RAID
   have_raid=SHOW_OPTION_YES;
 #else
@@ -5589,6 +5605,14 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
       have_isam= SHOW_OPTION_YES;
     else
       have_isam= SHOW_OPTION_DISABLED;
+#endif
+    break;
+  case OPT_NDBCLUSTER:
+#ifdef HAVE_NDBCLUSTER_DB
+    if (opt_ndbcluster)
+      have_ndbcluster=SHOW_OPTION_YES;
+    else
+      have_ndbcluster=SHOW_OPTION_DISABLED;
 #endif
     break;
   case OPT_INNODB:

sql/set_var.cc

@@ -59,6 +59,9 @@
 #ifdef HAVE_INNOBASE_DB
 #include "ha_innodb.h"
 #endif
+#ifdef HAVE_NDBCLUSTER_DB
+#include "ha_ndbcluster.h"
+#endif
 
 static HASH system_variable_hash;
 const char *bool_type_names[]= { "OFF", "ON", NullS };
@@ -638,6 +641,7 @@ struct show_var_st init_vars[]= {
   {"have_crypt",	      (char*) &have_crypt,		    SHOW_HAVE},
   {"have_innodb",	      (char*) &have_innodb,		    SHOW_HAVE},
   {"have_isam",		      (char*) &have_isam,		    SHOW_HAVE},
+  {"have_ndbcluster",         (char*) &have_ndbcluster,             SHOW_HAVE},
   {"have_openssl",	      (char*) &have_openssl,		    SHOW_HAVE},
   {"have_query_cache",        (char*) &have_query_cache,            SHOW_HAVE},
   {"have_raid",		      (char*) &have_raid,		    SHOW_HAVE},

sql/sql_base.cc

@@ -1317,18 +1317,34 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
 {
   char path[FN_REFLEN];
   int error;
+  uint discover_retry_count= 0;
   DBUG_ENTER("open_unireg_entry");
 
   strxmov(path, mysql_data_home, "/", db, "/", name, NullS);
-  if (openfrm(path,alias,
+  while (openfrm(path,alias,
 	       (uint) (HA_OPEN_KEYFILE | HA_OPEN_RNDFILE | HA_GET_INDEX |
 		       HA_TRY_READ_ONLY),
 	       READ_KEYINFO | COMPUTE_TYPES | EXTRA_RECORD,
 	      thd->open_options, entry))
   {
     if (!entry->crashed)
-      goto err;					// Can't repair the table
+    {
+      /*
+       Frm file could not be found on disk
+       Since it does not exist, no one can be using it
+       LOCK_open has been locked to protect from someone else
+       trying to discover the table at the same time.
+      */
+      if (discover_retry_count++ != 0)
+       goto err;
+      if (create_table_from_handler(db, name, true) != 0)
+       goto err;
+
+      thd->clear_error(); // Clear error message
+      continue;
+    }
 
+    // Code below is for repairing a crashed file
     TABLE_LIST table_list;
     bzero((char*) &table_list, sizeof(table_list)); // just for safe
     table_list.db=(char*) db;
@@ -1374,6 +1390,7 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
 
     if (error)
       goto err;
+    break;
   }
   /*
     If we are here, there was no fatal error (but error may be still

sql/sql_class.h

@@ -688,7 +688,10 @@ public:
     THD_TRANS all;			// Trans since BEGIN WORK
     THD_TRANS stmt;			// Trans for current statement
     uint bdb_lock_count;
-
+    uint ndb_lock_count;
+#ifdef HAVE_NDBCLUSTER_DB
+    void* ndb;
+#endif
     /*
        Tables changed in transaction (that must be invalidated in query cache).
        List contain only transactional tables, that not invalidated in query
@@ -878,7 +881,8 @@ public:
   {
 #ifdef USING_TRANSACTIONS    
     return (transaction.all.bdb_tid != 0 ||
-	    transaction.all.innodb_active_trans != 0);
+	    transaction.all.innodb_active_trans != 0 ||
+	    transaction.all.ndb_tid != 0);
 #else
     return 0;
 #endif

sql/sql_table.cc

@@ -1148,6 +1148,35 @@ int mysql_create_table(THD *thd,const char *db, const char *table_name,
     }
   }
 
+  /*
+    Check that table with given name does not already
+    exist in any storage engine. In such a case it should
+    be discovered and the error ER_TABLE_EXISTS_ERROR be returned
+    unless user specified CREATE TABLE IF EXISTS
+    The LOCK_open mutex has been locked to make sure no
+    one else is attempting to discover the table. Since
+    it's not on disk as a frm file, no one could be using it!
+  */
+  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+  {
+    bool create_if_not_exists =
+      create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS;
+    if (!create_table_from_handler(db, table_name,
+                                 create_if_not_exists))
+    {
+      DBUG_PRINT("info", ("Table already existed in handler"));
+
+      if (create_if_not_exists)
+      {
+       create_info->table_existed= 1;   // Mark that table existed
+       error= 0;
+      }
+      else
+       my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
+      goto end;
+    }
+  }
+
   thd->proc_info="creating table";
   create_info->table_existed= 0;		// Mark that table is created
 

sql/sql_yacc.yy

@@ -181,7 +181,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %token	ACTION
 %token	AGGREGATE_SYM
 %token	ALL
-%token	AND
+%token	AND_SYM
 %token	AS
 %token	ASC
 %token	AUTO_INC
@@ -305,6 +305,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %token	NAMES_SYM
 %token	NATIONAL_SYM
 %token	NATURAL
+%token  NDBCLUSTER_SYM
 %token	NEW_SYM
 %token	NCHAR_SYM
 %token	NCHAR_STRING
@@ -318,7 +319,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %token	OPEN_SYM
 %token	OPTION
 %token	OPTIONALLY
-%token	OR
+%token	OR_SYM
 %token	OR_OR_CONCAT
 %token	ORDER_SYM
 %token	OUTER
@@ -574,8 +575,8 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 
 %token  BEFORE_SYM
 %left   SET_VAR
-%left	OR_OR_CONCAT OR XOR
-%left	AND
+%left	OR_OR_CONCAT OR_SYM XOR
+%left	AND_SYM
 %left	BETWEEN_SYM CASE_SYM WHEN_SYM THEN_SYM ELSE
 %left	EQ EQUAL_SYM GE GT_SYM LE LT NE IS LIKE REGEXP IN_SYM
 %left	'|'
@@ -727,7 +728,7 @@ END_OF_INPUT
 
 %type <NONE>
 	'-' '+' '*' '/' '%' '(' ')'
-	',' '!' '{' '}' '&' '|' AND OR OR_OR_CONCAT BETWEEN_SYM CASE_SYM
+	',' '!' '{' '}' '&' '|' AND_SYM OR_SYM OR_OR_CONCAT BETWEEN_SYM CASE_SYM
 	THEN_SYM WHEN_SYM DIV_SYM MOD_SYM
 %%
 
@@ -2421,14 +2422,14 @@ expr_expr:
           {
             $$= new Item_func_not(new Item_in_subselect($1, $4));
           }
-	| expr BETWEEN_SYM no_and_expr AND expr
+	| expr BETWEEN_SYM no_and_expr AND_SYM expr
 	  { $$= new Item_func_between($1,$3,$5); }
-	| expr NOT BETWEEN_SYM no_and_expr AND expr
+	| expr NOT BETWEEN_SYM no_and_expr AND_SYM expr
 	  { $$= new Item_func_not(new Item_func_between($1,$4,$6)); }
 	| expr OR_OR_CONCAT expr { $$= or_or_concat(YYTHD, $1,$3); }
-	| expr OR expr		{ $$= new Item_cond_or($1,$3); }
+	| expr OR_SYM expr	{ $$= new Item_cond_or($1,$3); }
         | expr XOR expr		{ $$= new Item_cond_xor($1,$3); }
-	| expr AND expr		{ $$= new Item_cond_and($1,$3); }
+	| expr AND_SYM expr	{ $$= new Item_cond_and($1,$3); }
 	| expr SOUNDS_SYM LIKE expr
 	  {
 	    $$= new Item_func_eq(new Item_func_soundex($1),
@@ -2469,14 +2470,14 @@ expr_expr:
 
 /* expressions that begin with 'expr' that do NOT follow IN_SYM */
 no_in_expr:
-	no_in_expr BETWEEN_SYM no_and_expr AND expr
+	no_in_expr BETWEEN_SYM no_and_expr AND_SYM expr
 	  { $$= new Item_func_between($1,$3,$5); }
-	| no_in_expr NOT BETWEEN_SYM no_and_expr AND expr
+	| no_in_expr NOT BETWEEN_SYM no_and_expr AND_SYM expr
 	  { $$= new Item_func_not(new Item_func_between($1,$4,$6)); }
 	| no_in_expr OR_OR_CONCAT expr	{ $$= or_or_concat(YYTHD, $1,$3); }
-	| no_in_expr OR expr		{ $$= new Item_cond_or($1,$3); }
+	| no_in_expr OR_SYM expr	{ $$= new Item_cond_or($1,$3); }
         | no_in_expr XOR expr		{ $$= new Item_cond_xor($1,$3); }
-	| no_in_expr AND expr		{ $$= new Item_cond_and($1,$3); }
+	| no_in_expr AND_SYM expr	{ $$= new Item_cond_and($1,$3); }
 	| no_in_expr SOUNDS_SYM LIKE expr
 	  {
 	    $$= new Item_func_eq(new Item_func_soundex($1),
@@ -2527,12 +2528,12 @@ no_and_expr:
           {
             $$= new Item_func_not(new Item_in_subselect($1, $4));
           }
-	| no_and_expr BETWEEN_SYM no_and_expr AND expr
+	| no_and_expr BETWEEN_SYM no_and_expr AND_SYM expr
 	  { $$= new Item_func_between($1,$3,$5); }
-	| no_and_expr NOT BETWEEN_SYM no_and_expr AND expr
+	| no_and_expr NOT BETWEEN_SYM no_and_expr AND_SYM expr
 	  { $$= new Item_func_not(new Item_func_between($1,$4,$6)); }
 	| no_and_expr OR_OR_CONCAT expr	{ $$= or_or_concat(YYTHD, $1,$3); }
-	| no_and_expr OR expr		{ $$= new Item_cond_or($1,$3); }
+	| no_and_expr OR_SYM expr	{ $$= new Item_cond_or($1,$3); }
         | no_and_expr XOR expr		{ $$= new Item_cond_xor($1,$3); }
 	| no_and_expr SOUNDS_SYM LIKE expr
 	  {
@@ -4147,8 +4148,8 @@ show_param:
 	      YYABORT;
 	  }
         | NEW_SYM MASTER_SYM FOR_SYM SLAVE WITH MASTER_LOG_FILE_SYM EQ
-	  TEXT_STRING_sys AND MASTER_LOG_POS_SYM EQ ulonglong_num
-	  AND MASTER_SERVER_ID_SYM EQ
+	  TEXT_STRING_sys AND_SYM MASTER_LOG_POS_SYM EQ ulonglong_num
+	  AND_SYM MASTER_SERVER_ID_SYM EQ
 	ULONG_NUM
           {
 	    Lex->sql_command = SQLCOM_SHOW_NEW_MASTER;
@@ -4983,6 +4984,7 @@ keyword:
 	| NAMES_SYM		{}
 	| NATIONAL_SYM		{}
 	| NCHAR_SYM		{}
+	| NDBCLUSTER_SYM	{}
 	| NEXT_SYM		{}
 	| NEW_SYM		{}
 	| NO_SYM		{}
@@ -5434,7 +5436,7 @@ grant_privilege:
 
 opt_and:
 	/* empty */	{}
-	| AND		{}
+	| AND_SYM	{}
 	;
 
 require_list:

sql/table.cc

@@ -944,7 +944,8 @@ static void frm_error(int error, TABLE *form, const char *name, myf errortype)
     break;
   case 2:
   {
-    datext=form->file ? *form->file->bas_ext() : "";
+    datext= form->file ? *form->file->bas_ext() : "";
+    datext= datext==NullS ? "" : datext;
     err_no= (my_errno == ENOENT) ? ER_FILE_NOT_FOUND : (my_errno == EAGAIN) ?
       ER_FILE_USED : ER_CANT_OPEN_FILE;
     my_error(err_no,errortype,