Merge perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new-ndb

BUILD/compile-ndb-autotest

+#! /bin/sh
+
+path=`dirname $0`
+. "$path/SETUP.sh"
+
+extra_flags="$fast_cflags $max_cflags -g"
+extra_configs="$max_configs --with-ndb-test --with-ndb-ccflags="-DERROR_INSERT""
+
+. "$path/FINISH.sh"

mysql-test/r/ndb_dd_ddl.result

@@ -4,12 +4,12 @@ CREATE DATABASE mysqltest;
 **** Begin Duplicate Statement Testing ****
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE = 1M
 ENGINE=NDB;
 CREATE LOGFILE GROUP lg2
 ADD UNDOFILE 'undofile2.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE 1M
 ENGINE NDB;
 ERROR HY000: Failed to create LOGFILE GROUP
@@ -19,35 +19,35 @@ Error	1296	Got error 1514 'Currently there is a limit of one logfile group' from
 Error	1515	Failed to create LOGFILE GROUP
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE = 1M
 ENGINE=NDB;
 ERROR HY000: Failed to create LOGFILE GROUP
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE 4M ENGINE NDB;
+INITIAL_SIZE 1M ENGINE NDB;
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE 4M ENGINE=NDB;
+INITIAL_SIZE 1M ENGINE=NDB;
 ERROR HY000: Failed to alter: CREATE UNDOFILE
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 ERROR HY000: Failed to create TABLESPACE
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE=NDB;
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE=NDB;
 ERROR HY000: Failed to alter:  CREATE DATAFILE
 CREATE TABLE mysqltest.t1
@@ -94,20 +94,20 @@ DROP DATABASE IF EXISTS mysqltest;
 **** Begin Statment CaSe Testing ****
 creaTE LOgfilE GrOuP lg1
 adD undoFILE 'undofile.dat'
-initiAL_siZE 16M
+initiAL_siZE 1M
 UnDo_BuFfEr_SiZe = 1M
 ENGInE=NDb;
 altER LOgFiLE GrOUp lg1
 AdD UnDOfILe 'uNdOfiLe02.daT'
-INItIAl_SIzE 4M ENgINE nDB;
+INItIAl_SIzE 1M ENgINE nDB;
 CrEAtE TABLEspaCE ts1
 ADD DATAfilE 'datafile.dat'
 UsE LoGFiLE GRoUP lg1
-INITiaL_SizE 12M
+INITiaL_SizE 1M
 ENGiNe NDb;
 AlTeR tAbLeSpAcE ts1
 AdD DaTaFiLe 'dAtAfiLe2.daT'
-InItIaL_SiZe 12M
+InItIaL_SiZe 1M
 EnGiNe=NDB;
 CREATE TABLE t1
 (pk1 int not null primary key, b int not null, c int not null)
@@ -129,21 +129,21 @@ EnGiNe=nDb;
 **** Begin = And No = Testing ****
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE=16M
+INITIAL_SIZE=1M
 UNDO_BUFFER_SIZE=1M
 ENGINE=NDB;
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE=4M 
+INITIAL_SIZE=1M 
 ENGINE=NDB;
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE=12M
+INITIAL_SIZE=1M
 ENGINE=NDB;
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE=12M
+INITIAL_SIZE=1M
 ENGINE=NDB;
 CREATE TABLE t1
 (pk1 INT NOT NULL PRIMARY KEY, b INT NOT NULL, c INT NOT NULL)
@@ -165,21 +165,21 @@ ENGINE=NDB;
 
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE 1M
 ENGINE NDB;
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE 4M
+INITIAL_SIZE 1M
 ENGINE NDB;
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 CREATE TABLE t1
 (pk1 INT NOT NULL PRIMARY KEY, b INT NOT NULL, c INT NOT NULL)
@@ -199,3 +199,19 @@ ENGINE NDB;
 DROP LOGFILE GROUP lg1
 ENGINE NDB;
 **** End = And No =  ****
+create table t1 (a int primary key) engine = myisam;
+create logfile group lg1 add undofile 'MYSQLTEST_VARDIR/master-data/test/t1.frm' initial_size 1M undo_buffer_size = 1M engine=ndb;;
+ERROR HY000: Failed to create UNDOFILE
+create logfile group lg1
+add undofile 'undofile.dat'
+initial_size 1M
+undo_buffer_size = 1M
+engine=ndb;
+create tablespace ts1 add datafile 'MYSQLTEST_VARDIR/master-data/test/t1.frm' use logfile group lg1 initial_size 1M engine ndb;;
+ERROR HY000: Failed to create DATAFILE
+drop tablespace ts1
+engine ndb;
+ERROR HY000: Failed to drop TABLESPACE
+drop logfile group lg1
+engine ndb;
+drop table t1;

mysql-test/t/ndb_dd_ddl.test

@@ -40,7 +40,7 @@ CREATE DATABASE mysqltest;
 
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE = 1M
 ENGINE=NDB;
 
@@ -48,7 +48,7 @@ ENGINE=NDB;
 --error ER_CREATE_FILEGROUP_FAILED
 CREATE LOGFILE GROUP lg2
 ADD UNDOFILE 'undofile2.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE 1M
 ENGINE NDB;
 SHOW WARNINGS;
@@ -56,42 +56,42 @@ SHOW WARNINGS;
 --error ER_CREATE_FILEGROUP_FAILED
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE = 1M
 ENGINE=NDB;
 
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE 4M ENGINE NDB;
+INITIAL_SIZE 1M ENGINE NDB;
 
 --error ER_ALTER_FILEGROUP_FAILED
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE 4M ENGINE=NDB;
+INITIAL_SIZE 1M ENGINE=NDB;
 
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 
 --error ER_CREATE_FILEGROUP_FAILED
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 
 
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE=NDB;
 
 --error ER_ALTER_FILEGROUP_FAILED
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE=NDB;
 
 CREATE TABLE mysqltest.t1
@@ -155,23 +155,23 @@ DROP DATABASE IF EXISTS mysqltest;
 
 creaTE LOgfilE GrOuP lg1
 adD undoFILE 'undofile.dat'
-initiAL_siZE 16M
+initiAL_siZE 1M
 UnDo_BuFfEr_SiZe = 1M
 ENGInE=NDb;
 
 altER LOgFiLE GrOUp lg1
 AdD UnDOfILe 'uNdOfiLe02.daT'
-INItIAl_SIzE 4M ENgINE nDB;
+INItIAl_SIzE 1M ENgINE nDB;
 
 CrEAtE TABLEspaCE ts1
 ADD DATAfilE 'datafile.dat'
 UsE LoGFiLE GRoUP lg1
-INITiaL_SizE 12M
+INITiaL_SizE 1M
 ENGiNe NDb;
 
 AlTeR tAbLeSpAcE ts1
 AdD DaTaFiLe 'dAtAfiLe2.daT'
-InItIaL_SiZe 12M
+InItIaL_SiZe 1M
 EnGiNe=NDB;
 
 CREATE TABLE t1
@@ -203,24 +203,24 @@ EnGiNe=nDb;
 
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE=16M
+INITIAL_SIZE=1M
 UNDO_BUFFER_SIZE=1M
 ENGINE=NDB;
 
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE=4M 
+INITIAL_SIZE=1M 
 ENGINE=NDB;
 
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE=12M
+INITIAL_SIZE=1M
 ENGINE=NDB;
 
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE=12M
+INITIAL_SIZE=1M
 ENGINE=NDB;
 
 CREATE TABLE t1
@@ -250,24 +250,24 @@ ENGINE=NDB;
 
 CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
-INITIAL_SIZE 16M
+INITIAL_SIZE 1M
 UNDO_BUFFER_SIZE 1M
 ENGINE NDB;
 
 ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
-INITIAL_SIZE 4M
+INITIAL_SIZE 1M
 ENGINE NDB;
 
 CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 
 ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile2.dat'
-INITIAL_SIZE 12M
+INITIAL_SIZE 1M
 ENGINE NDB;
 
 CREATE TABLE t1
@@ -297,5 +297,32 @@ ENGINE NDB;
 --echo **** End = And No =  ****
 ############ End = And No = ##################
 
-# End 5.1 test
+###
+#
+# bug#16341
+create table t1 (a int primary key) engine = myisam;
+
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--error ER_CREATE_FILEGROUP_FAILED
+--eval create logfile group lg1 add undofile '$MYSQLTEST_VARDIR/master-data/test/t1.frm' initial_size 1M undo_buffer_size = 1M engine=ndb;
+
+create logfile group lg1
+add undofile 'undofile.dat'
+initial_size 1M
+undo_buffer_size = 1M
+engine=ndb;
+
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--error ER_CREATE_FILEGROUP_FAILED
+--eval create tablespace ts1 add datafile '$MYSQLTEST_VARDIR/master-data/test/t1.frm' use logfile group lg1 initial_size 1M engine ndb;
+
+--error ER_DROP_FILEGROUP_FAILED
+drop tablespace ts1
+engine ndb;
 
+drop logfile group lg1
+engine ndb;
+
+drop table t1;
+
+# End 5.1 test

sql/ha_ndbcluster.cc

@@ -9983,6 +9983,7 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   }
 
+  NdbError err;
   NDBDICT *dict = ndb->getDictionary();
   int error;
   const char * errmsg;
@@ -9995,6 +9996,7 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
     
     NdbDictionary::Tablespace ndb_ts;
     NdbDictionary::Datafile ndb_df;
+    NdbDictionary::ObjectId objid;
     if (set_up_tablespace(info, &ndb_ts))
     {
       DBUG_RETURN(1);
@@ -10004,7 +10006,7 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
       DBUG_RETURN(1);
     }
     errmsg= "TABLESPACE";
-    if (dict->createTablespace(ndb_ts))
+    if (dict->createTablespace(ndb_ts, &objid))
     {
       DBUG_PRINT("error", ("createTablespace returned %d", error));
       goto ndberror;
@@ -10013,8 +10015,17 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
     errmsg= "DATAFILE";
     if (dict->createDatafile(ndb_df))
     {
+      err= dict->getNdbError();
+      NdbDictionary::Tablespace tmp= dict->getTablespace(ndb_ts.getName());
+      if (dict->getNdbError().code == 0 &&
+	  tmp.getObjectId() == objid.getObjectId() &&
+	  tmp.getObjectVersion() == objid.getObjectVersion())
+      {
+	dict->dropTablespace(tmp);
+      }
+      
       DBUG_PRINT("error", ("createDatafile returned %d", error));
-      goto ndberror;
+      goto ndberror2;
     }
     is_tablespace= 1;
     break;
@@ -10068,6 +10079,7 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
     error= ER_CREATE_FILEGROUP_FAILED;
     NdbDictionary::LogfileGroup ndb_lg;
     NdbDictionary::Undofile ndb_uf;
+    NdbDictionary::ObjectId objid;
     if (info->undo_file_name == NULL)
     {
       /*
@@ -10080,7 +10092,7 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
       DBUG_RETURN(1);
     }
     errmsg= "LOGFILE GROUP";
-    if (dict->createLogfileGroup(ndb_lg))
+    if (dict->createLogfileGroup(ndb_lg, &objid))
     {
       goto ndberror;
     }
@@ -10092,7 +10104,15 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
     errmsg= "UNDOFILE";
     if (dict->createUndofile(ndb_uf))
     {
-      goto ndberror;
+      err= dict->getNdbError();
+      NdbDictionary::LogfileGroup tmp= dict->getLogfileGroup(ndb_lg.getName());
+      if (dict->getNdbError().code == 0 &&
+	  tmp.getObjectId() == objid.getObjectId() &&
+	  tmp.getObjectVersion() == objid.getObjectVersion())
+      {
+	dict->dropLogfileGroup(tmp);
+      }
+      goto ndberror2;
     }
     break;
   }
@@ -10169,7 +10189,8 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
   DBUG_RETURN(FALSE);
 
 ndberror:
-  const NdbError err= dict->getNdbError();
+  err= dict->getNdbError();
+ndberror2:
   ERR_PRINT(err);
   ndb_to_mysql_error(&err);
   

storage/ndb/include/kernel/signaldata/CreateFilegroup.hpp

@@ -193,6 +193,7 @@ struct CreateFileConf {
   Uint32 senderData;
   Uint32 senderRef;
   Uint32 fileId;
+  Uint32 fileVersion;
 };
 
 #endif

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -163,6 +163,31 @@ public:
     };
   };
 
+  class Dictionary; // Forward declaration
+  
+  class ObjectId : public Object 
+  {
+  public:
+    ObjectId();
+    virtual ~ObjectId();
+    
+    /**
+     * Get status of object
+     */
+    virtual Status getObjectStatus() const;
+    
+    /**
+     * Get version of object
+     */
+    virtual int getObjectVersion() const;
+    
+    virtual int getObjectId() const;
+    
+  private:
+    friend class Dictionary;
+    class NdbDictObjectImpl & m_impl;
+  };
+  
   class Table; // forward declaration
   class Tablespace; // forward declaration
 //  class NdbEventOperation; // forward declaration
@@ -1781,20 +1806,20 @@ public:
      * @{
      */
     
-    int createLogfileGroup(const LogfileGroup &);
+    int createLogfileGroup(const LogfileGroup &, ObjectId* = 0);
     int dropLogfileGroup(const LogfileGroup&);
     LogfileGroup getLogfileGroup(const char * name);
 
-    int createTablespace(const Tablespace &);
+    int createTablespace(const Tablespace &, ObjectId* = 0);
     int dropTablespace(const Tablespace&);
     Tablespace getTablespace(const char * name);
     Tablespace getTablespace(Uint32 tablespaceId);
 
-    int createDatafile(const Datafile &, bool overwrite_existing = false);
+    int createDatafile(const Datafile &, bool overwrite_existing = false, ObjectId* = 0);
     int dropDatafile(const Datafile&);
     Datafile getDatafile(Uint32 node, const char * path);
     
-    int createUndofile(const Undofile &, bool overwrite_existing = false);
+    int createUndofile(const Undofile &, bool overwrite_existing = false, ObjectId * = 0);
     int dropUndofile(const Undofile&);
     Undofile getUndofile(Uint32 node, const char * path);
     

storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp

@@ -4731,11 +4731,6 @@ Dbdict::alterTab_writeTableConf(Signal* signal,
   SegmentedSectionPtr tabInfoPtr;
   getSection(tabInfoPtr, alterTabPtr.p->m_tabInfoPtrI);
   signal->setSection(tabInfoPtr, AlterTabReq::DICT_TAB_INFO);
-#ifndef DBUG_OFF
-  ndbout_c("DICT_TAB_INFO in DICT");
-  SimplePropertiesSectionReader reader(tabInfoPtr, getSectionSegmentPool());
-  reader.printAll(ndbout);
-#endif
   EXECUTE_DIRECT(SUMA, GSN_ALTER_TAB_REQ, signal,
                  AlterTabReq::SignalLength);
   releaseSections(signal);
@@ -7002,13 +6997,37 @@ void Dbdict::releaseTableObject(Uint32 tableId, bool removeFromHash)
 {
   TableRecordPtr tablePtr;
   c_tableRecordPool.getPtr(tablePtr, tableId);
-  if (removeFromHash){
+  if (removeFromHash)
+  {
     jam();
     release_object(tablePtr.p->m_obj_ptr_i);
   }
+  else
+  {
+    Rope tmp(c_rope_pool, tablePtr.p->tableName);
+    tmp.erase();
+  }
+  
+  {
+    Rope tmp(c_rope_pool, tablePtr.p->frmData);
+    tmp.erase();
+  }
+
+  {
+    Rope tmp(c_rope_pool, tablePtr.p->tsData);
+    tmp.erase();
+  }
+
+  {
+    Rope tmp(c_rope_pool, tablePtr.p->ngData);
+    tmp.erase();
+  }
+
+  {
+    Rope tmp(c_rope_pool, tablePtr.p->rangeData);
+    tmp.erase();
+  }
 
-  Rope frm(c_rope_pool, tablePtr.p->frmData);
-  frm.erase();
   tablePtr.p->tabState = TableRecord::NOT_DEFINED;
   
   LocalDLFifoList<AttributeRecord> list(c_attributeRecordPool, 
@@ -14277,6 +14296,7 @@ Dbdict::trans_commit_complete_done(Signal* signal,
     conf->senderRef = reference();
     conf->senderData = trans_ptr.p->m_senderData;
     conf->fileId = f_ptr.p->key;
+    conf->fileVersion = f_ptr.p->m_version;
 
     //@todo check api failed
     sendSignal(trans_ptr.p->m_senderRef, GSN_CREATE_FILE_CONF, signal, 

storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -246,6 +246,7 @@ inline const Uint32* ALIGN_WORD(const void* ptr)
 #define ZTUP_SCAN 10
 #define ZFREE_EXTENT 11
 #define ZUNMAP_PAGES 12
+#define ZFREE_VAR_PAGES 13
 
 #define ZSCAN_PROCEDURE 0
 #define ZCOPY_PROCEDURE 2
@@ -620,6 +621,7 @@ struct Fragrecord {
   Uint32 m_tablespace_id;
   Uint32 m_logfile_group_id;
   Disk_alloc_info m_disk_alloc_info;
+  Uint32 m_var_page_chunks;
 };
 typedef Ptr<Fragrecord> FragrecordPtr;
 
@@ -2335,6 +2337,7 @@ private:
 
 
   void releaseFragment(Signal* signal, Uint32 tableId);
+  void drop_fragment_free_var_pages(Signal*);
   void drop_fragment_free_exent(Signal*, TablerecPtr, FragrecordPtr, Uint32);
   void drop_fragment_unmap_pages(Signal*, TablerecPtr, FragrecordPtr, Uint32);
   void drop_fragment_unmap_page_callback(Signal* signal, Uint32, Uint32);

storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp

@@ -227,6 +227,12 @@ void Dbtup::execCONTINUEB(Signal* signal)
     drop_fragment_unmap_pages(signal, tabPtr, fragPtr, signal->theData[3]);
     return;
   }
+  case ZFREE_VAR_PAGES:
+  {
+    ljam();
+    drop_fragment_free_var_pages(signal);
+    return;
+  }
   default:
     ndbrequire(false);
     break;

storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp

@@ -140,6 +140,7 @@ void Dbtup::execTUPFRAGREQ(Signal* signal)
   regFragPtr.p->m_undo_complete= false;
   regFragPtr.p->m_lcp_scan_op = RNIL; 
   regFragPtr.p->m_lcp_keep_list = RNIL;
+  regFragPtr.p->m_var_page_chunks = RNIL;
   
   Uint32 noAllocatedPages= allocFragPages(regFragPtr.p, pages);
   
@@ -970,7 +971,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
     case -1:
       break;
     default:
-      ndbrequire(res == pagePtr.i);
+      ndbrequire((Uint32)res == pagePtr.i);
       drop_fragment_unmap_page_callback(signal, pos, res);
     }
     return;
@@ -1052,6 +1053,44 @@ Dbtup::drop_fragment_free_exent(Signal *signal,
     }
   }
   
+  signal->theData[0] = ZFREE_VAR_PAGES;
+  signal->theData[1] = tabPtr.i;
+  signal->theData[2] = fragPtr.i;
+  sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);  
+}
+
+void
+Dbtup::drop_fragment_free_var_pages(Signal* signal)
+{
+  ljam();
+  Uint32 tableId = signal->theData[1];
+  Uint32 fragPtrI = signal->theData[2];
+  
+  TablerecPtr tabPtr;
+  tabPtr.i= tableId;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+  
+  FragrecordPtr fragPtr;
+  fragPtr.i = fragPtrI;
+  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
+  
+  PagePtr pagePtr;
+  if ((pagePtr.i = fragPtr.p->m_var_page_chunks) != RNIL)
+  {
+    c_page_pool.getPtr(pagePtr);
+    Var_page* page = (Var_page*)pagePtr.p;
+    fragPtr.p->m_var_page_chunks = page->next_chunk;
+
+    Uint32 sz = page->chunk_size;
+    returnCommonArea(pagePtr.i, sz);
+    
+    signal->theData[0] = ZFREE_VAR_PAGES;
+    signal->theData[1] = tabPtr.i;
+    signal->theData[2] = fragPtr.i;
+    sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);  
+    return;
+  }
+  
   releaseFragPages(fragPtr.p);
   
   Uint32 i;

storage/ndb/src/kernel/blocks/dbtup/DbtupPageMap.cpp

@@ -300,6 +300,11 @@ void Dbtup::releaseFragPages(Fragrecord* regFragPtr)
 	tmp.remove();
       }
 
+      {
+	LocalSLList<Page> tmp(c_page_pool, regFragPtr->m_empty_pages);
+	tmp.remove();
+      }
+      
       return;
     } else {
       if (regPRPtr.p->type[indexPos] == ZNON_LEAF) {

storage/ndb/src/kernel/blocks/dbtup/DbtupVarAlloc.cpp

@@ -324,6 +324,13 @@ Dbtup::get_empty_var_page(Fragrecord* fragPtr)
     list.add(ret.i + 1, ptr);
   }
 
+  c_page_pool.getPtr(ret);
+  
+  Var_page* page = (Var_page*)ret.p;
+  page->chunk_size = cnt;
+  page->next_chunk = fragPtr->m_var_page_chunks;
+  fragPtr->m_var_page_chunks = ret.i;
+  
   return ret.i;
 }
 

storage/ndb/src/kernel/blocks/dbtup/tuppage.hpp

@@ -107,8 +107,14 @@ struct Tup_varsize_page
   Uint32 page_state;
   Uint32 next_page;
   Uint32 prev_page;
+  union {
     Uint32 first_cluster_page;
+    Uint32 chunk_size;
+  };
+  union {
     Uint32 last_cluster_page;
+    Uint32 next_chunk;
+  };
   Uint32 next_cluster_page;
   Uint32 prev_cluster_page;
   Uint32 frag_page_id;

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -18,6 +18,32 @@
 #include "NdbDictionaryImpl.hpp"
 #include <NdbOut.hpp>
 
+NdbDictionary::ObjectId::ObjectId()
+  : m_impl(* new NdbDictObjectImpl(NdbDictionary::Object::TypeUndefined))
+{
+}
+
+NdbDictionary::ObjectId::~ObjectId()
+{
+  NdbDictObjectImpl * tmp = &m_impl;  
+  delete tmp;
+}
+
+NdbDictionary::Object::Status
+NdbDictionary::ObjectId::getObjectStatus() const {
+  return m_impl.m_status;
+}
+
+int 
+NdbDictionary::ObjectId::getObjectVersion() const {
+  return m_impl.m_version;
+}
+
+int 
+NdbDictionary::ObjectId::getObjectId() const {
+  return m_impl.m_id;
+}
+
 /*****************************************************************
  * Column facade
  */
@@ -1799,17 +1825,22 @@ operator<<(NdbOut& out, const NdbDictionary::Column& col)
 }
 
 int
-NdbDictionary::Dictionary::createLogfileGroup(const LogfileGroup & lg){
-  return m_impl.createLogfileGroup(NdbLogfileGroupImpl::getImpl(lg));
+NdbDictionary::Dictionary::createLogfileGroup(const LogfileGroup & lg,
+					      ObjectId * obj)
+{
+  return m_impl.createLogfileGroup(NdbLogfileGroupImpl::getImpl(lg),
+				   obj ? &obj->m_impl : 0);
 }
 
 int
-NdbDictionary::Dictionary::dropLogfileGroup(const LogfileGroup & lg){
+NdbDictionary::Dictionary::dropLogfileGroup(const LogfileGroup & lg)
+{
   return m_impl.dropLogfileGroup(NdbLogfileGroupImpl::getImpl(lg));
 }
 
 NdbDictionary::LogfileGroup
-NdbDictionary::Dictionary::getLogfileGroup(const char * name){
+NdbDictionary::Dictionary::getLogfileGroup(const char * name)
+{
   NdbDictionary::LogfileGroup tmp;
   m_impl.m_receiver.get_filegroup(NdbLogfileGroupImpl::getImpl(tmp), 
 				  NdbDictionary::Object::LogfileGroup, name);
@@ -1817,17 +1848,22 @@ NdbDictionary::Dictionary::getLogfileGroup(const char * name){
 }
 
 int
-NdbDictionary::Dictionary::createTablespace(const Tablespace & lg){
-  return m_impl.createTablespace(NdbTablespaceImpl::getImpl(lg));
+NdbDictionary::Dictionary::createTablespace(const Tablespace & lg,
+					    ObjectId * obj)
+{
+  return m_impl.createTablespace(NdbTablespaceImpl::getImpl(lg),
+				 obj ? &obj->m_impl : 0);
 }
 
 int
-NdbDictionary::Dictionary::dropTablespace(const Tablespace & lg){
+NdbDictionary::Dictionary::dropTablespace(const Tablespace & lg)
+{
   return m_impl.dropTablespace(NdbTablespaceImpl::getImpl(lg));
 }
 
 NdbDictionary::Tablespace
-NdbDictionary::Dictionary::getTablespace(const char * name){
+NdbDictionary::Dictionary::getTablespace(const char * name)
+{
   NdbDictionary::Tablespace tmp;
   m_impl.m_receiver.get_filegroup(NdbTablespaceImpl::getImpl(tmp), 
 				  NdbDictionary::Object::Tablespace, name);
@@ -1835,7 +1871,8 @@ NdbDictionary::Dictionary::getTablespace(const char * name){
 }
 
 NdbDictionary::Tablespace
-NdbDictionary::Dictionary::getTablespace(Uint32 tablespaceId){
+NdbDictionary::Dictionary::getTablespace(Uint32 tablespaceId)
+{
   NdbDictionary::Tablespace tmp;
   m_impl.m_receiver.get_filegroup(NdbTablespaceImpl::getImpl(tmp), 
 				  NdbDictionary::Object::Tablespace,
@@ -1844,17 +1881,24 @@ NdbDictionary::Dictionary::getTablespace(Uint32 tablespaceId){
 }
 
 int
-NdbDictionary::Dictionary::createDatafile(const Datafile & df, bool force){
-  return m_impl.createDatafile(NdbDatafileImpl::getImpl(df), force);
+NdbDictionary::Dictionary::createDatafile(const Datafile & df, 
+					  bool force,
+					  ObjectId * obj)
+{
+  return m_impl.createDatafile(NdbDatafileImpl::getImpl(df), 
+			       force,
+			       obj ? &obj->m_impl : 0);
 }
 
 int
-NdbDictionary::Dictionary::dropDatafile(const Datafile& df){
+NdbDictionary::Dictionary::dropDatafile(const Datafile& df)
+{
   return m_impl.dropDatafile(NdbDatafileImpl::getImpl(df));
 }
 
 NdbDictionary::Datafile
-NdbDictionary::Dictionary::getDatafile(Uint32 node, const char * path){
+NdbDictionary::Dictionary::getDatafile(Uint32 node, const char * path)
+{
   NdbDictionary::Datafile tmp;
   m_impl.m_receiver.get_file(NdbDatafileImpl::getImpl(tmp),
 			     NdbDictionary::Object::Datafile,
@@ -1863,17 +1907,24 @@ NdbDictionary::Dictionary::getDatafile(Uint32 node, const char * path){
 }
 
 int
-NdbDictionary::Dictionary::createUndofile(const Undofile & df, bool force){
-  return m_impl.createUndofile(NdbUndofileImpl::getImpl(df), force);
+NdbDictionary::Dictionary::createUndofile(const Undofile & df, 
+					  bool force,
+					  ObjectId * obj)
+{
+  return m_impl.createUndofile(NdbUndofileImpl::getImpl(df), 
+			       force,
+			       obj ? &obj->m_impl : 0);
 }
 
 int
-NdbDictionary::Dictionary::dropUndofile(const Undofile& df){
+NdbDictionary::Dictionary::dropUndofile(const Undofile& df)
+{
   return m_impl.dropUndofile(NdbUndofileImpl::getImpl(df));
 }
 
 NdbDictionary::Undofile
-NdbDictionary::Dictionary::getUndofile(Uint32 node, const char * path){
+NdbDictionary::Dictionary::getUndofile(Uint32 node, const char * path)
+{
   NdbDictionary::Undofile tmp;
   m_impl.m_receiver.get_file(NdbUndofileImpl::getImpl(tmp),
 			     NdbDictionary::Object::Undofile,

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -4391,19 +4391,23 @@ NdbUndofileImpl::assign(const NdbUndofileImpl& org)
 }
 
 int 
-NdbDictionaryImpl::createDatafile(const NdbDatafileImpl & file, bool force){
+NdbDictionaryImpl::createDatafile(const NdbDatafileImpl & file, 
+				  bool force,
+				  NdbDictObjectImpl* obj)
+  
+{
   DBUG_ENTER("NdbDictionaryImpl::createDatafile");
   NdbFilegroupImpl tmp(NdbDictionary::Object::Tablespace);
   if(file.m_filegroup_version != ~(Uint32)0){
     tmp.m_id = file.m_filegroup_id;
     tmp.m_version = file.m_filegroup_version;
-    DBUG_RETURN(m_receiver.create_file(file, tmp));
+    DBUG_RETURN(m_receiver.create_file(file, tmp, force, obj));
   }
   
   
   if(m_receiver.get_filegroup(tmp, NdbDictionary::Object::Tablespace,
 			      file.m_filegroup_name.c_str()) == 0){
-    DBUG_RETURN(m_receiver.create_file(file, tmp, force));
+    DBUG_RETURN(m_receiver.create_file(file, tmp, force, obj));
   }
   DBUG_RETURN(-1); 
 }
@@ -4414,53 +4418,65 @@ NdbDictionaryImpl::dropDatafile(const NdbDatafileImpl & file){
 }
 
 int
-NdbDictionaryImpl::createUndofile(const NdbUndofileImpl & file, bool force){
+NdbDictionaryImpl::createUndofile(const NdbUndofileImpl & file, 
+				  bool force,
+				  NdbDictObjectImpl* obj)
+{
   DBUG_ENTER("NdbDictionaryImpl::createUndofile");
   NdbFilegroupImpl tmp(NdbDictionary::Object::LogfileGroup);
   if(file.m_filegroup_version != ~(Uint32)0){
     tmp.m_id = file.m_filegroup_id;
     tmp.m_version = file.m_filegroup_version;
-    DBUG_RETURN(m_receiver.create_file(file, tmp));
+    DBUG_RETURN(m_receiver.create_file(file, tmp, force, obj));
   }
   
   
   if(m_receiver.get_filegroup(tmp, NdbDictionary::Object::LogfileGroup,
 			      file.m_filegroup_name.c_str()) == 0){
-    DBUG_RETURN(m_receiver.create_file(file, tmp, force));
+    DBUG_RETURN(m_receiver.create_file(file, tmp, force, obj));
   }
   DBUG_PRINT("info", ("Failed to find filegroup"));
   DBUG_RETURN(-1);
 }
 
 int
-NdbDictionaryImpl::dropUndofile(const NdbUndofileImpl & file){
+NdbDictionaryImpl::dropUndofile(const NdbUndofileImpl & file)
+{
   return m_receiver.drop_file(file);
 }
 
 int
-NdbDictionaryImpl::createTablespace(const NdbTablespaceImpl & fg){
-  return m_receiver.create_filegroup(fg);
+NdbDictionaryImpl::createTablespace(const NdbTablespaceImpl & fg,
+				    NdbDictObjectImpl* obj)
+{
+  return m_receiver.create_filegroup(fg, obj);
 }
 
 int
-NdbDictionaryImpl::dropTablespace(const NdbTablespaceImpl & fg){
+NdbDictionaryImpl::dropTablespace(const NdbTablespaceImpl & fg)
+{
   return m_receiver.drop_filegroup(fg);
 }
 
 int
-NdbDictionaryImpl::createLogfileGroup(const NdbLogfileGroupImpl & fg){
- return m_receiver.create_filegroup(fg);
+NdbDictionaryImpl::createLogfileGroup(const NdbLogfileGroupImpl & fg,
+				      NdbDictObjectImpl* obj)
+{
+  return m_receiver.create_filegroup(fg, obj);
 }
 
 int
-NdbDictionaryImpl::dropLogfileGroup(const NdbLogfileGroupImpl & fg){
+NdbDictionaryImpl::dropLogfileGroup(const NdbLogfileGroupImpl & fg)
+{
   return m_receiver.drop_filegroup(fg);
 }
 
 int
 NdbDictInterface::create_file(const NdbFileImpl & file,
 			      const NdbFilegroupImpl & group,
-			      bool overwrite){
+			      bool overwrite,
+			      NdbDictObjectImpl* obj)
+{
   DBUG_ENTER("NdbDictInterface::create_file"); 
   UtilBufferWriter w(m_buffer);
   DictFilegroupInfo::File f; f.init();
@@ -4503,17 +4519,33 @@ NdbDictInterface::create_file(const NdbFileImpl & file,
     Send signal without time-out since creating files can take a very long
     time if the file is very big.
   */
-  DBUG_RETURN(dictSignal(&tSignal, ptr, 1,
+  int ret = dictSignal(&tSignal, ptr, 1,
 		       0, // master
 		       WAIT_CREATE_INDX_REQ,
 		       -1, 100,
-		         err));
+		       err);
+
+  if (ret == 0 && obj)
+  {
+    Uint32* data = (Uint32*)m_buffer.get_data();
+    obj->m_id = data[0];
+    obj->m_version = data[1];
+  }
+
+  DBUG_RETURN(ret);
 }
 
 void
 NdbDictInterface::execCREATE_FILE_CONF(NdbApiSignal * signal,
 				       LinearSectionPtr ptr[3])
 {
+  const CreateFileConf* conf=
+    CAST_CONSTPTR(CreateFileConf, signal->getDataPtr());
+  m_buffer.grow(4 * 2); // 2 words
+  Uint32* data = (Uint32*)m_buffer.get_data();
+  data[0] = conf->fileId;
+  data[1] = conf->fileVersion;
+  
   m_waiter.signal(NO_WAIT);  
 }
 
@@ -4529,7 +4561,8 @@ NdbDictInterface::execCREATE_FILE_REF(NdbApiSignal * signal,
 }
 
 int
-NdbDictInterface::drop_file(const NdbFileImpl & file){
+NdbDictInterface::drop_file(const NdbFileImpl & file)
+{
   DBUG_ENTER("NdbDictInterface::drop_file");
   NdbApiSignal tSignal(m_reference);
   tSignal.theReceiversBlockNumber = DBDICT;
@@ -4569,7 +4602,9 @@ NdbDictInterface::execDROP_FILE_REF(NdbApiSignal * signal,
 }
 
 int
-NdbDictInterface::create_filegroup(const NdbFilegroupImpl & group){
+NdbDictInterface::create_filegroup(const NdbFilegroupImpl & group,
+				   NdbDictObjectImpl* obj)
+{
   DBUG_ENTER("NdbDictInterface::create_filegroup");
   UtilBufferWriter w(m_buffer);
   DictFilegroupInfo::Filegroup fg; fg.init();
@@ -4638,17 +4673,32 @@ NdbDictInterface::create_filegroup(const NdbFilegroupImpl & group){
   ptr[0].sz = m_buffer.length() / 4;
 
   int err[] = { CreateFilegroupRef::Busy, CreateFilegroupRef::NotMaster, 0};
-  DBUG_RETURN(dictSignal(&tSignal, ptr, 1,
+  int ret = dictSignal(&tSignal, ptr, 1,
 		       0, // master
 		       WAIT_CREATE_INDX_REQ,
 		       DICT_WAITFOR_TIMEOUT, 100,
-		         err));
+		       err);
+  
+  if (ret == 0 && obj)
+  {
+    Uint32* data = (Uint32*)m_buffer.get_data();
+    obj->m_id = data[0];
+    obj->m_version = data[1];
+  }
+  
+  DBUG_RETURN(ret);
 }
 
 void
 NdbDictInterface::execCREATE_FILEGROUP_CONF(NdbApiSignal * signal,
 					    LinearSectionPtr ptr[3])
 {
+  const CreateFilegroupConf* conf=
+    CAST_CONSTPTR(CreateFilegroupConf, signal->getDataPtr());
+  m_buffer.grow(4 * 2); // 2 words
+  Uint32* data = (Uint32*)m_buffer.get_data();
+  data[0] = conf->filegroupId;
+  data[1] = conf->filegroupVersion;
   m_waiter.signal(NO_WAIT);  
 }
 
@@ -4664,7 +4714,8 @@ NdbDictInterface::execCREATE_FILEGROUP_REF(NdbApiSignal * signal,
 }
 
 int
-NdbDictInterface::drop_filegroup(const NdbFilegroupImpl & group){
+NdbDictInterface::drop_filegroup(const NdbFilegroupImpl & group)
+{
   DBUG_ENTER("NdbDictInterface::drop_filegroup");
   NdbApiSignal tSignal(m_reference);
   tSignal.theReceiversBlockNumber = DBDICT;

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -52,6 +52,8 @@ protected:
     m_status(NdbDictionary::Object::New) {
     m_id = -1;
   }
+  
+  friend class NdbDictionary::ObjectId;
 };
 
 /**
@@ -468,9 +470,10 @@ public:
   static int parseFilegroupInfo(NdbFilegroupImpl &dst,
 				const Uint32 * data, Uint32 len);
   
-  int create_file(const NdbFileImpl &, const NdbFilegroupImpl&, bool overwrite = false);
+  int create_file(const NdbFileImpl &, const NdbFilegroupImpl&, 
+		  bool overwrite, NdbDictObjectImpl*);
   int drop_file(const NdbFileImpl &);
-  int create_filegroup(const NdbFilegroupImpl &);
+  int create_filegroup(const NdbFilegroupImpl &, NdbDictObjectImpl*);
   int drop_filegroup(const NdbFilegroupImpl &);
   
   int get_filegroup(NdbFilegroupImpl&, NdbDictionary::Object::Type, Uint32);
@@ -622,15 +625,15 @@ public:
   NdbEventImpl * getBlobEvent(const NdbEventImpl& ev, uint col_no);
   NdbEventImpl * getEventImpl(const char * internalName);
 
-  int createDatafile(const NdbDatafileImpl &, bool force = false);
+  int createDatafile(const NdbDatafileImpl &, bool force, NdbDictObjectImpl*);
   int dropDatafile(const NdbDatafileImpl &);
-  int createUndofile(const NdbUndofileImpl &, bool force = false);
+  int createUndofile(const NdbUndofileImpl &, bool force, NdbDictObjectImpl*);
   int dropUndofile(const NdbUndofileImpl &);
 
-  int createTablespace(const NdbTablespaceImpl &);
+  int createTablespace(const NdbTablespaceImpl &, NdbDictObjectImpl*);
   int dropTablespace(const NdbTablespaceImpl &);
 
-  int createLogfileGroup(const NdbLogfileGroupImpl &);
+  int createLogfileGroup(const NdbLogfileGroupImpl &, NdbDictObjectImpl*);
   int dropLogfileGroup(const NdbLogfileGroupImpl &);
   
   const NdbError & getNdbError() const;