Fixed recordNext(), add unit test for recordNext() on slotted and fixed pages.

src/stasis/page/fixed.c

@@ -75,14 +75,14 @@ static void fixedSetType(int xid, Page *p, recordid rid, int type) {
 }
 static int fixedGetLength(int xid, Page *p, recordid rid) {
   assertlocked(p->rwlatch);
-  //XXX this should be here...  assert(rid.slot < *recordcount_ptr(p));
+  assert(*page_type_ptr(p));
-  checkRid(p, rid); // <-- XXX KLUDGE checkRid init's the page if necessary...
+  return rid.slot > *recordcount_ptr(p) ?
-  return rid.slot > *recordcount_ptr(p) ? 
       INVALID_SLOT : physical_slot_length(*recordsize_ptr(p));
 }
 static recordid fixedNext(int xid, Page *p, recordid rid) {
   short n = *recordcount_ptr(p);
   rid.slot++;
+  rid.size = *recordsize_ptr(p);
   if(rid.slot >= n) {
     return NULLRID;
   } else {

src/stasis/page/indirect.c

@@ -219,7 +219,7 @@ compensated_function int indirectPageRecordCount(int xid, recordid rid) {
     int numslots = *numslots_ptr(p);
     ret = 0;
     for(int i = 0; i < numslots; i++) {
-      if(isValidSlot(p, i)) {
+      if(*slot_ptr(p,i) != INVALID_SLOT) {
 	ret++;
       }
     }

src/stasis/page/slotted.c

@@ -412,8 +412,7 @@ static void slottedSetType(int xid, Page *p, recordid rid, int type) {
 static int slottedGetLength(int xid, Page *p, recordid rid) {
   assert(p->id == rid.page);
   slottedFsck(p);
-  assert(slottedGetType(xid, p, rid) != INVALID_SLOT);
+  if( slottedGetType(xid, p, rid) == INVALID_SLOT)
-  if(rid.slot > *numslots_ptr(p) || slottedGetType(xid, p, rid) == INVALID_SLOT)
     return INVALID_SLOT;
   else
     return physical_slot_length(*slot_length_ptr(p, rid.slot));
@@ -424,9 +423,11 @@ static recordid slottedNext(int xid, Page *p, recordid rid) {
 
   short n = *numslots_ptr(p);
   rid.slot ++;
-  while(rid.slot < n && !isValidSlot(p, rid.slot)) { rid.slot++; }
+  while(rid.slot < n && slottedGetType(xid,p,rid)==INVALID_SLOT) {
-  if(isValidSlot(p, rid.slot)) {
+    rid.slot++;
-    rid.slot = *slot_length_ptr(p, rid.slot);
+  }
+  if(rid.slot != n) { 
+    rid.size = *slot_length_ptr(p, rid.slot);
     return rid;
   } else {
     return NULLRID;

src/stasis/page/slotted.h

@@ -76,7 +76,6 @@ Slotted page layout:
 #define slot_ptr(page, n)        shorts_from_end((page), (2*(n))+4)
 #define slot_length_ptr(page, n) shorts_from_end((page), (2*(n))+5)
 #define record_ptr(page, n)      bytes_from_start((page), *slot_ptr((page), (n)))
-#define isValidSlot(page, n)   ((*slot_ptr((page), (n)) == INVALID_SLOT) ? 0 : 1)
 
 void slottedPageInit();
 void slottedPageDeinit();

test/stasis/check_page.c

@@ -283,19 +283,82 @@ START_TEST(pageCheckMacros) {
   assert(*bytes_from_start(&p, 3) == 53);
   assert(*bytes_from_start(&p, 4) == 54);
 
-  assert(isValidSlot(&p, 0));
+} END_TEST
-  assert(isValidSlot(&p, 1));
-  assert(isValidSlot(&p, 40));
 
-  /*  invalidateSlot(&p, 0);
+static void assertRecordCountSizeType(int xid, Page *p, int count, int size, int type) {
-  invalidateSlot(&p, 1);
+  int foundRecords = 0;
-  invalidateSlot(&p, 40);
+
-  
+  recordid it = recordFirst(xid,p);
-  assert(!isValidSlot(&p, 0));
+  assert(it.size != INVALID_SLOT);
-  assert(!isValidSlot(&p, 1));
+  do {
-  assert(!isValidSlot(&p, 40));*/
+    foundRecords++;
+    assert(recordGetLength(xid,p,it)  == size);
+    assert(recordGetTypeNew(xid,p,it) == type);
+    it.size = 0;
+    assert(recordGetLength(xid,p,it)  == size);
+    assert(recordGetTypeNew(xid,p,it) == type);
+    it.size = INVALID_SLOT;
+    assert(recordGetLength(xid,p,it)  == size);
+    assert(recordGetTypeNew(xid,p,it) == type);
+    it = recordNext(xid,p,it);
+  } while(it.size != INVALID_SLOT);
+
+  assert(foundRecords == count);
+  assert(it.page == NULLRID.page);
+  assert(it.slot   == NULLRID.slot);
+  assert(it.size   == NULLRID.size);
+}
+
+static void checkPageIterators(int xid, Page *p,int record_count) {
+  recordid first = recordPreAlloc(xid, p, sizeof(int64_t));
+  recordPostAlloc(xid,p,first);
+
+  for(int i = 1; i < record_count; i++) {
+    recordPostAlloc(xid,p,recordPreAlloc(xid,p,sizeof(int64_t)));
+  }
+
+  assertRecordCountSizeType(xid, p, record_count, sizeof(int64_t), NORMAL_SLOT);
 
 
+  if(*page_type_ptr(p) == SLOTTED_PAGE) {
+    recordid other = first;
+    other.slot = 3;
+    recordFree(xid,p,other);
+    assertRecordCountSizeType(xid, p, record_count-1, sizeof(int64_t), NORMAL_SLOT);
+  }
+}
+/**
+   @test
+
+   Check functions used to iterate over pages
+
+   XXX this should also test indirect pages.
+*/
+START_TEST(pageRecordSizeTypeIteratorTest) {
+  Tinit();
+  int xid = Tbegin();
+  pageid_t pid = TpageAlloc(xid);
+
+  Page * p = loadPage(xid,pid);
+  writelock(p->rwlatch,0);
+  slottedPageInitialize(p);
+
+  checkPageIterators(xid,p,10);
+
+  unlock(p->rwlatch);
+
+  pid = TpageAlloc(xid);
+
+  p = loadPage(xid,pid);
+  writelock(p->rwlatch,0);
+  fixedPageInitialize(p,sizeof(int64_t),0);
+
+  checkPageIterators(xid,p,10);
+
+  unlock(p->rwlatch);
+
+  Tcommit(xid);
+  Tdeinit();
 } END_TEST
 /**
    @test 
@@ -430,6 +493,7 @@ START_TEST(pageCheckSlotTypeTest) {
 	Page * p = loadPage(-1, slot.page);
         readlock(p->rwlatch, 0);
         assert(recordGetTypeNew(xid, p, slot) == NORMAL_SLOT);
+        assert(recordGetLength(xid, p, slot) == sizeof(int));
         unlock(p->rwlatch);
 	releasePage(p);
 	
@@ -534,8 +598,8 @@ Suite * check_suite(void) {
   tcase_set_timeout(tc, 0); // disable timeouts
   /* Sub tests are added, one per line, here */
 
+  tcase_add_test(tc, pageRecordSizeTypeIteratorTest);
   tcase_add_test(tc, pageCheckMacros);
-
   tcase_add_test(tc, pageCheckSlotTypeTest);
   tcase_add_test(tc, pageTrecordTypeTest);
   tcase_add_test(tc, pageNoThreadMultPageTest);