ndb -

clean up varsize
by putting varpart ref "inside" fixpart instead of "optionally" at the end
parent 7b215a57
......@@ -940,18 +940,12 @@ ArrayPool<TupTriggerData> c_triggerPool;
Uint8 m_null_words;
Uint8 m_null_offset;
Uint16 m_disk_ref_offset; // In words relative m_data
union {
Uint16 m_varpart_offset; // In words relative m_data
Uint16 m_fix_header_size; // For fix size tuples= total rec size(part)
};
Uint16 m_max_var_offset; // In bytes relative m_var_data.m_data_ptr
} m_offsets[2];
Uint32 get_check_offset(Uint32 mm) const {
Uint32 cnt= m_attributes[mm].m_no_of_varsize;
Uint32 off= m_offsets[mm].m_varpart_offset;
return off - (cnt ? 0 : Tuple_header::HeaderSize);
return m_offsets[mm].m_fix_header_size;
}
struct {
......@@ -1240,6 +1234,11 @@ typedef Ptr<HostBuffer> HostBufferPtr;
#endif
};
struct Disk_part_ref
{
STATIC_CONST( SZ32 = 2 );
};
struct Tuple_header
{
union {
......@@ -1288,12 +1287,22 @@ typedef Ptr<HostBuffer> HostBufferPtr;
return m_null_bits+tabPtrP->m_offsets[mm].m_null_offset;
}
Uint32* get_var_part_ptr(const Tablerec* tabPtrP) {
return m_data + tabPtrP->m_offsets[MM].m_varpart_offset;
Var_part_ref* get_var_part_ref_ptr(const Tablerec* tabPtrP) {
return (Var_part_ref*)(get_disk_ref_ptr(tabPtrP) + Disk_part_ref::SZ32);
}
const Var_part_ref* get_var_part_ref_ptr(const Tablerec* tabPtrP) const {
return (Var_part_ref*)(get_disk_ref_ptr(tabPtrP) + Disk_part_ref::SZ32);
}
Uint32* get_end_of_fix_part_ptr(const Tablerec* tabPtrP) {
return m_data + tabPtrP->m_offsets[MM].m_fix_header_size -
Tuple_header::HeaderSize;
}
const Uint32* get_var_part_ptr(const Tablerec* tabPtrP) const {
return m_data + tabPtrP->m_offsets[MM].m_varpart_offset;
const Uint32* get_end_of_fix_part_ptr(const Tablerec* tabPtrP) const {
return m_data + tabPtrP->m_offsets[MM].m_fix_header_size -
Tuple_header::HeaderSize;
}
Uint32* get_disk_ref_ptr(const Tablerec* tabPtrP) {
......
......@@ -153,8 +153,7 @@ void Dbtup::do_tup_abortreq(Signal* signal, Uint32 flags)
ndbassert(tuple_ptr->m_header_bits & Tuple_header::CHAINED_ROW);
Var_part_ref *ref =
(Var_part_ref*)tuple_ptr->get_var_part_ptr(regTabPtr.p);
Var_part_ref *ref = tuple_ptr->get_var_part_ref_ptr(regTabPtr.p);
Local_key tmp;
ref->copyout(&tmp);
......
......@@ -235,16 +235,22 @@ Dbtup::commit_operation(Signal* signal,
}
else
{
Var_part_ref *ref= (Var_part_ref*)tuple_ptr->get_var_part_ptr(regTabPtr);
memcpy(tuple_ptr, copy, 4*(Tuple_header::HeaderSize+fixsize));
/**
* Var_part_ref is only stored in *allocated* tuple
* so memcpy from copy, will over write it...
* hence subtle copyout/assign...
*/
Local_key tmp;
Var_part_ref *ref= tuple_ptr->get_var_part_ref_ptr(regTabPtr);
ref->copyout(&tmp);
memcpy(tuple_ptr, copy, 4*fixsize);
ref->assign(&tmp);
PagePtr vpagePtr;
Uint32 *dst= get_ptr(&vpagePtr, *ref);
Var_page* vpagePtrP = (Var_page*)vpagePtr.p;
Uint32 *src= copy->get_var_part_ptr(regTabPtr);
Uint32 *src= copy->get_end_of_fix_part_ptr(regTabPtr);
Uint32 sz= ((mm_vars + 1) << 1) + (((Uint16*)src)[mm_vars]);
ndbassert(4*vpagePtrP->get_entry_len(tmp.m_page_idx) >= sz);
memcpy(dst, src, sz);
......@@ -257,8 +263,7 @@ Dbtup::commit_operation(Signal* signal,
update_free_page_list(regFragPtr, vpagePtr);
}
disk_ptr = (Tuple_header*)
(((Uint32*)copy)+Tuple_header::HeaderSize+fixsize+((sz + 3) >> 2));
disk_ptr = (Tuple_header*)(((Uint32*)copy)+fixsize+((sz + 3) >> 2));
}
if (regTabPtr->m_no_of_disk_attributes &&
......
......@@ -368,7 +368,7 @@ operator<<(NdbOut& out, const Dbtup::Tablerec::Tuple_offsets& off)
out << "[ null_words: " << (Uint32)off.m_null_words
<< " null off: " << (Uint32)off.m_null_offset
<< " disk_off: " << off.m_disk_ref_offset
<< " var_off: " << off.m_varpart_offset
<< " fixheadsz: " << off.m_fix_header_size
<< " max_var_off: " << off.m_max_var_offset
<< " ]";
......
......@@ -221,9 +221,6 @@ Dbtup::calculateChecksum(Tuple_header* tuple_ptr,
// includes tupVersion
//printf("%p - ", tuple_ptr);
if (regTabPtr->m_attributes[MM].m_no_of_varsize)
rec_size += Tuple_header::HeaderSize;
for (i= 0; i < rec_size-2; i++) {
checksum ^= tuple_header[i];
//printf("%.8x ", tuple_header[i]);
......@@ -1104,7 +1101,7 @@ 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_var_part_ptr(regTabPtr);
Uint32 *ptr= req_struct->m_tuple_ptr->get_end_of_fix_part_ptr(regTabPtr);
if(cnt1)
{
......@@ -1126,22 +1123,10 @@ Dbtup::prepare_initial_insert(KeyReqStruct *req_struct,
pos += AttributeDescriptor::getSizeInBytes(tab_descr[*order++].tabDescr);
}
}
else
{
ptr -= Tuple_header::HeaderSize;
}
req_struct->m_disk_ptr= (Tuple_header*)ptr;
if(cnt2)
{
KeyReqStruct::Var_data *dst= &req_struct->m_var_data[DD];
ptr=((Tuple_header*)ptr)->m_data+regTabPtr->m_offsets[DD].m_varpart_offset;
dst->m_data_ptr= (char*)(((Uint16*)ptr)+cnt2+1);
dst->m_offset_array_ptr= req_struct->var_pos_array + (cnt1 << 1);
dst->m_var_len_offset= cnt2;
dst->m_max_var_offset= regTabPtr->m_offsets[DD].m_max_var_offset;
}
ndbrequire(cnt2 == 0);
// Set all null bits
memset(req_struct->m_tuple_ptr->m_null_bits+
......@@ -2510,13 +2495,14 @@ Dbtup::expand_tuple(KeyReqStruct* req_struct,
Uint16 dd_tot= tabPtrP->m_no_of_disk_attributes;
Uint16 mm_vars= tabPtrP->m_attributes[MM].m_no_of_varsize;
Uint32 fix_size= tabPtrP->m_offsets[MM].m_varpart_offset;
Uint32 fix_size= tabPtrP->m_offsets[MM].m_fix_header_size;
Uint32 order_desc= tabPtrP->m_real_order_descriptor;
Uint32 *dst_ptr= ptr->get_var_part_ptr(tabPtrP);
Uint32 *dst_ptr= ptr->get_end_of_fix_part_ptr(tabPtrP);
const Uint32 *disk_ref= src->get_disk_ref_ptr(tabPtrP);
const Uint32 *src_ptr= src->get_var_part_ptr(tabPtrP);
const Uint32 * desc= (Uint32*)req_struct->attr_descr;
const Uint32 *src_ptr= src->get_end_of_fix_part_ptr(tabPtrP);
const Var_part_ref* var_ref = src->get_var_part_ref_ptr(tabPtrP);
const Uint32 *desc= (Uint32*)req_struct->attr_descr;
const Uint16 *order = (Uint16*)(&tableDescriptor[order_desc]);
order += tabPtrP->m_attributes[MM].m_no_of_fixsize;
......@@ -2529,7 +2515,7 @@ Dbtup::expand_tuple(KeyReqStruct* req_struct,
if(bits & Tuple_header::CHAINED_ROW)
{
Ptr<Page> var_page;
src_data= get_ptr(&var_page, * (Var_part_ref*)src_ptr);
src_data= get_ptr(&var_page, *var_ref);
step= 4;
sizes[MM]= (2 + (mm_vars << 1) + ((Uint16*)src_data)[mm_vars] + 3) >> 2;
req_struct->m_varpart_page_ptr = var_page;
......@@ -2550,14 +2536,12 @@ Dbtup::expand_tuple(KeyReqStruct* req_struct,
ndbassert((UintPtr(src_ptr) & 3) == 0);
src_ptr = ALIGN_WORD(((char*)src_ptr)+step);
sizes[MM] += fix_size + Tuple_header::HeaderSize;
memcpy(ptr, src, 4*(fix_size + Tuple_header::HeaderSize));
sizes[MM] += fix_size;
memcpy(ptr, src, 4*fix_size);
}
else
{
sizes[MM]= 1;
dst_ptr -= Tuple_header::HeaderSize;
src_ptr -= Tuple_header::HeaderSize;
memcpy(ptr, src, 4*fix_size);
}
......@@ -2591,20 +2575,7 @@ Dbtup::expand_tuple(KeyReqStruct* req_struct,
ndbassert(! (req_struct->m_disk_ptr->m_header_bits & Tuple_header::FREE));
if(dd_vars)
{
KeyReqStruct::Var_data* dst= &req_struct->m_var_data[DD];
dst_ptr += tabPtrP->m_offsets[DD].m_varpart_offset;
src_ptr += tabPtrP->m_offsets[DD].m_varpart_offset;
order += tabPtrP->m_attributes[DD].m_no_of_fixsize;
dst->m_data_ptr= (char*)(char*)(((Uint16*)dst_ptr)+dd_vars+1);
dst->m_offset_array_ptr= req_struct->var_pos_array + (mm_vars << 1);
dst->m_var_len_offset= dd_vars;
dst->m_max_var_offset= tabPtrP->m_offsets[DD].m_max_var_offset;
expand_var_part(dst, src_ptr, desc, order);
}
ndbrequire(dd_vars == 0);
}
ptr->m_header_bits= (bits & ~(Uint32)(Tuple_header::CHAINED_ROW));
......@@ -2620,9 +2591,9 @@ Dbtup::prepare_read(KeyReqStruct* req_struct,
Uint16 dd_tot= tabPtrP->m_no_of_disk_attributes;
Uint16 mm_vars= tabPtrP->m_attributes[MM].m_no_of_varsize;
const Uint32 *src_ptr= ptr->get_var_part_ptr(tabPtrP);
const Uint32 *src_ptr= ptr->get_end_of_fix_part_ptr(tabPtrP);
const Uint32 *disk_ref= ptr->get_disk_ref_ptr(tabPtrP);
const Var_part_ref* var_ref = ptr->get_var_part_ref_ptr(tabPtrP);
if(mm_vars)
{
const Uint32 *src_data= src_ptr;
......@@ -2632,7 +2603,7 @@ Dbtup::prepare_read(KeyReqStruct* req_struct,
#if VM_TRACE
#endif
src_data= get_ptr(* (Var_part_ref*)src_ptr);
src_data= get_ptr(* var_ref);
}
dst->m_data_ptr= (char*)(((Uint16*)src_data)+mm_vars+1);
dst->m_offset_array_ptr= (Uint16*)src_data;
......@@ -2642,11 +2613,6 @@ Dbtup::prepare_read(KeyReqStruct* req_struct,
// disk part start after varsize (aligned)
src_ptr = ALIGN_WORD(dst->m_data_ptr + dst->m_max_var_offset);
}
else
{
// disk part if after fixsize part...
src_ptr -= Tuple_header::HeaderSize;
}
if(disk && dd_tot)
{
......@@ -2668,16 +2634,7 @@ Dbtup::prepare_read(KeyReqStruct* req_struct,
// Fix diskpart
req_struct->m_disk_ptr= (Tuple_header*)src_ptr;
ndbassert(! (req_struct->m_disk_ptr->m_header_bits & Tuple_header::FREE));
if(dd_vars)
{
KeyReqStruct::Var_data* dst= &req_struct->m_var_data[DD];
src_ptr += tabPtrP->m_offsets[DD].m_varpart_offset;
dst->m_data_ptr= (char*)(char*)(((Uint16*)src_ptr)+dd_vars+1);
dst->m_offset_array_ptr= (Uint16*)src_ptr;
dst->m_var_len_offset= 1;
dst->m_max_var_offset= ((Uint16*)src_ptr)[dd_vars];
}
ndbrequire(dd_vars == 0);
}
}
......@@ -2692,10 +2649,11 @@ Dbtup::shrink_tuple(KeyReqStruct* req_struct, Uint32 sizes[2],
Uint16 mm_vars= tabPtrP->m_attributes[MM].m_no_of_varsize;
Uint16 dd_vars= tabPtrP->m_attributes[DD].m_no_of_varsize;
Uint32 *dst_ptr= ptr->get_var_part_ptr(tabPtrP);
Uint32 *dst_ptr= ptr->get_end_of_fix_part_ptr(tabPtrP);
Uint16* src_off_ptr= req_struct->var_pos_array;
sizes[MM]= sizes[DD]= 0;
sizes[MM] = 1;
sizes[DD] = 0;
if(mm_vars)
{
Uint16* dst_off_ptr= (Uint16*)dst_ptr;
......@@ -2719,26 +2677,15 @@ Dbtup::shrink_tuple(KeyReqStruct* req_struct, Uint32 sizes[2],
dst_ptr = ALIGN_WORD(dst_data_ptr);
}
else
{
sizes[MM] = 1;
dst_ptr -= Tuple_header::HeaderSize;
}
if(disk && dd_tot)
{
Uint32 * src_ptr = (Uint32*)req_struct->m_disk_ptr;
req_struct->m_disk_ptr = (Tuple_header*)dst_ptr;
if (unlikely(dd_vars))
{
abort();
}
else
{
ndbrequire(dd_vars == 0);
sizes[DD] = tabPtrP->m_offsets[DD].m_fix_header_size;
memmove(dst_ptr, src_ptr, 4*tabPtrP->m_offsets[DD].m_fix_header_size);
}
}
}
void
......@@ -2771,7 +2718,7 @@ Dbtup::validate_page(Tablerec* regTabPtr, Var_page* p)
if(!(idx & Var_page::FREE) && !(idx & Var_page::CHAIN))
{
Tuple_header *ptr= (Tuple_header*)page->get_ptr(i);
Uint32 *part= ptr->get_var_part_ptr(regTabPtr);
Uint32 *part= ptr->get_end_of_fix_part_ptr(regTabPtr);
if(ptr->m_header_bits & Tuple_header::CHAINED_ROW)
{
ndbassert(len == fix_sz + 1);
......@@ -2835,8 +2782,7 @@ Dbtup::handle_size_change_after_update(KeyReqStruct* req_struct,
Uint32 bits= org->m_header_bits;
Uint32 copy_bits= req_struct->m_tuple_ptr->m_header_bits;
Uint32 fix_sz = Tuple_header::HeaderSize +
regTabPtr->m_offsets[MM].m_fix_header_size;
Uint32 fix_sz = regTabPtr->m_offsets[MM].m_fix_header_size;
if(sizes[MM] == sizes[2+MM])
;
......@@ -2851,7 +2797,7 @@ Dbtup::handle_size_change_after_update(KeyReqStruct* req_struct,
Ptr<Page> pagePtr = req_struct->m_varpart_page_ptr;
Var_page* pageP= (Var_page*)pagePtr.p;
Uint32 idx, alloc, needed;
Var_part_ref *refptr = (Var_part_ref*)org->get_var_part_ptr(regTabPtr);
Var_part_ref *refptr = org->get_var_part_ref_ptr(regTabPtr);
ndbassert(bits & Tuple_header::CHAINED_ROW);
Local_key ref;
......@@ -2900,19 +2846,7 @@ Dbtup::nr_update_gci(Uint32 fragPtrI, const Local_key* key, Uint32 gci)
Local_key tmp = *key;
PagePtr page_ptr;
int ret;
if (tablePtr.p->m_attributes[MM].m_no_of_varsize)
{
tablePtr.p->m_offsets[MM].m_fix_header_size +=
Tuple_header::HeaderSize+1;
ret = alloc_page(tablePtr.p, fragPtr.p, &page_ptr, tmp.m_page_no);
tablePtr.p->m_offsets[MM].m_fix_header_size -=
Tuple_header::HeaderSize+1;
}
else
{
ret = alloc_page(tablePtr.p, fragPtr.p, &page_ptr, tmp.m_page_no);
}
int ret = alloc_page(tablePtr.p, fragPtr.p, &page_ptr, tmp.m_page_no);
if (ret)
return -1;
......@@ -2941,19 +2875,9 @@ Dbtup::nr_read_pk(Uint32 fragPtrI,
Local_key tmp = *key;
Uint32 pages = fragPtr.p->noOfPages;
int ret;
PagePtr page_ptr;
if (tablePtr.p->m_attributes[MM].m_no_of_varsize)
{
const Uint32 XXX = Tuple_header::HeaderSize+Var_part_ref::SZ32;
tablePtr.p->m_offsets[MM].m_fix_header_size += XXX;
ret = alloc_page(tablePtr.p, fragPtr.p, &page_ptr, tmp.m_page_no);
tablePtr.p->m_offsets[MM].m_fix_header_size -= XXX;
}
else
{
ret = alloc_page(tablePtr.p, fragPtr.p, &page_ptr, tmp.m_page_no);
}
int ret = alloc_page(tablePtr.p, fragPtr.p, &page_ptr, tmp.m_page_no);
if (ret)
return -1;
......
......@@ -478,8 +478,7 @@ Dbtup::buildIndex(Signal* signal, Uint32 buildPtrI)
ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
const Uint32 firstTupleNo = 0;
const Uint32 tupheadsize = tablePtr.p->m_offsets[MM].m_fix_header_size +
(buildPtr.p->m_build_vs? Tuple_header::HeaderSize + Var_part_ref::SZ32: 0);
const Uint32 tupheadsize = tablePtr.p->m_offsets[MM].m_fix_header_size;
#ifdef TIME_MEASUREMENT
MicroSecondTimer start;
......
......@@ -178,12 +178,12 @@ void Dbtup::execTUPFRAGREQ(Signal* signal)
regTabPtr.p->m_offsets[MM].m_disk_ref_offset= 0;
regTabPtr.p->m_offsets[MM].m_null_words= 0;
regTabPtr.p->m_offsets[MM].m_varpart_offset= 0;
regTabPtr.p->m_offsets[MM].m_fix_header_size= 0;
regTabPtr.p->m_offsets[MM].m_max_var_offset= 0;
regTabPtr.p->m_offsets[DD].m_disk_ref_offset= 0;
regTabPtr.p->m_offsets[DD].m_null_words= 0;
regTabPtr.p->m_offsets[DD].m_varpart_offset= 0;
regTabPtr.p->m_offsets[DD].m_fix_header_size= 0;
regTabPtr.p->m_offsets[DD].m_max_var_offset= 0;
regTabPtr.p->m_attributes[MM].m_no_of_fixsize= 0;
......@@ -466,7 +466,19 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
if(regTabPtr.p->m_no_of_disk_attributes > 0)
{
regTabPtr.p->m_offsets[MM].m_disk_ref_offset= pos[MM];
pos[MM] += 2; // 8 bytes
pos[MM] += Disk_part_ref::SZ32; // 8 bytes
}
else
{
/**
* var part ref is stored at m_disk_ref_offset + Disk_part_ref::SZ32
*/
regTabPtr.p->m_offsets[MM].m_disk_ref_offset= pos[MM]-Disk_part_ref::SZ32;
}
if (regTabPtr.p->m_attributes[MM].m_no_of_varsize)
{
pos[MM] += Var_part_ref::SZ32;
}
regTabPtr.p->m_offsets[MM].m_null_offset= pos[MM];
......@@ -492,6 +504,7 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
}
regTabPtr.p->m_offsets[MM].m_fix_header_size=
Tuple_header::HeaderSize +
fragOperPtr.p->m_fix_attributes_size[MM] +
pos[MM];
......@@ -499,9 +512,6 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
fragOperPtr.p->m_fix_attributes_size[DD] +
pos[DD];
if(regTabPtr.p->m_attributes[MM].m_no_of_varsize == 0)
regTabPtr.p->m_offsets[MM].m_fix_header_size += Tuple_header::HeaderSize;
if(regTabPtr.p->m_attributes[DD].m_no_of_varsize == 0 &&
regTabPtr.p->m_attributes[DD].m_no_of_fixsize > 0)
regTabPtr.p->m_offsets[DD].m_fix_header_size += Tuple_header::HeaderSize;
......@@ -541,8 +551,6 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
{
Uint32 fix_tupheader = regTabPtr.p->m_offsets[MM].m_fix_header_size;
if(regTabPtr.p->m_attributes[MM].m_no_of_varsize != 0)
fix_tupheader += Tuple_header::HeaderSize + Var_part_ref::SZ32;
ndbassert(fix_tupheader > 0);
Uint32 noRowsPerPage = ZWORDS_ON_PAGE / fix_tupheader;
Uint32 noAllocatedPages =
......
......@@ -599,8 +599,7 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
const bool lcp = (bits & ScanOp::SCAN_LCP);
Uint32 lcp_list = fragPtr.p->m_lcp_keep_list;
Uint32 size = table.m_offsets[mm].m_fix_header_size +
(bits&ScanOp::SCAN_VS ? Tuple_header::HeaderSize + Var_part_ref::SZ32 : 0);
Uint32 size = table.m_offsets[mm].m_fix_header_size;
if (lcp && lcp_list != RNIL)
goto found_lcp_keep;
......
......@@ -71,27 +71,22 @@ Uint32* Dbtup::alloc_var_rec(Fragrecord* fragPtr,
/**
* TODO alloc fix+var part
*/
const Uint32 XXX = Tuple_header::HeaderSize + Var_part_ref::SZ32;
tabPtr->m_offsets[MM].m_fix_header_size += XXX;
Uint32 *ptr = alloc_fix_rec(fragPtr, tabPtr, key, out_frag_page_id);
tabPtr->m_offsets[MM].m_fix_header_size -= XXX;
if (unlikely(ptr == 0))
{
return 0;
}
ndbassert(alloc_size >= tabPtr->m_offsets[MM].m_fix_header_size +
Tuple_header::HeaderSize);
ndbassert(alloc_size >= tabPtr->m_offsets[MM].m_fix_header_size);
alloc_size -= tabPtr->m_offsets[MM].m_fix_header_size +
Tuple_header::HeaderSize;
alloc_size -= tabPtr->m_offsets[MM].m_fix_header_size;
Local_key varref;
if (likely(alloc_var_part(fragPtr, tabPtr, alloc_size, &varref) != 0))
{
Tuple_header* tuple = (Tuple_header*)ptr;
Var_part_ref* dst = (Var_part_ref*)tuple->get_var_part_ptr(tabPtr);
Var_part_ref* dst = tuple->get_var_part_ref_ptr(tabPtr);
dst->assign(&varref);
return ptr;
}
......@@ -170,7 +165,7 @@ void Dbtup::free_var_rec(Fragrecord* fragPtr,
Tuple_header* tuple = (Tuple_header*)ptr;
Local_key ref;
Var_part_ref * varref = (Var_part_ref*)tuple->get_var_part_ptr(tabPtr);
Var_part_ref * varref = tuple->get_var_part_ref_ptr(tabPtr);
varref->copyout(&ref);
free_fix_rec(fragPtr, tabPtr, key, (Fix_page*)pagePtr.p);
......@@ -402,26 +397,21 @@ Dbtup::alloc_var_rowid(Fragrecord* fragPtr,
Local_key* key,
Uint32 * out_frag_page_id)
{
const Uint32 XXX = Tuple_header::HeaderSize + Var_part_ref::SZ32;
tabPtr->m_offsets[MM].m_fix_header_size += XXX;
Uint32 *ptr = alloc_fix_rowid(fragPtr, tabPtr, key, out_frag_page_id);
tabPtr->m_offsets[MM].m_fix_header_size -= XXX;
if (unlikely(ptr == 0))
{
return 0;
}
ndbassert(alloc_size >= tabPtr->m_offsets[MM].m_fix_header_size +
Tuple_header::HeaderSize);
ndbassert(alloc_size >= tabPtr->m_offsets[MM].m_fix_header_size);
alloc_size -= tabPtr->m_offsets[MM].m_fix_header_size +
Tuple_header::HeaderSize;
alloc_size -= tabPtr->m_offsets[MM].m_fix_header_size;
Local_key varref;
if (likely(alloc_var_part(fragPtr, tabPtr, alloc_size, &varref) != 0))
{
Tuple_header* tuple = (Tuple_header*)ptr;
Var_part_ref* dst = (Var_part_ref*)tuple->get_var_part_ptr(tabPtr);
Var_part_ref* dst = (Var_part_ref*)tuple->get_var_part_ref_ptr(tabPtr);
dst->assign(&varref);
return ptr;
}
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment