ndb - backport from 5.1-ndb-dd: AttributeHeader: store size in bytes

storage/ndb/include/kernel/AttributeHeader.hpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/dbutil/DbUtil.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/kernel/blocks/trix/Trix.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/ndbapi/NdbOperationDefine.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/ndbapi/NdbOperationSearch.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes
storage/ndb/src/ndbapi/NdbScanOperation.cpp:
  backport from 5.1-ndb-dd: AttributeHeader: store size in bytes

storage/ndb/include/kernel/AttributeHeader.hpp

@@ -43,9 +43,11 @@ public:
 
   STATIC_CONST( RECORDS_IN_RANGE = 0xFFF8 );
 
+  // NOTE: in 5.1 ctors and init take size in bytes
+
   /** Initialize AttributeHeader at location aHeaderPtr */
   static AttributeHeader& init(void* aHeaderPtr, Uint32 anAttributeId, 
-			       Uint32 aDataSize);
+			       Uint32 aByteSize);
 
   /** Returns size of AttributeHeader (usually one or two words) */
   Uint32 getHeaderSize() const; // In 32-bit words
@@ -62,8 +64,10 @@ public:
   /** Getters and Setters */
   Uint32  getAttributeId() const;
   void    setAttributeId(Uint32);
-  Uint32  getDataSize() const;   // In 32-bit words
-  void    setDataSize(Uint32);
+  Uint32  getByteSize() const;
+  void    setByteSize(Uint32);
+  Uint32  getDataSize() const;   // In 32-bit words, rounded up
+  void    setDataSize(Uint32);   // Set size to multiple of word size
   bool    isNULL() const;
   void    setNULL();
 
@@ -71,11 +75,12 @@ public:
   //void    print(NdbOut&);
   void    print(FILE*);
 
+  static Uint32 getByteSize(Uint32);
   static Uint32 getDataSize(Uint32);
   
 public:
   AttributeHeader(Uint32 = 0);
-  AttributeHeader(Uint32 anAttributeId, Uint32 aDataSize);
+  AttributeHeader(Uint32 anAttributeId, Uint32 aByteSize);
   ~AttributeHeader();
   
   Uint32 m_value;
@@ -84,12 +89,13 @@ public:
 /**
  *           1111111111222222222233
  * 01234567890123456789012345678901
- * ssssssssssssss eiiiiiiiiiiiiiiii
+ * ssssssssssssssssiiiiiiiiiiiiiiii
  *
  * i = Attribute Id
- * s = Size of current "chunk" - 14 Bits -> 16384 (words) = 65k
- *     Including optional extra word(s).
- * e - Element data/Blob, read element of array
+ * s = Size of current "chunk" in bytes - 16 bits.
+ *     To allow round up to word, max value is 0xFFFC (not checked).
+ * e - [ obsolete future ]
+ *     Element data/Blob, read element of array
  *     If == 0 next data word contains attribute value.
  *     If == 1 next data word contains:
  *       For Array of Fixed size Elements
@@ -98,15 +104,13 @@ public:
  *         Start offset (32 bit) (length is defined in previous word)
  * 
  * An attribute value equal to "null" is represented by setting s == 0.
- * 
- * Bit 14 is not yet used.
  */
 
 inline
 AttributeHeader& AttributeHeader::init(void* aHeaderPtr, Uint32 anAttributeId, 
-				       Uint32 aDataSize)
+				       Uint32 aByteSize)
 {
-  return * new (aHeaderPtr) AttributeHeader(anAttributeId, aDataSize);
+  return * new (aHeaderPtr) AttributeHeader(anAttributeId, aByteSize);
 }
 
 inline
@@ -116,11 +120,11 @@ AttributeHeader::AttributeHeader(Uint32 aHeader)
 }
 
 inline
-AttributeHeader::AttributeHeader(Uint32 anAttributeId, Uint32 aDataSize)
+AttributeHeader::AttributeHeader(Uint32 anAttributeId, Uint32 aByteSize)
 {
   m_value = 0;
   this->setAttributeId(anAttributeId);
-  this->setDataSize(aDataSize);
+  this->setByteSize(aByteSize);
 }
 
 inline
@@ -147,17 +151,30 @@ void AttributeHeader::setAttributeId(Uint32 anAttributeId)
   m_value |= (anAttributeId << 16);
 }
 
+inline
+Uint32 AttributeHeader::getByteSize() const
+{
+  return (m_value & 0xFFFF);
+}
+
+inline
+void AttributeHeader::setByteSize(Uint32 aByteSize)
+{
+  m_value &= (~0xFFFF);
+  m_value |= aByteSize;
+}
+
 inline
 Uint32 AttributeHeader::getDataSize() const
 {
-  return (m_value & 0x3FFF);
+  return (((m_value & 0xFFFF) + 3) >> 2);
 }
 
 inline
 void AttributeHeader::setDataSize(Uint32 aDataSize)
 {
-  m_value &= (~0x3FFF);
-  m_value |= aDataSize;
+  m_value &= (~0xFFFF);
+  m_value |= (aDataSize << 2);
 }
 
 inline
@@ -201,10 +218,16 @@ AttributeHeader::print(FILE* output) {
 	  isNULL());
 }
 
+inline
+Uint32
+AttributeHeader::getByteSize(Uint32 m_value){
+  return (m_value & 0xFFFF);  
+}
+
 inline
 Uint32
 AttributeHeader::getDataSize(Uint32 m_value){
-  return (m_value & 0x3FFF);  
+  return (((m_value & 0xFFFF) + 3) >> 2);
 }
 
 #endif

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

@@ -8151,7 +8151,7 @@ void Dbdict::executeTransEventSysTable(Callback *pcallback, Signal *signal,
   Uint32 id=0;
   // attribute 0 event name: Primary Key
   {
-    AttributeHeader::init(attrPtr, id, sysTab_NDBEVENTS_0_szs[id]/4);
+    AttributeHeader::init(attrPtr, id, sysTab_NDBEVENTS_0_szs[id]);
     total_len += sysTab_NDBEVENTS_0_szs[id];
     attrPtr++; id++;
   }
@@ -8169,7 +8169,7 @@ void Dbdict::executeTransEventSysTable(Callback *pcallback, Signal *signal,
     jam();
     EVENT_TRACE;
     while ( id < noAttr ) {
-      AttributeHeader::init(attrPtr, id, sysTab_NDBEVENTS_0_szs[id]/4);
+      AttributeHeader::init(attrPtr, id, sysTab_NDBEVENTS_0_szs[id]);
       total_len += sysTab_NDBEVENTS_0_szs[id];
       attrPtr++; id++;
     }

storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -12511,7 +12511,7 @@ void Dbtc::insertIntoIndexTable(Signal* signal,
     hops = attrHeader->getHeaderSize() + attrHeader->getDataSize();
     moreAttrData = keyValues.next(iter, hops);
   }
-  AttributeHeader pkAttrHeader(attrId, totalPrimaryKeyLength);
+  AttributeHeader pkAttrHeader(attrId, totalPrimaryKeyLength << 2);
   Uint32 attributesLength = afterValues.getSize() + 
     pkAttrHeader.getHeaderSize() + pkAttrHeader.getDataSize();
   

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

@@ -1548,7 +1548,8 @@ int Dbtup::interpreterNextLab(Signal* signal,
 	  Uint32 Tlen;
 
 	  AttributeHeader& ah = AttributeHeader::init(&TdataForUpdate[0], 
-						      TattrId, TattrNoOfWords);
+						      TattrId,
+                                                      TattrNoOfWords << 2);
 	  TdataForUpdate[1] = TregMemBuffer[theRegister + 2];
 	  TdataForUpdate[2] = TregMemBuffer[theRegister + 3];
 	  Tlen = TattrNoOfWords + 1;

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

@@ -213,7 +213,7 @@ int Dbtup::readAttributes(Page* const pagePtr,
     } else if(attributeId & AttributeHeader::PSEUDO){
       Uint32 sz = read_pseudo(attributeId, 
 			      outBuffer+tmpAttrBufIndex+1);
-      AttributeHeader::init(&outBuffer[tmpAttrBufIndex], attributeId, sz);
+      AttributeHeader::init(&outBuffer[tmpAttrBufIndex], attributeId, sz << 2);
       tOutBufIndex = tmpAttrBufIndex + 1 + sz;
     } else {
       terrorCode = ZATTRIBUTE_ID_ERROR;

storage/ndb/src/kernel/blocks/dbutil/DbUtil.cpp

@@ -1207,7 +1207,7 @@ DbUtil::prepareOperation(Signal* signal, PreparePtr prepPtr)
       if (operationType == UtilPrepareReq::Read) {
 	AttributeHeader::init(rsInfoIt.data, 
 			      attrDesc.AttributeId,    // 1. Store AttrId
-			      len);
+			      len << 2);
 	prepOpPtr.p->rsInfo.next(rsInfoIt, 1);
       }
     }
@@ -1394,7 +1394,7 @@ DbUtil::hardcodedPrepare() {
     ndbrequire(ptr.p->rsInfo.seize(1));
     ResultSetInfoBuffer::DataBufferIterator it; 
     ptr.p->rsInfo.first(it);
-    AttributeHeader::init(it.data, 1, 2); // Attribute 1 - 2 data words
+    AttributeHeader::init(it.data, 1, 2 << 2); // Attribute 1 - 2 data words
   }
 
   /**
@@ -1449,7 +1449,7 @@ DbUtil::hardcodedPrepare() {
       ndbrequire(ptr.p->rsInfo.seize(1));
       ResultSetInfoBuffer::DataBufferIterator it; 
       ptr.p->rsInfo.first(it);
-      AttributeHeader::init(it.data, 1, 2); // Attribute 1 - 2 data words
+      AttributeHeader::init(it.data, 1, 2 << 2); // Attribute 1 - 2 data words
     }
   }
 
@@ -1529,13 +1529,13 @@ DbUtil::execUTIL_SEQUENCE_REQ(Signal* signal){
     AttrInfoBuffer::DataBufferIterator it;   
 
     opPtr.p->attrInfo.first(it);
-    AttributeHeader::init(it.data, 0, 1);
+    AttributeHeader::init(it.data, 0, 1 << 2);
 
     ndbrequire(opPtr.p->attrInfo.next(it));
     * it.data = transPtr.p->sequence.sequenceId;
 
     ndbrequire(opPtr.p->attrInfo.next(it));
-    AttributeHeader::init(it.data, 1, 2);
+    AttributeHeader::init(it.data, 1, 2 << 2);
     
     ndbrequire(opPtr.p->attrInfo.next(it));
     * it.data = 0;

storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp

@@ -1709,9 +1709,9 @@ void Ndbcntr::crSystab7Lab(Signal* signal)
 
     tKeyDataPtr[0]               = tkey;
 
-    AttributeHeader::init(&tAIDataPtr[0], 0, 1);
+    AttributeHeader::init(&tAIDataPtr[0], 0, 1 << 2);
     tAIDataPtr[1]                = tkey;
-    AttributeHeader::init(&tAIDataPtr[2], 1, 2);
+    AttributeHeader::init(&tAIDataPtr[2], 1, 2 << 2);
     tAIDataPtr[3]                = (tkey << 16);
     tAIDataPtr[4]                = 1;    
     sendSignal(DBTC_REF, GSN_TCKEYREQ, signal, 

storage/ndb/src/kernel/blocks/trix/Trix.cpp

@@ -837,7 +837,7 @@ void Trix::executeInsertTransaction(Signal* signal,
   // Pack key attributes
   AttributeHeader::init(headerBuffer + subRec->noOfIndexColumns,
 			subRec->noOfIndexColumns,
-			noOfKeyData);
+			noOfKeyData << 2);
 
   struct LinearSectionPtr sectionsPtr[UtilExecuteReq::NoOfSections];
   sectionsPtr[UtilExecuteReq::HEADER_SECTION].p = headerBuffer;

storage/ndb/src/ndbapi/NdbOperationDefine.cpp

@@ -535,7 +535,7 @@ NdbOperation::setValue( const NdbColumnImpl* tAttrInfo,
   const Uint32 totalSizeInWords = (sizeInBytes + 3)/4; // Including bits in last word
   const Uint32 sizeInWords = sizeInBytes / 4;          // Excluding bits in last word
   AttributeHeader& ah = AttributeHeader::init(&ahValue, tAttrId, 
-					      totalSizeInWords);
+					      totalSizeInWords << 2);
   insertATTRINFO( ahValue );
 
   /***********************************************************************

storage/ndb/src/ndbapi/NdbOperationSearch.cpp

@@ -199,14 +199,14 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
       // XXX
       if(m_accessTable == m_currentTable)
       {
-	AttributeHeader::init(&ahValue, tAttrId, sz);
+	AttributeHeader::init(&ahValue, tAttrId, sz << 2);
       }
       else
       {
 	assert(m_accessTable->m_index);
 	int attr_id_current_table = 
 	  m_accessTable->m_index->m_columns[tAttrId]->m_keyInfoPos;
-	AttributeHeader::init(&ahValue, attr_id_current_table, sz);
+	AttributeHeader::init(&ahValue, attr_id_current_table, sz << 2);
       }
       
       insertATTRINFO( ahValue );

storage/ndb/src/ndbapi/NdbScanOperation.cpp

@@ -1104,7 +1104,7 @@ NdbIndexScanOperation::setBound(const NdbColumnImpl* tAttrInfo,
     // insert attribute header
     Uint32 tIndexAttrId = tAttrInfo->m_attrId;
     Uint32 sizeInWords = (len + 3) / 4;
-    AttributeHeader ah(tIndexAttrId, sizeInWords);
+    AttributeHeader ah(tIndexAttrId, sizeInWords << 2);
     const Uint32 ahValue = ah.m_value;
 
     const Uint32 align = (UintPtr(aValue) & 7);