diff --git a/BitKeeper/etc/logging_ok b/BitKeeper/etc/logging_ok
index 6c57b7254e0..e3aefbf3efc 100644
--- a/BitKeeper/etc/logging_ok
+++ b/BitKeeper/etc/logging_ok
@@ -94,6 +94,7 @@ kostja@oak.local
 lenz@kallisto.mysql.com
 lenz@mysql.com
 magnus@neptunus.(none)
+magnus@shellback.(none)
 marko@hundin.mysql.fi
 miguel@hegel.(none)
 miguel@hegel.br
diff --git a/include/my_base.h b/include/my_base.h
index 0ef66ef8123..d884113dc4d 100644
--- a/include/my_base.h
+++ b/include/my_base.h
@@ -241,6 +241,7 @@ enum ha_base_keytype {
 #define HA_CREATE_TMP_TABLE	4
 #define HA_CREATE_CHECKSUM	8
 #define HA_CREATE_DELAY_KEY_WRITE 64
+#define HA_CREATE_FROM_ENGINE   128 
 
 	/* Bits in flag to _status */
 
@@ -287,7 +288,7 @@ enum ha_base_keytype {
 #define HA_ERR_ROW_IS_REFERENCED 152     /* Cannot delete a parent row */
 #define HA_ERR_NO_SAVEPOINT	 153     /* No savepoint with that name */
 #define HA_ERR_NON_UNIQUE_BLOCK_SIZE 154 /* Non unique key block size */
-#define HA_ERR_OLD_METADATA      155  /* The frm file on disk is old */
+#define HA_ERR_NO_SUCH_TABLE     155  /* The table does not exist in engine */
 #define HA_ERR_TABLE_EXIST       156  /* The table existed in storage engine */
 #define HA_ERR_NO_CONNECTION     157  /* Could not connect to storage engine */
 
diff --git a/mysql-test/include/have_ndb.inc b/mysql-test/include/have_ndb.inc
index 095c2dcaaa4..7567a0ca362 100644
--- a/mysql-test/include/have_ndb.inc
+++ b/mysql-test/include/have_ndb.inc
@@ -2,3 +2,6 @@
 disable_query_log;
 show variables like "have_ndbcluster";
 enable_query_log;
+connect (server1,127.0.0.1,root,,test,$MASTER_MYPORT,$MASTER_MYSOCK);
+connect (server2,127.0.0.1,root,,test,$MASTER_MYPORT1,$MASTER_MYSOCK1);
+connection server1;
diff --git a/mysql-test/install_test_db.sh b/mysql-test/install_test_db.sh
index 8db338f0f58..4554b92857e 100644
--- a/mysql-test/install_test_db.sh
+++ b/mysql-test/install_test_db.sh
@@ -36,8 +36,13 @@ then
  data=var/slave-data
  ldata=$fix_bin/var/slave-data
 else
- data=var/master-data
- ldata=$fix_bin/var/master-data
+ if [ x$1 = x"-1" ] 
+ then
+   data=var/master-data1
+ else
+   data=var/master-data
+ fi
+ ldata=$fix_bin/$data
 fi
 
 mdata=$data/mysql
diff --git a/mysql-test/mysql-test-run.sh b/mysql-test/mysql-test-run.sh
index c447e96590f..930087f880a 100644
--- a/mysql-test/mysql-test-run.sh
+++ b/mysql-test/mysql-test-run.sh
@@ -198,9 +198,11 @@ DYLD_LIBRARY_PATH="$BASEDIR/lib:$BASEDIR/libmysql/.libs:$DYLD_LIBRARY_PATH"
 export LD_LIBRARY_PATH DYLD_LIBRARY_PATH
 
 MASTER_RUNNING=0
+MASTER1_RUNNING=0
 MASTER_MYPORT=9306
+MASTER_MYPORT1=9307
 SLAVE_RUNNING=0
-SLAVE_MYPORT=9307
+SLAVE_MYPORT=9308 # leave room for 2 masters for cluster tests
 MYSQL_MANAGER_PORT=9305 # needs to be out of the way of slaves
 NDBCLUSTER_PORT=9350
 MYSQL_MANAGER_PW_FILE=$MYSQL_TEST_DIR/var/tmp/manager.pwd
@@ -254,9 +256,9 @@ while test $# -gt 0; do
     --extern)  USE_RUNNING_SERVER="1" ;;
     --with-ndbcluster)
       USE_NDBCLUSTER="--ndbcluster" ;;
-    --ndbconnectstring=*)
+    --ndb-connectstring=*)
       USE_NDBCLUSTER="--ndbcluster" ;
-      USE_RUNNING_NDBCLUSTER=`$ECHO "$1" | $SED -e "s;--ndbconnectstring=;;"` ;;
+      USE_RUNNING_NDBCLUSTER=`$ECHO "$1" | $SED -e "s;--ndb-connectstring=;;"` ;;
     --tmpdir=*) MYSQL_TMP_DIR=`$ECHO "$1" | $SED -e "s;--tmpdir=;;"` ;;
     --local-master)
       MASTER_MYPORT=3306;
@@ -437,6 +439,7 @@ MANAGER_PID_FILE="$MYRUN_DIR/manager.pid"
 
 MASTER_MYDDIR="$MYSQL_TEST_DIR/var/master-data"
 MASTER_MYSOCK="$MYSQL_TMP_DIR/master.sock"
+MASTER_MYSOCK1=$MYSQL_MYSOCK"1"
 MASTER_MYPID="$MYRUN_DIR/master.pid"
 MASTER_MYLOG="$MYSQL_TEST_DIR/var/log/master.log"
 MASTER_MYERR="$MYSQL_TEST_DIR/var/log/master.err"
@@ -450,7 +453,7 @@ SLAVE_MYERR="$MYSQL_TEST_DIR/var/log/slave.err"
 CURRENT_TEST="$MYSQL_TEST_DIR/var/log/current_test"
 SMALL_SERVER="--key_buffer_size=1M --sort_buffer=256K --max_heap_table_size=1M"
 
-export MASTER_MYPORT SLAVE_MYPORT MYSQL_TCP_PORT MASTER_MYSOCK
+export MASTER_MYPORT MASTER_MYPORT1 SLAVE_MYPORT MYSQL_TCP_PORT MASTER_MYSOCK MASTER_MYSOCK1
 
 if [ x$SOURCE_DIST = x1 ] ; then
  MY_BASEDIR=$MYSQL_TEST_DIR
@@ -518,6 +521,7 @@ if [ x$SOURCE_DIST = x1 ] ; then
  CHARSETSDIR="$BASEDIR/sql/share/charsets"
  INSTALL_DB="./install_test_db"
  MYSQL_FIX_SYSTEM_TABLES="$BASEDIR/scripts/mysql_fix_privilege_tables"
+ NDB_TOOLS_DIR="$BASEDIR/ndb/tools"
 else
  if test -x "$BASEDIR/libexec/mysqld"
  then
@@ -538,6 +542,7 @@ else
  MYSQL="$CLIENT_BINDIR/mysql"
  INSTALL_DB="./install_test_db --bin"
  MYSQL_FIX_SYSTEM_TABLES="$CLIENT_BINDIR/mysql_fix_privilege_tables"
+ NDB_TOOLS_DIR="$CLIENT_BINDIR"
  if test -d "$BASEDIR/share/mysql/english"
  then
    LANGUAGE="$BASEDIR/share/mysql/english/"
@@ -585,6 +590,7 @@ MYSQL_BINLOG="$MYSQL_BINLOG --no-defaults --local-load=$MYSQL_TMP_DIR $EXTRA_MYS
 MYSQL_FIX_SYSTEM_TABLES="$MYSQL_FIX_SYSTEM_TABLES --no-defaults --host=localhost --port=$MASTER_MYPORT --socket=$MASTER_MYSOCK --user=root --password=$DBPASSWD --basedir=$BASEDIR --bindir=$CLIENT_BINDIR --verbose"
 MYSQL="$MYSQL --host=localhost --port=$MASTER_MYPORT --socket=$MASTER_MYSOCK --user=root --password=$DBPASSWD"
 export MYSQL MYSQL_DUMP MYSQL_BINLOG MYSQL_FIX_SYSTEM_TABLES CLIENT_BINDIR TESTS_BINDIR
+export NDB_TOOLS_DIR
 
 MYSQL_TEST_ARGS="--no-defaults --socket=$MASTER_MYSOCK --database=$DB \
  --user=$DBUSER --password=$DBPASSWD --silent -v --skip-safemalloc \
@@ -751,13 +757,22 @@ report_stats () {
 
 mysql_install_db () {
     $ECHO "Removing Stale Files"
-    $RM -rf $MASTER_MYDDIR $SLAVE_MYDDIR $MY_LOG_DIR/* 
+    $RM -rf $MASTER_MYDDIR $MASTER_MYDDIR"1" $SLAVE_MYDDIR $MY_LOG_DIR/* 
     $ECHO "Installing Master Databases"
     $INSTALL_DB
     if [ $? != 0 ]; then
 	error "Could not install master test DBs"
 	exit 1
     fi
+    if [ ! -z "$USE_NDBCLUSTER" ]
+    then
+      $ECHO "Installing Master Databases 1"
+      $INSTALL_DB -1
+      if [ $? != 0 ]; then
+	error "Could not install master test DBs 1"
+	exit 1
+      fi
+    fi
     $ECHO "Installing Slave Databases"
     $INSTALL_DB -slave
     if [ $? != 0 ]; then
@@ -896,10 +911,10 @@ manager_term()
   ident=$2
   if [ $USE_MANAGER = 0 ] ; then
     # Shutdown time must be high as slave may be in reconnect
-    $MYSQLADMIN --no-defaults -uroot --socket=$MYSQL_TMP_DIR/$ident.sock --connect_timeout=5 --shutdown_timeout=70 shutdown >> $MYSQL_MANAGER_LOG 2>&1
+    $MYSQLADMIN --no-defaults -uroot --socket=$MYSQL_TMP_DIR/$ident.sock$3 --connect_timeout=5 --shutdown_timeout=70 shutdown >> $MYSQL_MANAGER_LOG 2>&1
     res=$?
     # Some systems require an extra connect
-    $MYSQLADMIN --no-defaults -uroot --socket=$MYSQL_TMP_DIR/$ident.sock --connect_timeout=1 ping >> $MYSQL_MANAGER_LOG 2>&1
+    $MYSQLADMIN --no-defaults -uroot --socket=$MYSQL_TMP_DIR/$ident.sock$3 --connect_timeout=1 ping >> $MYSQL_MANAGER_LOG 2>&1
     if test $res = 0
     then
       wait_for_pid $pid
@@ -918,17 +933,18 @@ EOF
 
 start_master()
 {
-  if [ x$MASTER_RUNNING = x1 ] || [ x$LOCAL_MASTER = x1 ] ; then
+  eval "this_master_running=\$MASTER$1_RUNNING"
+  if [ x$this_master_running = x1 ] || [ x$LOCAL_MASTER = x1 ] ; then
     return
   fi
   # Remove stale binary logs except for 2 tests which need them
   if [ "$tname" != "rpl_crash_binlog_ib_1b" ] && [ "$tname" != "rpl_crash_binlog_ib_2b" ] && [ "$tname" != "rpl_crash_binlog_ib_3b" ] 
   then
-    $RM -f $MYSQL_TEST_DIR/var/log/master-bin.*
+    $RM -f $MYSQL_TEST_DIR/var/log/master-bin$1.*
   fi
 
   # Remove old master.info and relay-log.info files
-  $RM -f $MYSQL_TEST_DIR/var/master-data/master.info $MYSQL_TEST_DIR/var/master-data/relay-log.info
+  $RM -f $MYSQL_TEST_DIR/var/master-data$1/master.info $MYSQL_TEST_DIR/var/master-data$1/relay-log.info
 
   #run master initialization shell script if one exists
 
@@ -937,20 +953,27 @@ start_master()
       /bin/sh $master_init_script
   fi
   cd $BASEDIR # for gcov
+  if [ -n "$1" ] ; then
+   id=`$EXPR $1 + 101`;
+   this_master_myport=`$EXPR $MASTER_MYPORT + $1`
+  else
+   id=1;
+   this_master_myport=$MASTER_MYPORT
+  fi
   if [ -z "$DO_BENCH" ]
   then
-    master_args="--no-defaults --log-bin=$MYSQL_TEST_DIR/var/log/master-bin \
-  	    --server-id=1  \
+    master_args="--no-defaults --log-bin=$MYSQL_TEST_DIR/var/log/master-bin$1 \
+  	    --server-id=$id  \
           --basedir=$MY_BASEDIR \
-          --port=$MASTER_MYPORT \
+          --port=$this_master_myport \
           --local-infile \
           --exit-info=256 \
           --core \
           $USE_NDBCLUSTER \
-          --datadir=$MASTER_MYDDIR \
-          --pid-file=$MASTER_MYPID \
-          --socket=$MASTER_MYSOCK \
-          --log=$MASTER_MYLOG \
+          --datadir=$MASTER_MYDDIR$1 \
+          --pid-file=$MASTER_MYPID$1 \
+          --socket=$MASTER_MYSOCK$1 \
+          --log=$MASTER_MYLOG$1 \
           --character-sets-dir=$CHARSETSDIR \
           --default-character-set=$CHARACTER_SET \
           --tmpdir=$MYSQL_TMP_DIR \
@@ -961,14 +984,14 @@ start_master()
            $SMALL_SERVER \
            $EXTRA_MASTER_OPT $EXTRA_MASTER_MYSQLD_OPT"
   else
-    master_args="--no-defaults --log-bin=$MYSQL_TEST_DIR/var/log/master-bin \
-          --server-id=1 --rpl-recovery-rank=1 \
+    master_args="--no-defaults --log-bin=$MYSQL_TEST_DIR/var/log/master-bin$1 \
+          --server-id=$id --rpl-recovery-rank=1 \
           --basedir=$MY_BASEDIR --init-rpl-role=master \
-          --port=$MASTER_MYPORT \
+          --port=$this_master_myport \
           --local-infile \
-          --datadir=$MASTER_MYDDIR \
-          --pid-file=$MASTER_MYPID \
-          --socket=$MASTER_MYSOCK \
+          --datadir=$MASTER_MYDDIR$1 \
+          --pid-file=$MASTER_MYPID$1 \
+          --socket=$MASTER_MYSOCK$1 \
           --character-sets-dir=$CHARSETSDIR \
           --default-character-set=$CHARACTER_SET \
           --core \
@@ -1026,9 +1049,9 @@ EOF
   else
     manager_launch master $MASTER_MYSQLD $master_args
   fi
-  sleep_until_file_created $MASTER_MYPID $wait_for_master
+  sleep_until_file_created $MASTER_MYPID$1 $wait_for_master
   wait_for_master=$SLEEP_TIME_FOR_SECOND_MASTER
-  MASTER_RUNNING=1
+  eval "MASTER$1_RUNNING=1"
 }
 
 start_slave()
@@ -1204,22 +1227,23 @@ stop_slave_threads ()
 
 stop_master ()
 {
-  if [ x$MASTER_RUNNING = x1 ]
+  eval "this_master_running=\$MASTER$1_RUNNING"
+  if [ x$this_master_running = x1 ]
   then
     # For embedded server we don't stop anyting but mark that
     # MASTER_RUNNING=0 to get cleanup when calling start_master().
     if [ x$USE_EMBEDDED_SERVER != x1 ] ; then
-      pid=`$CAT $MASTER_MYPID`
-      manager_term $pid master
-      if [ $? != 0 ] && [ -f $MASTER_MYPID ]
+      pid=`$CAT $MASTER_MYPID$1`
+      manager_term $pid master $1
+      if [ $? != 0 ] && [ -f $MASTER_MYPID$1 ]
       then # try harder!
 	$ECHO "master not cooperating with mysqladmin, will try manual kill"
 	kill $pid
-	sleep_until_file_deleted $pid $MASTER_MYPID
-	if [ -f $MASTER_MYPID ] ; then
+	sleep_until_file_deleted $pid $MASTER_MYPID$1
+	if [ -f $MASTER_MYPID$1 ] ; then
 	  $ECHO "master refused to die. Sending SIGKILL"
-	  kill -9 `$CAT $MASTER_MYPID`
-	  $RM -f $MASTER_MYPID
+	  kill -9 `$CAT $MASTER_MYPID$1`
+	  $RM -f $MASTER_MYPID$1
 	else
 	  $ECHO "master responded to SIGTERM "
 	fi
@@ -1227,7 +1251,7 @@ stop_master ()
 	sleep $SLEEP_TIME_AFTER_RESTART
       fi
     fi
-    MASTER_RUNNING=0
+    eval "MASTER$1_RUNNING=0"
   fi
 }
 
@@ -1237,6 +1261,7 @@ mysql_stop ()
  $ECHO  "Shutting-down MySQL daemon"
  $ECHO  ""
  stop_master
+ stop_master 1
  $ECHO "Master shutdown finished"
  stop_slave
  stop_slave 1
@@ -1367,8 +1392,12 @@ run_testcase ()
          ;;
      esac
      stop_master
+     stop_master 1
      echo "CURRENT_TEST: $tname" >> $MASTER_MYERR
      start_master
+     if [ -n "$USE_NDBCLUSTER" -a -z "$DO_BENCH" ] ; then
+       start_master 1
+     fi
      TZ=$MY_TZ; export TZ
    else
      # If we had extra master opts to the previous run
@@ -1379,8 +1408,12 @@ run_testcase ()
      then
        EXTRA_MASTER_OPT=""
        stop_master
+       stop_master 1
        echo "CURRENT_TEST: $tname" >> $MASTER_MYERR
        start_master
+       if [ -n "$USE_NDBCLUSTER"  -a -z "$DO_BENCH" ] ; then
+         start_master 1
+       fi
      else
        echo "CURRENT_TEST: $tname" >> $MASTER_MYERR
      fi
@@ -1534,11 +1567,11 @@ then
 
   if [ ! -z "$USE_NDBCLUSTER" ]
   then
-  if [ -z "$USE_RUNNING_NDBCLUSTER" ]
-  then
-    # Kill any running ndbcluster stuff
-    ./ndb/ndbcluster --data-dir=$MYSQL_TEST_DIR/var --port-base=$NDBCLUSTER_PORT --stop
-  fi
+    if [ -z "$USE_RUNNING_NDBCLUSTER" ]
+    then
+      # Kill any running ndbcluster stuff
+      ./ndb/ndbcluster --data-dir=$MYSQL_TEST_DIR/var --port-base=$NDBCLUSTER_PORT --stop
+    fi
   fi
 
   # Remove files that can cause problems
@@ -1547,6 +1580,7 @@ then
 
   # Remove old berkeley db log files that can confuse the server
   $RM -f $MASTER_MYDDIR/log.*
+  $RM -f $MASTER_MYDDIR"1"/log.*
 
   wait_for_master=$SLEEP_TIME_FOR_FIRST_MASTER
   wait_for_slave=$SLEEP_TIME_FOR_FIRST_SLAVE
@@ -1558,7 +1592,7 @@ then
   if [ -z "$USE_RUNNING_NDBCLUSTER" ]
   then
     echo "Starting ndbcluster"
-    if [ "$DO_BENCH" = 1 ]
+    if [ "$DO_BENCH" = 1  -a ! "$DO_SMALL_BENCH" = 1 ]
     then
       NDBCLUSTER_OPTS=""
     else
@@ -1566,10 +1600,13 @@ then
     fi
     ./ndb/ndbcluster --port-base=$NDBCLUSTER_PORT $NDBCLUSTER_OPTS --diskless --initial --data-dir=$MYSQL_TEST_DIR/var || exit 1
     USE_NDBCLUSTER="$USE_NDBCLUSTER --ndb-connectstring=\"host=localhost:$NDBCLUSTER_PORT\""
+     NDB_CONNECTSTRING="localhost:$NDBCLUSTER_PORT"
   else
     USE_NDBCLUSTER="$USE_NDBCLUSTER --ndb-connectstring=\"$USE_RUNNING_NDBCLUSTER\""
+    NDB_CONNECTSTRING="$USE_RUNNING_NDBCLUSTER"
     echo "Using ndbcluster at $USE_NDBCLUSTER"
   fi
+  export NDB_CONNECTSTRING
   fi
 
   start_manager
diff --git a/mysql-test/r/ndb_autodiscover.result b/mysql-test/r/ndb_autodiscover.result
index f5b908c39e2..031be91fcb2 100644
--- a/mysql-test/r/ndb_autodiscover.result
+++ b/mysql-test/r/ndb_autodiscover.result
@@ -1,4 +1,4 @@
-drop table if exists t1,t2,t3,t4,t5,t6,t9;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
 flush status;
 create table t1(
 id int not null primary key,
@@ -94,8 +94,6 @@ ERROR 42S01: Table 't3' already exists
 show status like 'handler_discover%';
 Variable_name	Value
 Handler_discover	1
-SHOW TABLES FROM test;
-Tables_in_test
 create table IF NOT EXISTS t3(
 id int not null primary key,
 id2 int not null,
@@ -119,6 +117,162 @@ Variable_name	Value
 Handler_discover	2
 drop table t3;
 flush status;
+create table t7(
+id int not null primary key,
+name char(255)
+) engine=ndb;
+create table t6(
+id int not null primary key,
+name char(255)
+) engine=MyISAM;
+insert into t7 values (1, "Explorer");
+insert into t6 values (2, "MyISAM table");
+select * from t7;
+id	name
+1	Explorer
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	0
+flush tables;
+show tables from test;
+Tables_in_test
+t6
+t7
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	1
+flush tables;
+show table status;
+Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
+t6	MyISAM	9	Fixed	1	260	#	#	#	0	NULL	#	#	NULL	#	NULL		
+t7	ndbcluster	9	Fixed	1	0	#	#	#	0	NULL	#	#	NULL	#	NULL		
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	2
+drop table t6, t7;
+flush status;
+create table t4(
+id int not null primary key,
+name char(27)
+) engine=ndb;
+insert into t4 values (1, "Automatic");
+select * from t4;
+id	name
+1	Automatic
+select * from t4;
+ERROR 42S02: Table 'test.t4' doesn't exist
+select * from t4;
+ERROR 42S02: Table 'test.t4' doesn't exist
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	0
+drop table t4;
+flush status;
+create table t4(
+id int not null primary key,
+id2 int,
+name char(27)
+) engine=ndb;
+insert into t4 values (1, 76, "Automatic2");
+select * from t4;
+id	id2	name
+1	76	Automatic2
+flush tables;
+SHOW TABLES;
+Tables_in_test
+select * from t4;
+ERROR 42S02: Table 'test.t4' doesn't exist
+flush status;
+create table t1(id int) engine=ndbcluster;
+create table t2(id int, b char(255)) engine=myisam;
+create table t3(id int, c char(255)) engine=ndbcluster;
+create table t4(id int) engine=myisam;
+create table t5(id int, d char(56)) engine=ndbcluster;
+create table t6(id int) engine=ndbcluster;
+create table t7(id int) engine=ndbcluster;
+create table t8(id int, e char(34)) engine=myisam;
+create table t9(id int) engine=myisam;
+insert into t2 values (2, "myisam table 2");
+insert into t3 values (3, "ndb table 3");
+insert into t5 values (5, "ndb table 5");
+insert into t6 values (6);
+insert into t8 values (8, "myisam table 8");
+insert into t9 values (9);
+SHOW TABLES;
+Tables_in_test
+t1
+t2
+t4
+t8
+t9
+t7
+t6
+select * from t6;
+id
+6
+select * from t7;
+id
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	2
+drop table t1, t2, t4, t6, t7, t8, t9;
+flush status;
+create table t1(id int) engine=ndbcluster;
+create table t2(id int, b char(255)) engine=myisam;
+create table t3(id int, c char(255)) engine=ndbcluster;
+create table t4(id int) engine=myisam;
+create table t5(id int, d char(56)) engine=ndbcluster;
+create table t6(id int) engine=ndbcluster;
+create table t7(id int) engine=ndbcluster;
+create table t8(id int, e char(34)) engine=myisam;
+create table t9(id int) engine=myisam;
+insert into t2 values (2, "myisam table 2");
+insert into t3 values (3, "ndb table 3");
+insert into t5 values (5, "ndb table 5");
+insert into t6 values (6);
+insert into t8 values (8, "myisam table 8");
+insert into t9 values (9);
+SHOW TABLES LIKE 't6';
+Tables_in_test (t6)
+t6
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	1
+create table t3(a int);
+ERROR 42S01: Table 't3' already exists
+create table t5(a int);
+ERROR 42S01: Table 't5' already exists
+SHOW TABLES LIKE 't%';
+Tables_in_test (t%)
+t1
+t2
+t4
+t6
+t8
+t9
+t7
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	2
+drop table t1, t2, t4, t6, t7, t8, t9;
+flush status;
+create table t1(id int) engine=ndbcluster;
+create table t2(id int, b char(255)) engine=ndbcluster;
+create table t3(id int, c char(255)) engine=ndbcluster;
+create table t4(id int) engine=myisam;
+insert into t1 values (1);
+insert into t2 values (2, "table 2");
+insert into t3 values (3, "ndb table 3");
+insert into t4 values (4);
+flush tables;
+select * from t1, t2, t3, t4;
+id	id	b	id	c	id
+1	2	table 2	3	ndb table 3	4
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	3
+drop table t1, t2, t3, t4;
+flush status;
 show status like 'handler_discover%';
 Variable_name	Value
 Handler_discover	0
diff --git a/mysql-test/t/ndb_autodiscover.test b/mysql-test/t/ndb_autodiscover.test
index 371a130291b..13ec01cfb85 100644
--- a/mysql-test/t/ndb_autodiscover.test
+++ b/mysql-test/t/ndb_autodiscover.test
@@ -1,7 +1,7 @@
 -- source include/have_ndb.inc
 
 --disable_warnings
-drop table if exists t1,t2,t3,t4,t5,t6,t9;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
 --enable_warnings
 
 ################################################
@@ -122,7 +122,6 @@ create table t3(
 # IF NOT EXISTS wasn't specified
 
 show status like 'handler_discover%';
-SHOW TABLES FROM test;
 
 # now it should be discovered
 create table IF NOT EXISTS t3(
@@ -145,38 +144,226 @@ show status like 'handler_discover%';
 
 drop table t3;
 
-#######################################################
-# Test that a table that already exists as frm file
-# but not in NDB can be deleted from disk.
+##################################################
+# Test that a table that already exists in NDB 
+# is discovered when SHOW TABLES
+# is used
 #
 
-# Manual test
-#flush status;
-#
-#create table t4(
-#  id int not null primary key,
-#  name char(27)
-#) engine=ndb;
-#insert into t4 values (1, "Automatic");
-#select * from t4;
+flush status;
+
+create table t7(
+  id int not null primary key,
+  name char(255)
+) engine=ndb;
+create table t6(
+  id int not null primary key,
+  name char(255)
+) engine=MyISAM;
+insert into t7 values (1, "Explorer");
+insert into t6 values (2, "MyISAM table");
+select * from t7;
+show status like 'handler_discover%';
+
+# Remove the frm file from disk
+flush tables;
+system rm var/master-data/test/t7.frm ;
+
+show tables from test;
+show status like 'handler_discover%';
+
+# Remove the frm file from disk again
+flush tables;
+system rm var/master-data/test/t7.frm ;
+
+--replace_column 7 # 8 # 9 # 12 # 13 # 15 #
+show table status;
+show status like 'handler_discover%';
+
+drop table t6, t7;
+
+
+#######################################################
+# Test that a table that has been dropped from NDB
+# but still exists on disk, get a consistent error message
+# saying "No such table existed"
 #
+
+flush status;
+
+create table t4(
+  id int not null primary key,
+  name char(27)
+) engine=ndb;
+insert into t4 values (1, "Automatic");
+select * from t4;
+
 # Remove the table from NDB
-#system drop_tab -c "$NDB_CONNECTSTRING2" -d test t4 > /dev/null ;
-#system drop_tab -c "host=localhost:2200;nodeid=5" -d test t4 > /dev/null ;
+system exec $NDB_TOOLS_DIR/ndb_drop_table -d test t4 > /dev/null ; 
+
 #
-#--error 1296
-#select * from t4;
+# Test that correct error is returned
+--error 1146
+select * from t4;
+--error 1146
+select * from t4;
+
+show status like 'handler_discover%';
+drop table t4;
+
+
+#######################################################
+# Test that a table that has been dropped from NDB
+# but still exists on disk is deleted from disk 
+# when SHOW TABLES is called
 #
-#flush table t4;
-#--error 1016
-#select * from t4;
+
+flush status;
+
+create table t4(
+  id int not null primary key,
+  id2 int,
+  name char(27)
+) engine=ndb;
+insert into t4 values (1, 76, "Automatic2");
+select * from t4;
+flush tables;
+
+# Remove the table from NDB
+system exec $NDB_TOOLS_DIR/ndb_drop_table -d test t4 > /dev/null ; 
+
+SHOW TABLES;
+
+--error 1146
+select * from t4;
+
+#######################################################
+# Test SHOW TABLES ability to detect new and delete old
+# tables. Test all at once using many tables
 #
-#show status like 'handler_discover%';
-#drop table t4;
-#flush tables;
-#show tables;
-#--error 1146
-#select * from t4;
+
+flush status;
+
+# Create tables
+create table t1(id int) engine=ndbcluster;
+create table t2(id int, b char(255)) engine=myisam;
+create table t3(id int, c char(255)) engine=ndbcluster;
+create table t4(id int) engine=myisam;
+create table t5(id int, d char(56)) engine=ndbcluster;
+create table t6(id int) engine=ndbcluster;
+create table t7(id int) engine=ndbcluster;
+create table t8(id int, e char(34)) engine=myisam;
+create table t9(id int) engine=myisam;
+
+# Populate tables
+insert into t2 values (2, "myisam table 2");
+insert into t3 values (3, "ndb table 3");
+insert into t5 values (5, "ndb table 5");
+insert into t6 values (6);
+insert into t8 values (8, "myisam table 8");
+insert into t9 values (9);
+
+# Remove t3, t5 from NDB
+system exec $NDB_TOOLS_DIR/ndb_drop_table -d test t3 > /dev/null  ;
+system exec $NDB_TOOLS_DIR/ndb_drop_table -d test t5 > /dev/null  ;
+# Remove t6, t7 from disk
+system rm var/master-data/test/t6.frm > /dev/null ; 
+system rm var/master-data/test/t7.frm > /dev/null ; 
+
+SHOW TABLES;
+
+select * from t6;
+select * from t7;
+
+show status like 'handler_discover%';
+
+drop table t1, t2, t4, t6, t7, t8, t9;
+
+#######################################################
+# Test SHOW TABLES LIKE ability to detect new and delete old
+# tables. Test all at once using many tables.
+#
+
+flush status;
+
+# Create tables
+create table t1(id int) engine=ndbcluster;
+create table t2(id int, b char(255)) engine=myisam;
+create table t3(id int, c char(255)) engine=ndbcluster;
+create table t4(id int) engine=myisam;
+create table t5(id int, d char(56)) engine=ndbcluster;
+create table t6(id int) engine=ndbcluster;
+create table t7(id int) engine=ndbcluster;
+create table t8(id int, e char(34)) engine=myisam;
+create table t9(id int) engine=myisam;
+
+# Populate tables
+insert into t2 values (2, "myisam table 2");
+insert into t3 values (3, "ndb table 3");
+insert into t5 values (5, "ndb table 5");
+insert into t6 values (6);
+insert into t8 values (8, "myisam table 8");
+insert into t9 values (9);
+
+# Remove t3, t5 from NDB
+system exec $NDB_TOOLS_DIR/ndb_drop_table -d test t3 > /dev/null  ;
+system exec $NDB_TOOLS_DIR/ndb_drop_table -d test t5 > /dev/null  ;
+# Remove t6, t7 from disk
+system rm var/master-data/test/t6.frm > /dev/null ; 
+system rm var/master-data/test/t7.frm > /dev/null ; 
+
+
+SHOW TABLES LIKE 't6';
+
+show status like 'handler_discover%';
+
+# Check that t3 or t5 can't be created
+# frm files for these tables is stilll on disk
+--error 1050
+create table t3(a int);
+--error 1050
+create table t5(a int);
+
+SHOW TABLES LIKE 't%';
+
+show status like 'handler_discover%';
+
+drop table t1, t2, t4, t6, t7, t8, t9;
+
+
+
+######################################################
+# Test that several tables can be discovered when 
+# one statement access several table at once.
+#
+
+flush status;
+
+# Create tables
+create table t1(id int) engine=ndbcluster;
+create table t2(id int, b char(255)) engine=ndbcluster;
+create table t3(id int, c char(255)) engine=ndbcluster;
+create table t4(id int) engine=myisam;
+
+# Populate tables
+insert into t1 values (1);
+insert into t2 values (2, "table 2");
+insert into t3 values (3, "ndb table 3");
+insert into t4 values (4);
+
+# Remove t1, t2, t3 from disk
+system rm var/master-data/test/t1.frm > /dev/null ; 
+system rm var/master-data/test/t2.frm > /dev/null ; 
+system rm var/master-data/test/t3.frm > /dev/null ; 
+flush tables;
+
+# Select from the table which only exists in NDB.
+select * from t1, t2, t3, t4;
+
+# 3 table should have been discovered
+show status like 'handler_discover%';
+
+drop table t1, t2, t3, t4;
 
 
 #########################################################
@@ -241,7 +428,6 @@ show status like 'handler_discover%';
 drop table t6;
 
 ######################################################
-# Simple test to show use of discover on startup
 # Note! This should always be the last step in this 
 # file, the table t9 will be used and dropped 
 # by ndb_autodiscover2
@@ -259,9 +445,7 @@ system rm var/master-data/test/t9.frm ;
 
 # Now leave test case, when ndb_autodiscover2 will  run, this 
 # MySQL Server will have been restarted because it has a 
-# ndb_autodiscover2-master.opt file. And thus the table should 
-# have been discovered by the "discover on startup" function.
+# ndb_autodiscover2-master.opt file.
+
+
 
-#TODO
-#SLECT * FROM t1, t2, t4;
-#handler discover 3;
diff --git a/mysql-test/t/ndb_autodiscover2.test b/mysql-test/t/ndb_autodiscover2.test
index 6a3f2092148..cce75d5ca4f 100644
--- a/mysql-test/t/ndb_autodiscover2.test
+++ b/mysql-test/t/ndb_autodiscover2.test
@@ -1,14 +1,16 @@
 -- source include/have_ndb.inc
 
 #
-# Simple test to show use of discover on startup
+# Simple test to show use of discover when the server has been restarted
 # The previous step has simply removed the frm file
 # from disk, but left the table in NDB
 #
 --sleep 3;
 select * from t9 order by a;
 
-# handler_discover should be zero
+# handler_discover should be 1
 show status like 'handler_discover%';
 
 drop table t9;
+
+
diff --git a/ndb/include/ndbapi/Ndb.hpp b/ndb/include/ndbapi/Ndb.hpp
index 9d0e9714eef..550d0c0931a 100644
--- a/ndb/include/ndbapi/Ndb.hpp
+++ b/ndb/include/ndbapi/Ndb.hpp
@@ -1087,8 +1087,6 @@ public:
 
   int waitUntilReady(int timeout = 60);
 
-  void connected(Uint32 block_reference);
-  
   /** @} *********************************************************************/
 
   /** 
@@ -1447,6 +1445,9 @@ private:
   void setup(Ndb_cluster_connection *ndb_cluster_connection,
 	     const char* aCatalogName, const char* aSchemaName);
 
+  void connected(Uint32 block_reference);
+ 
+
   NdbConnection*  startTransactionLocal(Uint32 aPrio, Uint32 aFragmentId); 
 
 // Connect the connection object to the Database.
diff --git a/ndb/src/mgmsrv/ConfigInfo.cpp b/ndb/src/mgmsrv/ConfigInfo.cpp
index 0b7c664dfd3..e8f768d1968 100644
--- a/ndb/src/mgmsrv/ConfigInfo.cpp
+++ b/ndb/src/mgmsrv/ConfigInfo.cpp
@@ -3479,8 +3479,9 @@ check_node_vs_replicas(Vector<ConfigInfo::ConfigRuleSection>&sections,
 		       struct InitConfigFileParser::Context &ctx, 
 		       const char * rule_data)
 {
-  Uint32 db_nodes = 0;
-  Uint32 replicas = 0;
+  Uint32 db_nodes= 0;
+  Uint32 replicas= 0;
+  Uint32 db_host_count= 0;
   ctx.m_userProperties.get(DB_TOKEN, &db_nodes);
   ctx.m_userProperties.get("NoOfReplicas", &replicas);
   if((db_nodes % replicas) != 0){
@@ -3488,7 +3489,108 @@ check_node_vs_replicas(Vector<ConfigInfo::ConfigRuleSection>&sections,
 		    "No of nodes must be dividable with no or replicas");
     return false;
   }
-  
+  // check that node groups and arbitrators are ok
+  // just issue warning if not
+  if(replicas > 1){
+    Properties * props= ctx.m_config;
+    Properties p_db_hosts(true); // store hosts which db nodes run on
+    Properties p_arbitrators(true); // store hosts which arbitrators run on
+    // arbitrator should not run together with db node on same host
+    Uint32 i, n, group= 0, i_group= 0;
+    Uint32 n_nodes;
+    BaseString node_group_warning, arbitration_warning;
+    const char *arbit_warn_fmt=
+      "\n  arbitrator with id %d and db node with id %d on same host %s";
+
+    ctx.m_userProperties.get("NoOfNodes", &n_nodes);
+    for (i= 0, n= 0; n < n_nodes; i++){
+      const Properties * tmp;
+      if(!props->get("Node", i, &tmp)) continue;
+      n++;
+
+      const char * type;
+      if(!tmp->get("Type", &type)) continue;
+
+      const char* host= 0;
+      tmp->get("HostName", &host);
+
+      if (strcmp(type,DB_TOKEN) == 0)
+      {
+	{
+	  Uint32 ii;
+	  if (!p_db_hosts.get(host,&ii))
+	    db_host_count++;
+	  p_db_hosts.put(host,i);
+	  if (p_arbitrators.get(host,&ii))
+	  {
+	    arbitration_warning.appfmt(arbit_warn_fmt, ii, i, host);
+	    p_arbitrators.remove(host); // only one warning per db node
+	  }
+	}
+	{
+	  unsigned j;
+	  BaseString str, str2;
+	  str.assfmt("#group%d_",group);
+	  p_db_hosts.put(str.c_str(),i_group,host);
+	  str2.assfmt("##group%d_",group);
+	  p_db_hosts.put(str2.c_str(),i_group,i);
+	  for (j= 0; j < i_group; j++)
+	  {
+	    const char *other_host;
+	    p_db_hosts.get(str.c_str(),j,&other_host);
+	    if (strcmp(host,other_host) == 0) {
+	      unsigned int other_i, c= 0;
+	      p_db_hosts.get(str2.c_str(),j,&other_i);
+	      p_db_hosts.get(str.c_str(),&c);
+	      if (c == 0) // first warning in this node group
+		node_group_warning.appfmt("  Node group %d", group);
+	      c|= 1 << j;
+	      p_db_hosts.put(str.c_str(),c);
+
+	      node_group_warning.appfmt(",\n    db node with id %d and id %d "
+					"on same host %s", other_i, i, host);
+	    }
+	  }
+	  i_group++;
+	  DBUG_ASSERT(i_group <= replicas);
+	  if (i_group == replicas)
+	  {
+	    unsigned c= 0;
+	    p_db_hosts.get(str.c_str(),&c);
+	    if (c+1 == (1 << (replicas-1))) // all nodes on same machine
+	      node_group_warning.append(".\n    Host failure will "
+					"cause complete cluster shutdown.");
+	    else if (c > 0)
+	      node_group_warning.append(".\n    Host failure may "
+					"cause complete cluster shutdown.");
+	    group++;
+	    i_group= 0;
+	  }
+	}
+      }
+      else if (strcmp(type,API_TOKEN) == 0 ||
+	       strcmp(type,MGM_TOKEN) == 0)
+      {
+	Uint32 rank;
+	if(tmp->get("ArbitrationRank", &rank) && rank > 0)
+	{
+	  if(host && host[0] != 0)
+	  {
+	    Uint32 ii;
+	    p_arbitrators.put(host,i);
+	    if (p_db_hosts.get(host,&ii))
+	    {
+	      arbitration_warning.appfmt(arbit_warn_fmt, i, ii, host);
+	    }
+	  }
+	}
+      }
+    }
+    if (db_host_count > 1 && node_group_warning.length() > 0)
+      ndbout_c("Cluster configuration warning:\n%s",node_group_warning.c_str());
+    if (db_host_count > 1 && arbitration_warning.length() > 0)
+      ndbout_c("Cluster configuration warning:%s",arbitration_warning.c_str());
+  }
   return true;
 }
 
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index 9abe52fb030..b9b932663f6 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -868,6 +868,8 @@ NdbDictInterface::dictSignal(NdbApiSignal* signal,
 			     const int noerrcodes,
 			     const int temporaryMask)
 {
+  DBUG_ENTER("NdbDictInterface::dictSignal");
+  DBUG_PRINT("enter", ("useMasterNodeId: %d", useMasterNodeId));
   for(Uint32 i = 0; i<RETRIES; i++){
     //if (useMasterNodeId == 0)
     m_buffer.clear();
@@ -887,7 +889,7 @@ NdbDictInterface::dictSignal(NdbApiSignal* signal,
     if(aNodeId == 0){
       m_error.code = 4009;
       m_transporter->unlock_mutex();
-      return -1;
+      DBUG_RETURN(-1);
     }
     {
       int r;
@@ -923,7 +925,7 @@ NdbDictInterface::dictSignal(NdbApiSignal* signal,
     
     if(m_waiter.m_state == NO_WAIT && m_error.code == 0){
       // Normal return
-      return 0;
+      DBUG_RETURN(0);
     }
     
     /**
@@ -946,9 +948,9 @@ NdbDictInterface::dictSignal(NdbApiSignal* signal,
 	continue;
     }
 
-    return -1;
+    DBUG_RETURN(-1);
   }
-  return -1; 
+  DBUG_RETURN(-1);
 }
 
 /*****************************************************************
diff --git a/ndb/src/ndbapi/TransporterFacade.cpp b/ndb/src/ndbapi/TransporterFacade.cpp
index f0b0bfe2de4..bc24110ea14 100644
--- a/ndb/src/ndbapi/TransporterFacade.cpp
+++ b/ndb/src/ndbapi/TransporterFacade.cpp
@@ -942,6 +942,8 @@ TransporterFacade::isConnected(NodeId aNodeId){
 NodeId
 TransporterFacade::get_an_alive_node()
 {
+  DBUG_ENTER("TransporterFacade::get_an_alive_node");
+  DBUG_PRINT("enter", ("theStartNodeId: %d", theStartNodeId));
 #ifdef VM_TRACE
   const char* p = NdbEnv_GetEnv("NDB_ALIVE_NODE_ID", (char*)0, 0);
   if (p != 0 && *p != 0)
@@ -950,17 +952,19 @@ TransporterFacade::get_an_alive_node()
   NodeId i;
   for (i = theStartNodeId; i < MAX_NDB_NODES; i++) {
     if (get_node_alive(i)){
+      DBUG_PRINT("info", ("Node %d is alive", i));
       theStartNodeId = ((i + 1) % MAX_NDB_NODES);
-      return i;
+      DBUG_RETURN(i);
     }
   }
   for (i = 1; i < theStartNodeId; i++) {
     if (get_node_alive(i)){
+      DBUG_PRINT("info", ("Node %d is alive", i));
       theStartNodeId = ((i + 1) % MAX_NDB_NODES);
-      return i;
+      DBUG_RETURN(i);
     }
   }
-  return (NodeId)0;
+  DBUG_RETURN((NodeId)0);
 }
 
 TransporterFacade::ThreadData::ThreadData(Uint32 size){
diff --git a/ndb/src/ndbapi/ndb_cluster_connection.cpp b/ndb/src/ndbapi/ndb_cluster_connection.cpp
index 5be4f0f9f91..358dfa74785 100644
--- a/ndb/src/ndbapi/ndb_cluster_connection.cpp
+++ b/ndb/src/ndbapi/ndb_cluster_connection.cpp
@@ -67,6 +67,9 @@ void Ndb_cluster_connection::connect_thread()
       printf("Ndb_cluster_connection::connect_thread error\n");
       DBUG_ASSERT(false);
       g_run_connect_thread= 0;
+    } else {
+      // Wait before making a new connect attempt
+      NdbSleep_SecSleep(1);
     }
   } while (g_run_connect_thread);
   if (m_connect_callback)
diff --git a/sql/discover.cc b/sql/discover.cc
index 696be193148..1251055c70e 100644
--- a/sql/discover.cc
+++ b/sql/discover.cc
@@ -125,47 +125,4 @@ int writefrm(const char *name, const void *frmdata, uint len)
 
 
 
-/*
-  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);
-  }
-    
-  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
index e48ac8cc75e..d7a580dbb88 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -32,9 +32,6 @@
 #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;
 
@@ -48,11 +45,13 @@ static const ha_rows autoincrement_prefetch= 32;
 // connectstring to cluster if given by mysqld
 const char *ndbcluster_connectstring= 0;
 
+static const char *ha_ndb_ext=".ndb";
+
 #define NDB_HIDDEN_PRIMARY_KEY_LENGTH 8
 
 
 #define ERR_PRINT(err) \
-  DBUG_PRINT("error", ("Error: %d  message: %s", err.code, err.message))
+  DBUG_PRINT("error", ("%d  message: %s", err.code, err.message))
 
 #define ERR_RETURN(err)		         \
 {				         \
@@ -107,7 +106,9 @@ static const err_code_mapping err_map[]=
   { 893, HA_ERR_FOUND_DUPP_UNIQUE },
   { 721, HA_ERR_TABLE_EXIST },
   { 4244, HA_ERR_TABLE_EXIST },
-  { 241, HA_ERR_OLD_METADATA },
+
+  { 709, HA_ERR_NO_SUCH_TABLE },
+  { 284, HA_ERR_NO_SUCH_TABLE },
 
   { 266, HA_ERR_LOCK_WAIT_TIMEOUT },
   { 274, HA_ERR_LOCK_WAIT_TIMEOUT },
@@ -147,7 +148,25 @@ int execute_no_commit(ha_ndbcluster *h, NdbConnection *trans)
   int m_batch_execute= 0;
   if (false && m_batch_execute)
     return 0;
-  return trans->execute(NoCommit);
+  return trans->execute(NoCommit,AbortOnError,1);
+}
+
+inline
+int execute_commit(ha_ndbcluster *h, NdbConnection *trans)
+{
+  int m_batch_execute= 0;
+  if (false && m_batch_execute)
+    return 0;
+  return trans->execute(Commit,AbortOnError,1);
+}
+
+inline
+int execute_no_commit_ie(ha_ndbcluster *h, NdbConnection *trans)
+{
+  int m_batch_execute= 0;
+  if (false && m_batch_execute)
+    return 0;
+  return trans->execute(NoCommit,IgnoreError,1);
 }
 
 /*
@@ -178,19 +197,9 @@ struct Ndb_table_local_info {
   ha_rows records;
 };
 
-void ha_ndbcluster::set_rec_per_key()
-{
-  DBUG_ENTER("ha_ndbcluster::get_status_const");
-  for (uint i=0 ; i < table->keys ; i++)
-  {
-    table->key_info[i].rec_per_key[table->key_info[i].key_parts-1]= 1;
-  }
-  DBUG_VOID_RETURN;
-}
-
 void ha_ndbcluster::records_update()
 {
-  DBUG_ENTER("ha_ndbcluster::get_status_variable");
+  DBUG_ENTER("ha_ndbcluster::records_update");
   struct Ndb_table_local_info *info= (struct Ndb_table_local_info *)m_table_info;
   DBUG_PRINT("info", ("id=%d, no_uncommitted_rows_count=%d",
 		      ((const NDBTAB *)m_table)->getTableId(),
@@ -278,6 +287,7 @@ int ha_ndbcluster::ndb_err(NdbConnection *trans)
     NDBDICT *dict= m_ndb->getDictionary();
     DBUG_PRINT("info", ("invalidateTable %s", m_tabname));
     dict->invalidateTable(m_tabname);
+    table->version=0L;			/* Free when thread is ready */
     break;
   }
   default:
@@ -318,7 +328,8 @@ bool ha_ndbcluster::get_error_message(int error,
 
 /*
   Check if type is supported by NDB.
-  TODO Use this once, not in every operation
+  TODO Use this once in open(), not in every operation
+
 */
 
 static inline bool ndb_supported_type(enum_field_types type)
@@ -665,7 +676,7 @@ int ha_ndbcluster::get_metadata(const char *path)
 		    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;
+	error= 3;
 	invalidating_ndb_table= false;
       }
     }
@@ -689,11 +700,11 @@ int ha_ndbcluster::get_metadata(const char *path)
 
 int ha_ndbcluster::build_index_list(TABLE *tab, enum ILBP phase)
 {
+  uint i;
   int error= 0;
-  char *name;
-  const char *index_name;
+  const char *name, *index_name;
+  char unique_index_name[FN_LEN];
   static const char* unique_suffix= "$unique";
-  uint i, name_len;
   KEY* key_info= tab->key_info;
   const char **key_name= tab->keynames.type_names;
   NdbDictionary::Dictionary *dict= m_ndb->getDictionary();
@@ -707,21 +718,15 @@ int ha_ndbcluster::build_index_list(TABLE *tab, enum ILBP phase)
     m_index[i].type= idx_type;
     if (idx_type == UNIQUE_ORDERED_INDEX || idx_type == UNIQUE_INDEX)
     {
-      name_len= strlen(index_name)+strlen(unique_suffix)+1;
-      // Create name for unique index by appending "$unique";     
-      if (!(name= my_malloc(name_len, MYF(MY_WME))))
-	DBUG_RETURN(2);
-      strxnmov(name, name_len, index_name, unique_suffix, NullS);
-      m_index[i].unique_name= name;
-      DBUG_PRINT("info", ("Created unique index name: %s for index %d",
-			  name, i));
+      strxnmov(unique_index_name, FN_LEN, index_name, unique_suffix, NullS);
+      DBUG_PRINT("info", ("Created unique index name \'%s\' for index %d",
+			  unique_index_name, i));
     }
     // Create secondary indexes if in create phase
     if (phase == ILBP_CREATE)
     {
-      DBUG_PRINT("info", ("Creating index %u: %s", i, index_name));
-      
-      switch (m_index[i].type){
+      DBUG_PRINT("info", ("Creating index %u: %s", i, index_name));      
+      switch (idx_type){
 	
       case PRIMARY_KEY_INDEX:
 	// Do nothing, already created
@@ -731,10 +736,10 @@ int ha_ndbcluster::build_index_list(TABLE *tab, enum ILBP phase)
 	break;
       case UNIQUE_ORDERED_INDEX:
 	if (!(error= create_ordered_index(index_name, key_info)))
-	  error= create_unique_index(get_unique_index_name(i), key_info);
+	  error= create_unique_index(unique_index_name, key_info);
 	break;
       case UNIQUE_INDEX:
-	error= create_unique_index(get_unique_index_name(i), key_info);
+	error= create_unique_index(unique_index_name, key_info);
 	break;
       case ORDERED_INDEX:
 	error= create_ordered_index(index_name, key_info);
@@ -751,21 +756,20 @@ int ha_ndbcluster::build_index_list(TABLE *tab, enum ILBP phase)
       }
     }
     // Add handles to index objects
-    DBUG_PRINT("info", ("Trying to add handle to index %s", index_name));
-    if ((m_index[i].type != PRIMARY_KEY_INDEX) &&
-	(m_index[i].type != UNIQUE_INDEX))
+    if (idx_type != PRIMARY_KEY_INDEX && idx_type != UNIQUE_INDEX)
     {
+      DBUG_PRINT("info", ("Get handle to index %s", index_name));
       const NDBINDEX *index= dict->getIndex(index_name, m_tabname);
       if (!index) DBUG_RETURN(1);
       m_index[i].index= (void *) index;
     }
-    if (m_index[i].unique_name)
+    if (idx_type == UNIQUE_ORDERED_INDEX || idx_type == UNIQUE_INDEX)
     {
-      const NDBINDEX *index= dict->getIndex(m_index[i].unique_name, m_tabname);
+      DBUG_PRINT("info", ("Get handle to unique_index %s", unique_index_name));
+      const NDBINDEX *index= dict->getIndex(unique_index_name, m_tabname);
       if (!index) DBUG_RETURN(1);
       m_index[i].unique_index= (void *) index;
     }      
-    DBUG_PRINT("info", ("Added handle to index %s", index_name));
   }
   
   DBUG_RETURN(error);
@@ -801,9 +805,6 @@ void ha_ndbcluster::release_metadata()
   // Release index list 
   for (i= 0; i < MAX_KEY; i++)
   {
-    if (m_index[i].unique_name)
-      my_free((char*)m_index[i].unique_name, MYF(0));
-    m_index[i].unique_name= NULL;
     m_index[i].unique_index= NULL;      
     m_index[i].index= NULL;      
   }
@@ -813,17 +814,15 @@ void ha_ndbcluster::release_metadata()
 
 int ha_ndbcluster::get_ndb_lock_type(enum thr_lock_type type)
 {
-  int lm;
   if (type == TL_WRITE_ALLOW_WRITE)
-    lm= NdbOperation::LM_Exclusive;
+    return NdbOperation::LM_Exclusive;
   else if (uses_blob_value(retrieve_all_fields))
     /*
       TODO use a new scan mode to read + lock + keyinfo
     */
-    lm= NdbOperation::LM_Exclusive;
+    return NdbOperation::LM_Exclusive;
   else
-    lm= NdbOperation::LM_CommittedRead;
-  return lm;
+    return NdbOperation::LM_CommittedRead;
 }
 
 static const ulong index_type_flags[]=
@@ -861,16 +860,6 @@ static const ulong index_type_flags[]=
 
 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 const char* ha_ndbcluster::get_unique_index_name(uint idx_no) const
-{
-  return m_index[idx_no].unique_name;
-}
-
 inline NDB_INDEX_TYPE ha_ndbcluster::get_index_type(uint idx_no) const
 {
   DBUG_ASSERT(idx_no < MAX_KEY);
@@ -1006,7 +995,7 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf)
     }
   }
   
-  if (trans->execute(NoCommit, IgnoreError) != 0) 
+  if (execute_no_commit_ie(this,trans) != 0) 
   {
     table->status= STATUS_NOT_FOUND;
     DBUG_RETURN(ndb_err(trans));
@@ -1087,7 +1076,6 @@ int ha_ndbcluster::unique_index_read(const byte *key,
   DBUG_ENTER("unique_index_read");
   DBUG_PRINT("enter", ("key_len: %u, index: %u", key_len, active_index));
   DBUG_DUMP("key", (char*)key, key_len);
-  DBUG_PRINT("enter", ("name: %s", get_unique_index_name(active_index)));
   
   NdbOperation::LockMode lm=
     (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
@@ -1127,7 +1115,7 @@ int ha_ndbcluster::unique_index_read(const byte *key,
     }
   }
 
-  if (trans->execute(NoCommit, IgnoreError) != 0) 
+  if (execute_no_commit_ie(this,trans) != 0) 
   {
     table->status= STATUS_NOT_FOUND;
     DBUG_RETURN(ndb_err(trans));
@@ -1204,7 +1192,7 @@ inline int ha_ndbcluster::next_result(byte *buf)
       }
       else
       {
-	if (ops_pending && (trans->execute(Commit) != 0))
+	if (ops_pending && (execute_commit(this,trans) != 0))
 	  DBUG_RETURN(ndb_err(trans));
 	trans->restart();
       }
@@ -1343,7 +1331,6 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
   NdbConnection *trans= m_active_trans;
   NdbResultSet *cursor;
   NdbIndexScanOperation *op;
-  const char *index_name;
 
   DBUG_ENTER("ordered_index_scan");
   DBUG_PRINT("enter", ("index: %u, sorted: %d", active_index, sorted));  
@@ -1352,7 +1339,6 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
   DBUG_EXECUTE("enter", print_key(start_key, "start_key"););
   DBUG_EXECUTE("enter", print_key(end_key, "end_key"););
   
-  index_name= get_index_name(active_index);
 
   NdbOperation::LockMode lm=
     (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
@@ -1561,8 +1547,8 @@ int ha_ndbcluster::write_row(byte *record)
   }
   
   statistic_increment(ha_write_count,&LOCK_status);
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
-    table->timestamp_field->set_time();
+  if (table->timestamp_default_now)
+    update_timestamp(record+table->timestamp_default_now-1);
   has_auto_increment= (table->next_number_field && record == table->record[0]);
 
   if (!(op= trans->getNdbOperation((const NDBTAB *) m_table)))
@@ -1639,7 +1625,7 @@ int ha_ndbcluster::write_row(byte *record)
     }
     else
     {
-      if (trans->execute(Commit) != 0)
+      if (execute_commit(this,trans) != 0)
       {
 	skip_auto_increment= true;
 	no_uncommitted_rows_execute_failure();
@@ -1712,9 +1698,9 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
   DBUG_ENTER("update_row");
   
   statistic_increment(ha_update_count,&LOCK_status);
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
-    table->timestamp_field->set_time();
-
+  if (table->timestamp_on_update_now)
+    update_timestamp(new_data+table->timestamp_on_update_now-1);
+  
   /* Check for update of primary key for special handling */  
   if ((table->primary_key != MAX_KEY) &&
       (key_cmp(table->primary_key, old_data, new_data)))
@@ -2419,6 +2405,8 @@ void ha_ndbcluster::info(uint flag)
     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"));
@@ -2434,11 +2422,6 @@ void ha_ndbcluster::info(uint flag)
       }
     }
   }
-  if (flag & HA_STATUS_CONST)
-  {
-    DBUG_PRINT("info", ("HA_STATUS_CONST"));
-    set_rec_per_key();
-  }
   if (flag & HA_STATUS_ERRKEY)
   {
     DBUG_PRINT("info", ("HA_STATUS_ERRKEY"));
@@ -2660,7 +2643,7 @@ int ha_ndbcluster::reset()
 
 
 const char **ha_ndbcluster::bas_ext() const
-{ static const char *ext[1]= { NullS }; return ext; }
+{ static const char *ext[]= { ".ndb", NullS }; return ext; }
 
 
 /*
@@ -2684,14 +2667,13 @@ THR_LOCK_DATA **ha_ndbcluster::store_lock(THD *thd,
                                           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 &&
+    if ((lock_type >= TL_WRITE_ALLOW_WRITE &&
          lock_type <= TL_WRITE) && !thd->in_lock_tables)      
       lock_type= TL_WRITE_ALLOW_WRITE;
     
@@ -2858,18 +2840,26 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     }
     m_table= NULL;
     m_table_info= NULL;
-    if (m_active_trans)
-      DBUG_PRINT("warning", ("m_active_trans != NULL"));
+    /*
+      This is the place to make sure this handler instance
+      no longer are connected to the active transaction.
+
+      And since the handler is no longer part of the transaction 
+      it can't have open cursors, ops or blobs pending.
+    */
+    m_active_trans= NULL;    
+
     if (m_active_cursor)
       DBUG_PRINT("warning", ("m_active_cursor != NULL"));
+    m_active_cursor= NULL;
+
     if (blobs_pending)
       DBUG_PRINT("warning", ("blobs_pending != 0"));
+    blobs_pending= 0;
+    
     if (ops_pending)
       DBUG_PRINT("warning", ("ops_pending != 0L"));
-    m_active_trans= NULL;
-    m_active_cursor= NULL;
     ops_pending= 0;
-    blobs_pending= 0;
   }
   DBUG_RETURN(error);
 }
@@ -2929,7 +2919,7 @@ int ndbcluster_commit(THD *thd, void *ndb_transaction)
                             "stmt" : "all"));
   DBUG_ASSERT(ndb && trans);
 
-  if (trans->execute(Commit) != 0)
+  if (execute_commit(0,trans) != 0)
   {
     const NdbError err= trans->getNdbError();
     const NdbOperation *error_op= trans->getNdbErrorOperation();
@@ -3180,12 +3170,24 @@ int ha_ndbcluster::create(const char *name,
   const void *data, *pack_data;
   const char **key_names= form->keynames.type_names;
   char name2[FN_HEADLEN];
+  bool create_from_engine= (info->table_options & HA_CREATE_FROM_ENGINE);
    
   DBUG_ENTER("create");
   DBUG_PRINT("enter", ("name: %s", name));
   fn_format(name2, name, "", "",2);       // Remove the .frm extension
   set_dbname(name2);
-  set_tabname(name2);  
+  set_tabname(name2);    
+
+  if (create_from_engine)
+  {
+    /*
+      Table alreay exists in NDB and frm file has been created by 
+      caller.
+      Do Ndb specific stuff, such as create a .ndb file
+    */
+    my_errno= write_ndb_file();
+    DBUG_RETURN(my_errno);
+  }
 
   DBUG_PRINT("table", ("name: %s", m_tabname));  
   tab.setName(m_tabname);
@@ -3226,16 +3228,12 @@ int ha_ndbcluster::create(const char *name,
     tab.addColumn(col);
   }
   
-  my_errno= 0;
-  if (check_ndb_connection())
-  {
-    my_errno= HA_ERR_NO_CONNECTION;
+  if ((my_errno= check_ndb_connection()))
     DBUG_RETURN(my_errno);
-  }
   
   // Create the table in NDB     
   NDBDICT *dict= m_ndb->getDictionary();
-  if (dict->createTable(tab)) 
+  if (dict->createTable(tab) != 0) 
   {
     const NdbError err= dict->getNdbError();
     ERR_PRINT(err);
@@ -3248,6 +3246,9 @@ int ha_ndbcluster::create(const char *name,
   // Create secondary indexes
   my_errno= build_index_list(form, ILBP_CREATE);
 
+  if (!my_errno)
+    my_errno= write_ndb_file();
+
   DBUG_RETURN(my_errno);
 }
 
@@ -3283,7 +3284,6 @@ int ha_ndbcluster::create_index(const char *name,
   DBUG_ENTER("create_index");
   DBUG_PRINT("enter", ("name: %s ", name));
 
-  //  NdbDictionary::Index ndb_index(name);
   NdbDictionary::Index ndb_index(name);
   if (unique)
     ndb_index.setType(NdbDictionary::Index::UniqueHashIndex);
@@ -3324,14 +3324,16 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
   set_tabname(from);
   set_tabname(to, new_tabname);
 
-  if (check_ndb_connection()) {
-    my_errno= HA_ERR_NO_CONNECTION;
-    DBUG_RETURN(my_errno);
-  }
+  if (check_ndb_connection())
+    DBUG_RETURN(my_errno= HA_ERR_NO_CONNECTION);
+
 
   int result= alter_table_name(m_tabname, new_tabname);
   if (result == 0)
+  {
     set_tabname(to);
+    handler::rename_table(from, to);
+  }
   
   DBUG_RETURN(result);
 }
@@ -3376,6 +3378,8 @@ int ha_ndbcluster::delete_table(const char *name)
   
   if (check_ndb_connection())
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
+
+  handler::delete_table(name);
   DBUG_RETURN(drop_table());
 }
 
@@ -3477,7 +3481,6 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
   for (i= 0; i < MAX_KEY; i++)
   {
     m_index[i].type= UNDEFINED_INDEX;   
-    m_index[i].unique_name= NULL;      
     m_index[i].unique_index= NULL;      
     m_index[i].index= NULL;      
   }
@@ -3520,7 +3523,6 @@ ha_ndbcluster::~ha_ndbcluster()
 
 int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
 {
-  int res;
   KEY *key;
   DBUG_ENTER("open");
   DBUG_PRINT("enter", ("name: %s mode: %d test_if_locked: %d",
@@ -3547,11 +3549,8 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
     free_share(m_share); m_share= 0;
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   }
-  res= get_metadata(name);
-  if (!res)
-    info(HA_STATUS_VARIABLE | HA_STATUS_CONST);
-
-  DBUG_RETURN(res);
+  
+  DBUG_RETURN(get_metadata(name));
 }
 
 
@@ -3575,13 +3574,7 @@ Thd_ndb* ha_ndbcluster::seize_thd_ndb()
   Thd_ndb *thd_ndb;
   DBUG_ENTER("seize_thd_ndb");
 
-#ifdef USE_NDB_POOL
-  // Seize from pool
-  ndb= Ndb::seize();
-  xxxxxxxxxxxxxx error
-#else
   thd_ndb= new Thd_ndb();
-#endif
   thd_ndb->ndb->getDictionary()->set_local_table_data_size(sizeof(Ndb_table_local_info));
   if (thd_ndb->ndb->init(max_transactions) != 0)
   {
@@ -3602,46 +3595,45 @@ Thd_ndb* ha_ndbcluster::seize_thd_ndb()
 void ha_ndbcluster::release_thd_ndb(Thd_ndb* thd_ndb)
 {
   DBUG_ENTER("release_thd_ndb");
-#ifdef USE_NDB_POOL
-  // Release to  pool
-  Ndb::release(ndb);
-  xxxxxxxxxxxx error
-#else
   delete thd_ndb;
-#endif
   DBUG_VOID_RETURN;
 }
 
 
 /*
-  If this thread already has a Ndb object allocated
+  If this thread already has a Thd_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.
+  seize a Thd_ndb object, assign it to current THD and use it.
  
 */
 
-int ha_ndbcluster::check_ndb_connection()
+Ndb* check_ndb_in_thd(THD* thd)
 {
-  THD *thd= current_thd;
+  DBUG_ENTER("check_ndb_in_thd");
   Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;
-  DBUG_ENTER("check_ndb_connection");
   
   if (!thd_ndb)
   {
-    thd_ndb= seize_thd_ndb();
-    if (!thd_ndb)
-      DBUG_RETURN(2);
+    if (!(thd_ndb= ha_ndbcluster::seize_thd_ndb()))
+      DBUG_RETURN(NULL);
     thd->transaction.thd_ndb= thd_ndb;
   }
-  m_ndb= thd_ndb->ndb;
+  DBUG_RETURN(thd_ndb->ndb);
+}
+
+
+int ha_ndbcluster::check_ndb_connection()
+{
+  THD* thd= current_thd;
+  DBUG_ENTER("check_ndb_connection");
+  
+  if (!(m_ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
   m_ndb->setDatabaseName(m_dbname);
   DBUG_RETURN(0);
 }
 
+
 void ndbcluster_close_connection(THD *thd)
 {
   Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;
@@ -3659,28 +3651,29 @@ void ndbcluster_close_connection(THD *thd)
   Try to discover one table from NDB
  */
 
-int ndbcluster_discover(const char *dbname, const char *name,
+int ndbcluster_discover(THD* thd, const char *db, const char *name,
 			const void** frmblob, uint* frmlen)
 {
   uint len;
   const void* data;
   const NDBTAB* tab;
+  Ndb* ndb;
   DBUG_ENTER("ndbcluster_discover");
-  DBUG_PRINT("enter", ("db: %s, name: %s", dbname, name)); 
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name)); 
 
-  Ndb ndb(g_ndb_cluster_connection, dbname);
-  ndb.getDictionary()->set_local_table_data_size(sizeof(Ndb_table_local_info));
+  if (!(ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);  
+  ndb->setDatabaseName(db);
 
-  if (ndb.init())
-    ERR_RETURN(ndb.getNdbError());
-
-  if (ndb.waitUntilReady(0))
-    ERR_RETURN(ndb.getNdbError());
-  
-  if (!(tab= ndb.getDictionary()->getTable(name)))
-  {
-    DBUG_PRINT("info", ("Table %s not found", name));
-    DBUG_RETURN(1);
+  NDBDICT* dict= ndb->getDictionary();
+  dict->set_local_table_data_size(sizeof(Ndb_table_local_info));
+  dict->invalidateTable(name);
+  if (!(tab= dict->getTable(name)))
+  {    
+    const NdbError err= dict->getNdbError();
+    if (err.code == 709)
+      DBUG_RETURN(1);
+    ERR_RETURN(err);
   }
   
   DBUG_PRINT("info", ("Found table %s", tab->getName()));
@@ -3702,41 +3695,196 @@ int ndbcluster_discover(const char *dbname, const char *name,
   DBUG_RETURN(0);
 }
 
-
-#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
-*/
+  Check if a table exists in NDB
+   
+ */
 
-int ndb_discover_tables()
+int ndbcluster_table_exists(THD* thd, const char *db, const char *name)
+{
+  uint len;
+  const void* data;
+  const NDBTAB* tab;
+  Ndb* ndb;
+  DBUG_ENTER("ndbcluster_table_exists");
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name)); 
+
+  if (!(ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);  
+  ndb->setDatabaseName(db);
+
+  NDBDICT* dict= ndb->getDictionary();
+  dict->set_local_table_data_size(sizeof(Ndb_table_local_info));
+  dict->invalidateTable(name);
+  if (!(tab= dict->getTable(name)))
+  {    
+    const NdbError err= dict->getNdbError();
+    if (err.code == 709)
+      DBUG_RETURN(0);
+    ERR_RETURN(err);
+  }
+  
+  DBUG_PRINT("info", ("Found table %s", tab->getName()));
+  DBUG_RETURN(1);
+}
+
+
+
+extern "C" byte* tables_get_key(const char *entry, uint *length,
+				my_bool not_used __attribute__((unused)))
+{
+  *length= strlen(entry);
+  return (byte*) entry;
+}
+
+
+int ndbcluster_find_files(THD *thd,const char *db,const char *path,
+			  const char *wild, bool dir, List<char> *files)
 {
   uint i;
+  Ndb* ndb;
+  char name[FN_REFLEN];
+  HASH ndb_tables, ok_tables;
   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();
+  DBUG_ENTER("ndbcluster_find_files");
+  DBUG_PRINT("enter", ("db: %s", db));
+
+  if (!(ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
+
+  if (dir)
+    DBUG_RETURN(0); // Discover of databases not yet supported
+
+  // List tables in NDB
+  NDBDICT *dict= ndb->getDictionary();
   if (dict->listObjects(list, 
 			NdbDictionary::Object::UserTable) != 0)
     ERR_RETURN(dict->getNdbError());
-  
+
+  if (hash_init(&ndb_tables, system_charset_info,list.count,0,0,
+		(hash_get_key)tables_get_key,0,0))
+  {
+    DBUG_PRINT("error", ("Failed to init HASH ndb_tables"));
+    DBUG_RETURN(-1);
+  }
+
+  if (hash_init(&ok_tables, system_charset_info,32,0,0,
+		(hash_get_key)tables_get_key,0,0))
+  {
+    DBUG_PRINT("error", ("Failed to init HASH ok_tables"));
+    hash_free(&ndb_tables);
+    DBUG_RETURN(-1);
+  }  
+
   for (i= 0 ; i < list.count ; i++)
   {
     NdbDictionary::Dictionary::List::Element& t= list.elements[i];
+    DBUG_PRINT("info", ("Found %s/%s in NDB", t.database, t.name));     
 
-    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));
+    // Apply wildcard to list of tables in NDB
+    if (wild)
+    {
+      if (lower_case_table_names)
+      {
+	if (wild_case_compare(files_charset_info, t.name, wild))
+	  continue;
+      }
+      else if (wild_compare(t.name,wild,0))
+	continue;
+    }
+    DBUG_PRINT("info", ("Inserting %s into ndb_tables hash", t.name));     
+    my_hash_insert(&ndb_tables, (byte*)thd->strdup(t.name));
   }
-  DBUG_RETURN(0);  
+
+  char *file_name;
+  List_iterator<char> it(*files);
+  List<char> delete_list;
+  while ((file_name=it++))
+  {
+    DBUG_PRINT("info", ("%s", file_name));     
+    if (hash_search(&ndb_tables, file_name, strlen(file_name)))
+    {
+      DBUG_PRINT("info", ("%s existed in NDB _and_ on disk ", file_name));
+      // File existed in NDB and as frm file, put in ok_tables list
+      my_hash_insert(&ok_tables, (byte*)file_name);
+      continue;
+    }
+    
+    // File is not in NDB, check for .ndb file with this name
+    (void)strxnmov(name, FN_REFLEN, 
+		   mysql_data_home,"/",db,"/",file_name,ha_ndb_ext,NullS);
+    DBUG_PRINT("info", ("Check access for %s", name));
+    if (access(name, F_OK))
+    {
+      DBUG_PRINT("info", ("%s did not exist on disk", name));     
+      // .ndb file did not exist on disk, another table type
+      continue;
+    }
+
+    DBUG_PRINT("info", ("%s existed on disk", name));     
+    // The .ndb file exists on disk, but it's not in list of tables in ndb
+    // Verify that handler agrees table is gone.
+    if (ndbcluster_table_exists(thd, db, file_name) == 0)    
+    {
+      DBUG_PRINT("info", ("NDB says %s does not exists", file_name));     
+      it.remove();
+      // Put in list of tables to remove from disk
+      delete_list.push_back(thd->strdup(file_name));
+    }
+  }
+
+  // Check for new files to discover
+  DBUG_PRINT("info", ("Checking for new files to discover"));       
+  List<char> create_list;
+  for (i= 0 ; i < ndb_tables.records ; i++)
+  {
+    file_name= hash_element(&ndb_tables, i);
+    if (!hash_search(&ok_tables, file_name, strlen(file_name)))
+    {
+      DBUG_PRINT("info", ("%s must be discovered", file_name));       
+      // File is in list of ndb tables and not in ok_tables
+      // This table need to be created
+      create_list.push_back(thd->strdup(file_name));
+    }
+  }
+
+  // Lock mutex before deleting and creating frm files
+  pthread_mutex_lock(&LOCK_open);
+
+  if (!global_read_lock)
+  {
+    // Delete old files
+    List_iterator_fast<char> it3(delete_list);
+    while ((file_name=it3++))
+    {  
+      DBUG_PRINT("info", ("Remove table %s/%s",db, file_name ));
+      // Delete the table and all related files
+      TABLE_LIST table_list;
+      bzero((char*) &table_list,sizeof(table_list));
+      table_list.db= (char*) db;
+      table_list.real_name=(char*)file_name;
+      (void)mysql_rm_table_part2(thd, &table_list, 
+				 /* if_exists */ true, 
+				 /* drop_temporary */ false, 
+				 /* dont_log_query*/ true);
+    }
+  }
+
+  // Create new files
+  List_iterator_fast<char> it2(create_list);
+  while ((file_name=it2++))
+  {  
+    DBUG_PRINT("info", ("Table %s need discovery", name));
+    ha_create_table_from_engine(thd, db, file_name, true);    
+    files->push_back(thd->strdup(file_name)); 
+  }
+
+  pthread_mutex_unlock(&LOCK_open);      
+  
+  hash_free(&ok_tables);
+  hash_free(&ndb_tables);
+  DBUG_RETURN(0);    
 }
-#endif
 
 
 /*
@@ -3791,7 +3939,7 @@ bool ndbcluster_init()
 
   ndbcluster_inited= 1;
 #ifdef USE_DISCOVER_ON_STARTUP
-  if (res == 0 && ndb_discover_tables() != 0)
+  if (ndb_discover_tables() != 0)
     DBUG_RETURN(TRUE);    
 #endif
   DBUG_RETURN(false);
@@ -3807,7 +3955,6 @@ bool ndbcluster_init()
 bool ndbcluster_end()
 {
   DBUG_ENTER("ndbcluster_end");
-
   if(g_ndb)
     delete g_ndb;
   g_ndb= NULL;
@@ -3817,9 +3964,6 @@ bool ndbcluster_end()
   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);
@@ -4138,7 +4282,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
     if (pOp == NULL)
       break;
     
-    NdbResultSet* rs= pOp->readTuples(NdbScanOperation::LM_Dirty); 
+    NdbResultSet* rs= pOp->readTuples(NdbOperation::LM_CommittedRead); 
     if (rs == 0)
       break;
     
@@ -4178,4 +4322,30 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
   DBUG_RETURN(-1);
 }
 
+/*
+  Create a .ndb file to serve as a placeholder indicating 
+  that the table with this name is a ndb table
+*/
+
+int ha_ndbcluster::write_ndb_file()
+{
+  File file;
+  bool error=1;
+  char path[FN_REFLEN];
+  
+  DBUG_ENTER("write_ndb_file");
+  DBUG_PRINT("enter", ("db: %s, name: %s", m_dbname, m_tabname));
+
+  (void)strxnmov(path, FN_REFLEN, 
+		 mysql_data_home,"/",m_dbname,"/",m_tabname,ha_ndb_ext,NullS);
+
+  if ((file=my_create(path, CREATE_MODE,O_RDWR | O_TRUNC,MYF(MY_WME))) >= 0)
+  {
+    // It's an empty file
+    error=0;
+    my_close(file,MYF(0));
+  }
+  DBUG_RETURN(error);
+}
+
 #endif /* HAVE_NDBCLUSTER_DB */
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index e484f4703b2..36452033516 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -52,7 +52,6 @@ typedef enum ndb_index_type {
 typedef struct ndb_index_data {
   NDB_INDEX_TYPE type;
   void *index;
-  const char * unique_name;
   void *unique_index;
 } NDB_INDEX_DATA;
 
@@ -180,8 +179,6 @@ class ha_ndbcluster: public handler
   int build_index_list(TABLE *tab, enum ILBP phase);
   int get_metadata(const char* path);
   void release_metadata();
-  const char* get_index_name(uint idx_no) const;
-  const char* get_unique_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;
   
@@ -226,6 +223,8 @@ class ha_ndbcluster: public handler
   int ndb_err(NdbConnection*);
   bool uses_blob_value(bool all_fields);
 
+  int write_ndb_file();
+
  private:
   int check_ndb_connection();
 
@@ -277,8 +276,11 @@ int ndbcluster_rollback(THD *thd, void* ndb_transaction);
 
 void ndbcluster_close_connection(THD *thd);
 
-int ndbcluster_discover(const char* dbname, const char* name,
+int ndbcluster_discover(THD* thd, const char* dbname, const char* name,
 			const void** frmblob, uint* frmlen);
+int ndbcluster_find_files(THD *thd,const char *db,const char *path,
+			  const char *wild, bool dir, List<char> *files);
+int ndbcluster_table_exists(THD* thd, const char *db, const char *name);
 int ndbcluster_drop_database(const char* path);
 
 void ndbcluster_print_error(int error, const NdbOperation *error_op);
diff --git a/sql/handler.cc b/sql/handler.cc
index da911ad9172..23b4fbe4835 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1103,6 +1103,16 @@ void handler::print_error(int error, myf errflag)
   case HA_ERR_NO_REFERENCED_ROW:
     textno=ER_NO_REFERENCED_ROW;
     break;
+  case HA_ERR_NO_SUCH_TABLE:
+  {
+    char *db;
+    char buff[FN_REFLEN];
+    uint length=dirname_part(buff,table->path);
+    buff[length-1]=0;
+    db=buff+dirname_length(buff);
+    my_error(ER_NO_SUCH_TABLE,MYF(0),db,table->table_name);
+    break;
+  }
   default:
     {
       /* The error was "unknown" to this function.
@@ -1249,6 +1259,71 @@ int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
   DBUG_RETURN(error != 0);
 }
 
+/*
+  Try to discover table from engine and 
+  if found, write the frm file to disk.
+  
+  RETURN VALUES:
+   0 : Table existed in engine and created 
+       on disk if so requested
+   1 : Table does not exist
+  >1 : error
+
+*/
+
+int ha_create_table_from_engine(THD* thd, 
+				const char *db, 
+				const char *name,
+				bool create_if_found)
+{
+  int error= 0;  
+  const void* frmblob = NULL;
+  uint frmlen = 0;
+  char path[FN_REFLEN];
+  HA_CREATE_INFO create_info;
+  TABLE table;
+  DBUG_ENTER("ha_create_table_from_engine");
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
+  DBUG_PRINT("enter", ("create_if_found: %d", create_if_found));
+
+  bzero((char*) &create_info,sizeof(create_info));
+
+  if ((error= ha_discover(thd, db, name, &frmblob, &frmlen)))
+    DBUG_RETURN(error); 
+
+  // Table exists in handler  
+  if  (create_if_found)
+  {
+    (void)strxnmov(path,FN_REFLEN,mysql_data_home,"/",db,"/",name,NullS);
+    // Save the frm file    
+    if ((error = writefrm(path, frmblob, frmlen)))
+      goto err_end;
+
+    if (openfrm(path,"",0,(uint) READ_ALL, 0, &table))
+      DBUG_RETURN(1);
+
+    update_create_info_from_table(&create_info, &table);
+    create_info.table_options|= HA_CREATE_FROM_ENGINE;
+
+    if (lower_case_table_names == 2 &&
+	!(table.file->table_flags() & HA_FILE_BASED))
+    {
+      /* Ensure that handler gets name in lower case */
+      strmov(path, name);
+      my_casedn_str(files_charset_info, path);
+      name= path;
+    }
+    
+    error=table.file->create(path,&table,&create_info);
+    VOID(closefrm(&table));
+  }
+
+err_end:
+  if (frmblob)
+    my_free((char*) frmblob,MYF(0));
+  DBUG_RETURN(error);  
+}
+
 static int NEAR_F delete_file(const char *name,const char *ext,int extflag)
 {
   char buff[FN_REFLEN];
@@ -1356,15 +1431,15 @@ 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 ha_discover(THD* thd, const char* db, 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));
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
 #ifdef HAVE_NDBCLUSTER_DB
   if (have_ndbcluster == SHOW_OPTION_YES)
-    error= ndbcluster_discover(dbname, name, frmblob, frmlen);
+    error= ndbcluster_discover(thd, db, name, frmblob, frmlen);
 #endif
   if (!error)
     statistic_increment(ha_discover_count,&LOCK_status);
@@ -1372,6 +1447,52 @@ int ha_discover(const char* dbname, const char* name,
 }
 
 
+/*
+  Call this function in order to give the handler the possiblity 
+  to ask engine if there are any new tables that should be written to disk 
+  or any dropped tables that need to be removed from disk
+*/
+
+int
+ha_find_files(THD *thd,const char *db,const char *path,
+	      const char *wild, bool dir, List<char> *files)
+{
+  int error= 0;
+  DBUG_ENTER("ha_find_files");
+  DBUG_PRINT("enter", ("db: %s, path: %s, wild: %s, dir: %d", 
+		       db, path, wild, dir));
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    error= ndbcluster_find_files(thd, db, path, wild, dir, files);
+#endif
+  DBUG_RETURN(error);
+  
+  
+}
+
+/*
+  Ask handler if the table exists in engine
+
+  RETURN
+    0                   Table does not exist
+    1                   Table exists
+    #                   Error code
+
+ */
+int ha_table_exists(THD* thd, const char* db, const char* name)
+{
+  int error= 2;
+  DBUG_ENTER("ha_table_exists");
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    error= ndbcluster_table_exists(thd, db, name);
+#endif
+  DBUG_RETURN(error);
+}
+
+
+
 /*
   Read first row between two ranges.
   Store ranges for future calls to read_range_next
diff --git a/sql/handler.h b/sql/handler.h
index 443b43a5ea3..a7ce4e708fd 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -539,6 +539,8 @@ void ha_close_connection(THD* thd);
 enum db_type ha_checktype(enum db_type database_type);
 int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
 		    bool update_create_info);
+int ha_create_table_from_engine(THD* thd, const char *db, const char *name,
+				bool create_if_found);
 int ha_delete_table(enum db_type db_type, const char *path);
 void ha_drop_database(char* path);
 int ha_init_key_cache(const char *name, KEY_CACHE *key_cache);
@@ -560,5 +562,10 @@ bool ha_flush_logs(void);
 int ha_enable_transaction(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,
+int ha_discover(THD* thd, const char* dbname, const char* name,
 		const void** frmblob, uint* frmlen);
+int ha_find_files(THD *thd,const char *db,const char *path,
+		  const char *wild, bool dir,List<char>* files);
+int ha_table_exists(THD* thd, const char* db, const char* name);
+
+
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index e5aa638ca35..5d04d145563 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1005,8 +1005,6 @@ 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/sql_base.cc b/sql/sql_base.cc
index 65ac38cdd6e..2770d93bf26 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1343,7 +1343,7 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
       */
       if (discover_retry_count++ != 0)
        goto err;
-      if (create_table_from_handler(db, name, true) != 0)
+      if (ha_create_table_from_engine(thd, db, name, true) != 0)
        goto err;
 
       thd->clear_error(); // Clear error message
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 714fb0a2efe..413de53f4bc 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -374,6 +374,7 @@ mysql_find_files(THD *thd,List<char> *files, const char *db,const char *path,
 
   if (wild && !wild[0])
     wild=0;
+
   bzero((char*) &table_list,sizeof(table_list));
 
   if (!(dirp = my_dir(path,MYF(MY_WME | (dir ? MY_WANT_STAT : 0)))))
@@ -445,6 +446,9 @@ mysql_find_files(THD *thd,List<char> *files, const char *db,const char *path,
   }
   DBUG_PRINT("info",("found: %d files", files->elements));
   my_dirend(dirp);
+
+  VOID(ha_find_files(thd,db,path,wild,dir,files));
+
   DBUG_RETURN(0);
 }
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 113f71236bf..3d5aaf0c2ec 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -222,7 +222,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       strxmov(path, mysql_data_home, "/", db, "/", alias, reg_ext, NullS);
       (void) unpack_filename(path,path);
     }
-    if (drop_temporary || access(path,F_OK))
+    if (drop_temporary || 
+	(access(path,F_OK) && ha_create_table_from_engine(thd,db,alias,true)))
     {
       if (if_exists)
 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
@@ -1243,8 +1244,8 @@ int mysql_create_table(THD *thd,const char *db, const char *table_name,
   {
     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))
+    if (!ha_create_table_from_engine(thd, db, table_name,
+				     create_if_not_exists))
     {
       DBUG_PRINT("info", ("Table already existed in handler"));
 
diff --git a/sql/table.cc b/sql/table.cc
index 8b018d61e5a..7b64ef7a48c 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -705,6 +705,14 @@ int openfrm(const char *name, const char *alias, uint db_stat, uint prgflag,
       outparam->crashed=((err == HA_ERR_CRASHED_ON_USAGE) &&
 			 outparam->file->auto_repair() &&
 			 !(ha_open_flags & HA_OPEN_FOR_REPAIR));
+
+      if (err==HA_ERR_NO_SUCH_TABLE)
+      {
+	/* The table did not exists in storage engine, use same error message
+	   as if the .frm file didn't exist */
+	error= 1;
+	my_errno= ENOENT;
+      }
       goto err_not_open; /* purecov: inspected */
     }
   }