ndb - force var part, part III

storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Fix forceVarPart also for ordered indexes
storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  Add new table flag
storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp:
  set null ref in case of force var part
storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp:
  add varref also for force var part
storage/ndb/test/src/NDBT_Tables.cpp:
  "randomly" set forcevart on NDBT tables

storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -1365,6 +1365,7 @@ Dblqh::sendAddFragReq(Signal* signal)
       tupFragReq->noOfCharsets = addfragptr.p->noOfCharsets;
       tupFragReq->checksumIndicator = addfragptr.p->checksumIndicator;
       tupFragReq->globalCheckpointIdIndicator = addfragptr.p->GCPIndicator;
+      tupFragReq->forceVarPartFlag = addfragptr.p->forceVarPartFlag;      
       sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ,
 		 signal, TupFragReq::SignalLength, JBB);
       return;

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

@@ -899,7 +899,8 @@ ArrayPool<TupTriggerData> c_triggerPool;
     enum Bits
     {
       TR_Checksum = 0x1, // Need to be 1
-      TR_RowGCI   = 0x2
+      TR_RowGCI   = 0x2,
+      TR_ForceVarPart = 0x4
     };
     Uint16 m_bits;
     Uint16 total_rec_size; // Max total size for entire tuple in words

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

@@ -1102,7 +1102,14 @@ Dbtup::prepare_initial_insert(KeyReqStruct *req_struct,
   const Uint32 cnt1= regTabPtr->m_attributes[MM].m_no_of_varsize;
   const Uint32 cnt2= regTabPtr->m_attributes[DD].m_no_of_varsize;
   Uint32 *ptr= req_struct->m_tuple_ptr->get_end_of_fix_part_ptr(regTabPtr);
+  Var_part_ref* ref = req_struct->m_tuple_ptr->get_var_part_ref_ptr(regTabPtr);
 
+  if (regTabPtr->m_bits & Tablerec::TR_ForceVarPart)
+  {
+    ref->m_page_no = RNIL;
+    ref->m_page_idx = Tup_varsize_page::END_OF_FREE_LIST;
+  }
+  
   if(cnt1)
   {
     KeyReqStruct::Var_data* dst= &req_struct->m_var_data[MM];

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

@@ -61,6 +61,7 @@ void Dbtup::execTUPFRAGREQ(Signal* signal)
   Uint32 checksumIndicator = tupFragReq->checksumIndicator;
   Uint32 gcpIndicator = tupFragReq->globalCheckpointIdIndicator;
   Uint32 tablespace_id= tupFragReq->tablespaceid;
+  Uint32 forceVarPart = tupFragReq->forceVarPartFlag;
 
   Uint64 maxRows =
     (((Uint64)tupFragReq->maxRowsHigh) << 32) + tupFragReq->maxRowsLow;
@@ -175,6 +176,7 @@ void Dbtup::execTUPFRAGREQ(Signal* signal)
     regTabPtr.p->m_bits = 0;
     regTabPtr.p->m_bits |= (checksumIndicator ? Tablerec::TR_Checksum : 0);
     regTabPtr.p->m_bits |= (gcpIndicator ? Tablerec::TR_RowGCI : 0);
+    regTabPtr.p->m_bits |= (forceVarPart ? Tablerec::TR_ForceVarPart : 0);
     
     regTabPtr.p->m_offsets[MM].m_disk_ref_offset= 0;
     regTabPtr.p->m_offsets[MM].m_null_words= 0;
@@ -477,6 +479,11 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
   }
 
   if (regTabPtr.p->m_attributes[MM].m_no_of_varsize)
+  {
+    pos[MM] += Var_part_ref::SZ32;
+    regTabPtr.p->m_bits &= ~(Uint32)Tablerec::TR_ForceVarPart;
+  }
+  else if (regTabPtr.p->m_bits & Tablerec::TR_ForceVarPart)
   {
     pos[MM] += Var_part_ref::SZ32;
   }

storage/ndb/test/src/NDBT_Tables.cpp

@@ -971,11 +971,19 @@ NDBT_Tables::createTable(Ndb* pNdb, const char* _name, bool _temp,
       "that NDBT_Tables can create!" << endl;
     return NDBT_WRONGARGS;
   }
+
+  Uint32 sum = 0;
+  for (Uint32 i = 0; i<strlen(_name); i++)
+    sum += 33 * sum + (Uint32)_name[i];
+  
+  bool forceVarPart = (sum & 1);
   
   int r = 0;
   do {
     NdbDictionary::Table tmpTab(* tab);
     tmpTab.setStoredTable(_temp ? 0 : 1);
+    tmpTab.setForceVarPart(forceVarPart);
+
     {
       NdbError error;
       int ret = tmpTab.validate(error);