Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.1-new

into  mysql.com:/usr/local/home/marty/MySQL/mysql-5.1-new-wl1892


sql/ha_ndbcluster_binlog.cc:
  Auto merged
storage/ndb/include/ndbapi/NdbDictionary.hpp:
  Auto merged
storage/ndb/src/ndbapi/NdbDictionary.cpp:
  Auto merged

configure.in

@@ -7,7 +7,7 @@ AC_INIT(sql/mysqld.cc)
 AC_CANONICAL_SYSTEM
 # The Docs Makefile.am parses this line!
 # remember to also change ndb version below and update version.c in ndb
-AM_INIT_AUTOMAKE(mysql, 5.1.6-alpha)
+AM_INIT_AUTOMAKE(mysql, 5.1.7-beta)
 AM_CONFIG_HEADER(config.h)
 
 PROTOCOL_VERSION=10

mysql-test/r/ndb_binlog_ddl_multi.result

@@ -31,7 +31,6 @@ master-bin1.000001	#	Query	#	#	use `test`; alter table t2 add column (b int)
 reset master;
 reset master;
 ALTER DATABASE mysqltest CHARACTER SET latin1;
-insert into mysqltest.t1 values (1);
 drop table mysqltest.t1;
 show binlog events from 102;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -39,12 +38,6 @@ master-bin.000001	#	Query	#	#	ALTER DATABASE mysqltest CHARACTER SET latin1
 master-bin.000001	#	Query	#	#	BEGIN
 master-bin.000001	#	Table_map	#	#	cluster_replication.apply_status
 master-bin.000001	#	Write_rows	#	#	
-master-bin.000001	#	Table_map	#	#	mysqltest.t1
-master-bin.000001	#	Write_rows	#	#	
-master-bin.000001	#	Query	#	#	COMMIT
-master-bin.000001	#	Query	#	#	BEGIN
-master-bin.000001	#	Table_map	#	#	cluster_replication.apply_status
-master-bin.000001	#	Write_rows	#	#	
 master-bin.000001	#	Query	#	#	COMMIT
 master-bin.000001	#	Query	#	#	use `mysqltest`; drop table `t1`
 reset master;

mysql-test/r/ndb_dd_basic.result

@@ -21,6 +21,14 @@ CREATE TABLE t1
 (pk1 INT NOT NULL PRIMARY KEY, b INT NOT NULL, c INT NOT NULL)
 TABLESPACE ts1 STORAGE DISK
 ENGINE=NDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `pk1` int(11) NOT NULL,
+  `b` int(11) NOT NULL,
+  `c` int(11) NOT NULL,
+  PRIMARY KEY  (`pk1`)
+) TABLESPACE ts1 STORAGE DISK ENGINE=ndbcluster DEFAULT CHARSET=latin1 PARTITION BY KEY () 
 INSERT INTO t1 VALUES (0, 0, 0);
 SELECT * FROM t1;
 pk1	b	c

mysql-test/r/ndb_dd_disk2memory.result

@@ -226,7 +226,7 @@ t1	CREATE TABLE `t1` (
   `b` int(11) NOT NULL,
   `c` int(11) NOT NULL,
   PRIMARY KEY  (`pk1`)
-) ENGINE=ndbcluster DEFAULT CHARSET=latin1 PARTITION BY KEY () 
+) TABLESPACE table_space1 STORAGE DISK ENGINE=ndbcluster DEFAULT CHARSET=latin1 PARTITION BY KEY () 
 ALTER TABLE test.t2 TABLESPACE table_space1 STORAGE DISK
 ENGINE=NDB;
 SHOW CREATE TABLE test.t2;
@@ -236,7 +236,7 @@ t2	CREATE TABLE `t2` (
   `b2` int(11) NOT NULL,
   `c2` int(11) NOT NULL,
   PRIMARY KEY  (`pk2`)
-) ENGINE=ndbcluster DEFAULT CHARSET=latin1 PARTITION BY KEY () 
+) TABLESPACE table_space1 STORAGE DISK ENGINE=ndbcluster DEFAULT CHARSET=latin1 PARTITION BY KEY () 
 ALTER TABLE test.t1 ENGINE=NDBCLUSTER;
 SHOW CREATE TABLE test.t1;
 Table	Create Table

mysql-test/t/ndb_binlog_ddl_multi.test

@@ -67,7 +67,6 @@ ALTER DATABASE mysqltest CHARACTER SET latin1;
 
 # drop table and drop should come after data events
 --connection server2
-insert into mysqltest.t1 values (1);
 drop table mysqltest.t1;
 
 --connection server1

mysql-test/t/ndb_dd_basic.test

@@ -54,6 +54,8 @@ CREATE TABLE t1
 TABLESPACE ts1 STORAGE DISK
 ENGINE=NDB;
 
+SHOW CREATE TABLE t1;
+
 INSERT INTO t1 VALUES (0, 0, 0);
 SELECT * FROM t1;
 

sql/ha_ndbcluster.cc

@@ -8906,28 +8906,40 @@ ha_ndbcluster::generate_scan_filter(Ndb_cond_stack *ndb_cond_stack,
 /*
   get table space info for SHOW CREATE TABLE
 */
-char* ha_ndbcluster::get_tablespace_create_info()
+char* ha_ndbcluster::get_tablespace_name()
 {
-  const char tablespace_key[]= " TABLESPACE ";
-  const char storage_key[]= " STORAGE DISK";
-  char*	str= 0;
-
   Ndb *ndb= get_ndb();
   NDBDICT *ndbdict= ndb->getDictionary();
+  NdbError ndberr;
+  Uint32 id;
   ndb->setDatabaseName(m_dbname);
   const NDBTAB *ndbtab= ndbdict->getTable(m_tabname);
   if (ndbtab == 0)
+  {
+    ndberr= ndbdict->getNdbError();
+    goto err;
+  }
+  if (!ndbtab->getTablespace(&id))
+  {
     return 0;
-
-  // TODO retrieve table space name if there is one
+  }
+  {
+    NdbDictionary::Tablespace ts= ndbdict->getTablespace(id);
+    ndberr= ndbdict->getNdbError();
+    if(ndberr.classification != ndberror_cl_none)
+      goto err;
+    return (my_strdup(ts.getName(), MYF(0)));
+  }
+err:
+  if (ndberr.status == NdbError::TemporaryError)
+    push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+			ER_GET_TEMPORARY_ERRMSG, ER(ER_GET_TEMPORARY_ERRMSG),
+			ndberr.code, ndberr.message, "NDB");
+  else
+    push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+			ER_GET_ERRMSG, ER(ER_GET_ERRMSG),
+			ndberr.code, ndberr.message, "NDB");
   return 0;
-
-  const char *tablespace_name= "<name>";
-
-  uint len= sizeof(tablespace_key) + strlen(tablespace_name) + sizeof(storage_key);
-  str= my_malloc(len, MYF(0));
-  strxnmov(str, len, tablespace_key, tablespace_name, storage_key, NullS);
-  return(str);
 }
 
 /*

sql/ha_ndbcluster.h

@@ -721,7 +721,7 @@ static void set_tabname(const char *pathname, char *tabname);
   uint set_up_partition_info(partition_info *part_info,
                              TABLE *table,
                              void *tab);
-  char* get_tablespace_create_info();
+  char* get_tablespace_name();
   int set_range_data(void *tab, partition_info* part_info);
   int set_list_data(void *tab, partition_info* part_info);
   int complemented_pk_read(const byte *old_data, byte *new_data,

sql/ha_ndbcluster_binlog.cc

@@ -1715,6 +1715,12 @@ int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
 
   if (share)
   {
+    if (share->op || share->op_old)
+    {
+      my_errno= HA_ERR_TABLE_EXIST;
+      pthread_mutex_unlock(&ndbcluster_mutex);
+      DBUG_RETURN(1);
+    }
     handle_trailing_share(share);
   }
 

sql/handler.h

@@ -1713,8 +1713,8 @@ class handler :public Sql_alloc
   { return FALSE; }
   virtual char* get_foreign_key_create_info()
   { return(NULL);}  /* gets foreign key create string from InnoDB */
-  virtual char* get_tablespace_create_info()
-  { return(NULL);}  /* gets tablespace create string from handler */
+  virtual char* get_tablespace_name()
+  { return(NULL);}  /* gets tablespace name from handler */
   /* used in ALTER TABLE; 1 if changing storage engine is allowed */
   virtual bool can_switch_engines() { return 1; }
   /* used in REPLACE; is > 0 if table is referred by a FOREIGN KEY */

sql/sql_show.cc

@@ -1141,9 +1141,11 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       to the CREATE TABLE statement
     */
 
-    if ((for_str= file->get_tablespace_create_info()))
+    if ((for_str= file->get_tablespace_name()))
     {
+      packet->append(" TABLESPACE ");
       packet->append(for_str, strlen(for_str));
+      packet->append(" STORAGE DISK");
       my_free(for_str, MYF(0));
     }
 

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -784,7 +784,7 @@ public:
     void setTablespace(const char * name);
     void setTablespace(const class Tablespace &);
     const char * getTablespace() const;
-    Uint32 getTablespaceId() const;
+    bool getTablespace(Uint32 *id= 0, Uint32 *version= 0) const;
 
     /**
      * Get table object type
@@ -1744,6 +1744,7 @@ public:
     int createTablespace(const Tablespace &);
     int dropTablespace(const Tablespace&);
     Tablespace getTablespace(const char * name);
+    Tablespace getTablespace(Uint32 tablespaceId);
 
     int createDatafile(const Datafile &, bool overwrite_existing = false);
     int dropDatafile(const Datafile&);

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -600,10 +600,16 @@ NdbDictionary::Table::createTableInDb(Ndb* pNdb, bool equalOk) const {
   return pNdb->getDictionary()->createTable(* this);
 }
 
-Uint32
-NdbDictionary::Table::getTablespaceId() const 
+bool
+NdbDictionary::Table::getTablespace(Uint32 *id, Uint32 *version) const 
 {
-  return m_impl.m_tablespace_id;
+  if (m_impl.m_tablespace_id == RNIL)
+    return false;
+  if (id)
+    *id= m_impl.m_tablespace_id;
+  if (version)
+    *version= m_impl.m_version;
+  return true;
 }
 
 void 
@@ -1693,6 +1699,15 @@ NdbDictionary::Dictionary::getTablespace(const char * name){
   return tmp;
 }
 
+NdbDictionary::Tablespace
+NdbDictionary::Dictionary::getTablespace(Uint32 tablespaceId){
+  NdbDictionary::Tablespace tmp;
+  m_impl.m_receiver.get_filegroup(NdbTablespaceImpl::getImpl(tmp), 
+				  NdbDictionary::Object::Tablespace,
+                                  tablespaceId);
+  return tmp;
+}
+
 int
 NdbDictionary::Dictionary::createDatafile(const Datafile & df, bool force){
   return m_impl.createDatafile(NdbDatafileImpl::getImpl(df), force);

storage/ndb/tools/restore/consumer_restore.cpp

@@ -658,11 +658,11 @@ BackupRestore::table(const TableS & table){
     NdbDictionary::Table copy(*table.m_dictTable);
 
     copy.setName(split[2].c_str());
-    if (copy.getTablespaceId() != RNIL)
+    Uint32 id;
+    if (copy.getTablespace(&id))
     {
-      Uint32 id = copy.getTablespaceId();
       debug << "Connecting " << name << " to tablespace oldid: " << id << flush;
-      NdbDictionary::Tablespace* ts = m_tablespaces[copy.getTablespaceId()];
+      NdbDictionary::Tablespace* ts = m_tablespaces[id];
       debug << " newid: " << ts->getObjectId() << endl;
       copy.setTablespace(* ts);
     }