Added NDB storage engine

include/my_base.h:
  Added three new errorcodes to be returned by the handler
sql/Makefile.am:
  Add new files discover.cc, ha_ndbcluster.cc and ha_ndbcluster.h
  Add include path of NDB files
sql/handler.cc:
  Added variable for keeping track of number of "discovers"
  Added NDB to list of storage engines
  Added calls to NDB for commit, rollback etc.
  Added function ha_discover for discovering a table from handler
sql/handler.h:
  Added NDB to list of storage engines
  Added vbariable in transaction for keeping a ndb transaction handle
sql/lex.h:
  Changed AND to AND_SYM and OR to OR_SYM to avoid nameclash
sql/mysql_priv.h:
  Added prototypes for new functions readfrm, writefrm and create_table_from_handler
sql/mysqld.cc:
  Added NDB support
  Disable NDB with --skip-ndbcluster
sql/set_var.cc:
  Add posibilty to show if NDB handler is supported
sql/ha_ndbcluster.cc:
  Add ifdef for whole file for not compiling anything if NDB sholdn't be included
  Updated timestamp handling to use new vars timestamp_default_now and timestamp_on_update_now
sql/sql_base.cc:
  If frm file is not found on disk, ask handler if it knows about the table. Then retry the open.
  This new functionality is called "discover" and can be used by any handler.
sql/sql_class.h:
  Added variable for keeping a NDB connection handle
sql/sql_table.cc:
  Before trying to create a table, ask handler if a table with that name already exists.
  If user said CREATE TABLE IF NOT EXISTS, disocver the table from handler
sql/sql_yacc.yy:
  Add NDBCLUSTER_SYM
  Change AND to AND_SYM
  Change OR to OR_SYM
sql/table.cc:
  Fixe for probelm when NullS is returned from bas_ext of a handler.

15 files changed, 3561 insertions, 48 deletions

diff --git a/sql/Makefile.am b/sql/Makefile.am
index 4aa6aaaa1ee..6845e0b5de8 100644
--- a/sql/Makefile.am
+++ b/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 \
diff --git a/sql/discover.cc b/sql/discover.cc
new file mode 100644
index 00000000000..e260f44a8db
--- /dev/null
+++ b/sql/discover.cc
@@ -0,0 +1,172 @@
+/* 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);
+}
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
new file mode 100755
index 00000000000..3bc322878d1
--- /dev/null
+++ b/sql/ha_ndbcluster.cc
@@ -0,0 +1,2943 @@
+/* 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 */
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
new file mode 100755
index 00000000000..ed66d07d79b
--- /dev/null
+++ b/sql/ha_ndbcluster.h
@@ -0,0 +1,218 @@
+/* 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);
+
+
+
+
+
+
+
+
diff --git a/sql/handler.cc b/sql/handler.cc
index 2b22563bc3c..e1ee3315f40 100644
--- a/sql/handler.cc
+++ b/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}
 };
 
@@ -172,6 +179,10 @@ handler *get_new_handler(TABLE *table, enum db_type db_type)
   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);
   default:					// should never happen
@@ -216,6 +227,18 @@ int ha_init()
       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;
 }
 
@@ -243,6 +266,10 @@ int ha_panic(enum ha_panic_function flag)
   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))
@@ -1290,6 +1352,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;
 }
+
+
diff --git a/sql/handler.h b/sql/handler.h
index 4cb6ab86a37..a5d4f78ef29 100644
--- a/sql/handler.h
+++ b/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);
diff --git a/sql/lex.h b/sql/lex.h
index 94ea0295f05..e5bc537c213 100644
--- a/sql/lex.h
+++ b/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)},
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 1175b93b9ba..9e7db79a7b2 100644
--- a/sql/mysql_priv.h
+++ b/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);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 5176ee33a17..2e0e8493aad 100644
--- a/sql/mysqld.cc
+++ b/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
@@ -5591,6 +5607,14 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
       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:
 #ifdef HAVE_INNOBASE_DB
     if (opt_innodb)
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 5ccda5c7052..0cf4d2d2691 100644
--- a/sql/set_var.cc
+++ b/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},
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 8c4571dd540..d24b13ff96f 100644
--- a/sql/sql_base.cc
+++ b/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
diff --git a/sql/sql_class.h b/sql/sql_class.h
index aa526d5e474..d5eb7a9fd0e 100644
--- a/sql/sql_class.h
+++ b/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
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index d5f77bf545d..c46a9823a52 100644
--- a/sql/sql_table.cc
+++ b/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
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index b87b0b29677..568a526fd58 100644
--- a/sql/sql_yacc.yy
+++ b/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:
diff --git a/sql/table.cc b/sql/table.cc
index 1b7d30560ef..281a8c10409 100644
--- a/sql/table.cc
+++ b/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,