2.4. table.cc

This is the implementation of the Table class, with some associated functions. The class is declared in table.hh.

#include "table.hh"
#include <cstdio>
#include <cstring>
#include <cassert>
#include <fcntl.h>
#include <sys/mman.h>

namespace libdoc {

This string is expected to occur at the very start of a valid table file.

const char *Table::MAGIC = "DOC!DB\xFF\0";

Free any dynamic memory associated with the record rec.

void
record_delete (TableRecord &rec)
{
   for (size_t i = 0; i < rec.fields.size(); ++i)
   {
      TableField &fld = rec.fields[i];

      if (fld.type == 's')
         delete fld.v_string;
      else if (fld.type == 'I')
         delete fld.v_int_a;
   }
}

Round the number pos up to the nearest higher multiple of `4'.

template <typename T>
static inline T
jiggle (T pos)
{
   return T (((unsigned long) pos + 3) & ~3);
}

Seek to offset on file descriptor fd, checking for errors.

static off_t
seek_carefully (int fd, off_t offset, int whence)
{
   off_t newoff = lseek (fd, offset, whence);

   if (newoff == (off_t) -1)
      throw ExceptionSystemError ("can't seek in table file");

   return newoff;
}

Read bytes bytes of data from the file descriptor fd and put it in buf, checking carefully for errors. If EOF is encountered it is treated as an error.

static void
read_carefully (int fd, void *buf, size_t bytes)
{
   ssize_t rc = read (fd, buf, bytes);

   if (rc < 0)
      throw ExceptionSystemError ("error reading table file");
   else if ((size_t) rc < bytes)
      throw ExceptionSystemError ("unexpected end of table file");
}

Write bytes bytes of data to the file descriptor fd from buf, checking for errors.

static void
write_carefully (int fd, const void *buf, size_t bytes)
{
   ssize_t rc = write (fd, buf, bytes);

   if ((size_t) rc ≠ bytes)
      throw ExceptionSystemError ("error reading table file");
}

Work out what filename to use for a GDBM index associated with a table, whose filename is filename. If the table file has a file extension then that is removed. We then add the extension .gdbm.

string
Table::make_index_filename (const string &filename)
{
   string fname;

   string::size_type last_slash = filename.rfind ('/');
   if (last_slash ≠ string::npos)
   {
      fname = filename.substr (0, last_slash);
      fname += '/';
   }

   fname += filename;
   fname += ".gdbm";

   return fname;
}

Create a new key in the sequence which is generated automatically for a primary key. If the key provided is non-zero then we are using a key specified by the application, so we use that instead.

Key
Table::autogen_id (Key id)
{
   if (id ≠ 0)
      return id;
   else if (next_id == 0)
      throw ExceptionTable ("this table does not use autogen IDs");
   else if (next_id == KEY_MAX)
      throw ExceptionTable ("can't generate a new ID; they're all used up");

   Key nex = next_id;
   if (next_id ≤ KEY_MAX - key_skip)
      next_id += key_skip;
   else
   {
      key_skip = next_key_skip;
      next_key_skip >>= 1;
      next_id = next_key_skip;
   }

   return nex;
}

Look at the record stored at offset pos in the table file, and work out how big it is.

size_t
Table::find_record_size (off_t pos)
{
   off_t start = pos;

   for (size_t i = 0; i < fields.size(); ++i)
   {
      if (fields[i] == 'i')
         pos += 4;
      else if (fields[i] == 's')
      {
         string s;
         read_string (pos, s, (i + 1 < fields.size()));
      }
      else if (fields[i] == 'I')
      {
         size_t sz;
         read_int (pos, sz);
         pos += sz * 4;
      }
      else
         assert (false);
   }

   return pos - start;
}

Work out how many bytes would be required to store the record represented by rec in the table file.

size_t
Table::calc_record_size (const TableRecord &rec)
{
   size_t sz = 0;

   for (size_t i = 0; i < fields.size(); ++i)
   {
      if (rec.fields[i].type == 'i')
         sz += 4;
      else if (rec.fields[i].type == 's')
      {
         if (i + 1 < fields.size())
            sz += 2;
         sz += rec.fields[i].v_string->size() + 1;
      }
      else if (rec.fields[i].type == 'I')
         sz += (rec.fields[i].v_int_a->size() + 1) * 4;
      else
         assert (false);
   }

   return sz;
}

Write the character c to the file descriptor fd at position offset, checking for errors.

void
Table::write_char (int fd, off_t &offset, unsigned char c)
{
   seek_carefully (fd, offset, SEEK_SET);

   if (size_t (write (fd, &c, 1)) ≠ 1)
      throw ExceptionSystemError ("can't write character to table file");

   ++offset;
}

Skip over a string in the table file, starting at position pos, until we find the terminating null character (which is also skipped over).

void
Table::ignore_string (off_t &pos) const
{
   //TODO: make this read in blocks, and the same for the other functions.
   seek_carefully (table_fd, pos, SEEK_SET);

   char c;
   do
   {
      read_carefully (table_fd, &c, 1);
      ++pos;
   }
   while (c ≠ '\0');
}

Read a string from the table file, starting at offset pos, and put it in s. If read_sz is `true' then we read a 4-byte size value first and read that many bytes, otherwise we read a null-terminated string (eating and discarding the null character).

XXX: this doesn't check the range of p before the first dereference of it. Note that sz includes the `\0'.

void
Table::read_string (off_t &pos, string &s, bool read_sz) const
{
   seek_carefully (table_fd, pos, SEEK_SET);

   unsigned char c;
   if (read_sz)
   {
      read_carefully (table_fd, &c, 1);   ++pos;
      size_t sz = c;
      read_carefully (table_fd, &c, 1);   ++pos;
      sz += (unsigned int) c << 8;

      char *buf = new char[sz + 1];
      read_carefully (table_fd, buf, sz + 1);
      pos += sz + 1;
      s += buf;
      delete buf;
   }
   else
   {
      do
      {
         read_carefully (table_fd, &c, 1);

         ++pos;
         if (c ≠ '\0')
            s += c;
      }
      while (c ≠ '\0');
   }
}

void
Table::write_string (int fd, off_t &offset, const char *s, size_t sz)
{
   seek_carefully (fd, offset, SEEK_SET);

   if (size_t (write (fd, s, sz)) ≠ sz)
      throw ExceptionSystemError ("can't write in write_string()");

   offset += sz;
}

Read a 4 byte integer from the table file, starting at offset pos, into i

TODO: fix this so that it will work on a big-endian machine.

void
Table::read_int (off_t &pos, unsigned int &i) const
{
   seek_carefully (table_fd, pos, SEEK_SET);

   unsigned int x;
   read_carefully (table_fd, &x, 4);

//#ifdef LITTLEENDIAN
   i = x;
//#else
//   i = (x >> 24) & 0x000000FF;
//   i |= (x >> 8) & 0x0000FF00;
//   i |= (x << 8) & 0x00FF0000;
//   i |= (x << 24) & 0xFF000000;
//#endif

   pos += 4;
}

Encode the integer i into 4 bytes appended to the string s.

void
Table::write_int (string &s, unsigned int i)
{
   s += (unsigned char) (i & 0xFF);
   s += (unsigned char) ((i >> 8) & 0xFF);
   s += (unsigned char) ((i >> 16) & 0xFF);
   s += (unsigned char) ((i >> 24) & 0xFF);
}

Write the integer i into 4 bytes on the file descriptor fd, starting at position offset, checking for errors.

void
Table::write_int (int fd, off_t &offset, unsigned int i)
{
   seek_carefully (fd, offset, SEEK_SET);

#ifdef LITTLEENDIAN
   if (size_t (write (fd, &i, 4)) ≠ 4)
      throw ExceptionSystemError ("can't write integer to table file");
#else
   for (int j = 0; j < 4; ++j)
   {
      char byte = (i >> (j * 8)) & 0xFF;
      if (size_t (write (fd, &byte, 1)) ≠ 1)
         throw ExceptionSystemError ("can't write in write_int()");
   }
#endif

   offset += 4;
}

Read a whole record from the table file, starting at offset pos, and put the values for the fields into vals.

void
Table::read_record (off_t &pos, vector <TableField> &vals) const
{
   vals.clear();

   for (size_t i = 0; i < fields.size(); ++i)
   {
      TableField fld;

      fld.type = fields[i];
      if (fld.type == 'i')
         read_int (pos, fld.v_int);
      else if (fld.type == 's')
      {
         fld.v_string = new string;
         read_string (pos, *fld.v_string, (i + 1 < fields.size()));
      }
      else if (fld.type == 'I')
      {
         size_t sz;
         read_int (pos, sz);
         fld.v_int_a = new vector <unsigned int> (sz);

         for (size_t j = 0; j < sz; ++j)
            read_int (pos, (*fld.v_int_a)[j]);
      }
      else
         assert (false);

      vals.push_back (fld);
   }
}

This is dirty and non-portable.

void
Table::read_record (const datum dat, vector <TableField> &vals) const
{
   vals.clear();
   const char *p = dat.dptr;

   for (size_t i = 0; i < fields.size(); ++i)
   {
      TableField fld;

      fld.type = fields[i];
      if (fld.type == 'i')
      {
         fld.v_int = *(const unsigned int *) p;
         p += 4;
      }
      else if (fld.type == 's')
      {
         if (i < fields.size() - 1)
         {
            fld.v_string = new string (p + 2);
            p += 3 + *(const unsigned short *) p;
         }
         else
            fld.v_string = new string (p);
      }
      else if (fld.type == 'I')
      {
         size_t sz = *(const size_t *) p;
         p += 4;
         fld.v_int_a = new vector <unsigned int> (sz);

         for (size_t j = 0; j < sz; ++j)
         {
            (*fld.v_int_a)[j] = *(const unsigned int *) p;
            p += 4;
         }
      }
      else
         assert (false);

      vals.push_back (fld);
   }
}

Add null characters to the file descriptor fd, starting at position offset, to bring it to a position which is a multiple of 4.

void
Table::pad_file (int fd, off_t &offset)
{
   off_t padding = 3 - (offset + 3) % 4;
   write (fd, "\0\0\0", padding);
   offset += padding;
}

Read the header of the table file, storing the values we find (or at least the ones we need) to member variables.

void
Table::read_header ()
{
   seek_carefully (table_fd, 0, SEEK_SET);
   char magbuf[MAGIC_LEN];
   read_carefully (table_fd, magbuf, MAGIC_LEN);
   if (strncmp (magbuf, MAGIC, MAGIC_LEN) ≠ 0)
      throw ExceptionTable ("file is not a DocDB table (bad magic)");

   off_t p = MAGIC_LEN;
   ignore_string (p);         // Version number.
   read_string (p, fields);   // Field types.
   ignore_string (p);         // Field names.
   ignore_string (p);         // Field descriptions.
   string indices;            // List of secondary indices.
   read_string (p, indices);
   ignore_string (p);         // Table commets.

   unsigned int rt, pk;
   p = jiggle (p);
   read_int (p, pk);
   primkey = (PrimKeyType) pk;
   header_offsets = p;
   read_int (p, num_recs);
   read_int (p, next_id);
   read_int (p, rt);
   root = rt;

Decode the other values we need from `next_id', of we are using it.

   if (next_id ≠ 0)
   {
      if (next_id == KEY_MIDDLE)
         key_skip = next_key_skip = KEY_MIDDLE;
      else
      {
         key_skip = 2;
         next_key_skip = 1;
         for (Key x = next_id; (x & 1) == 0; x >>= 1)
            next_key_skip = key_skip, key_skip <<= 1;
      }
   }

Open the secondary index files.

   index_filename = open_secondary_db (index_file, index_field, filename,
                                       indices, fields, readonly);
}

Write the record rec to the table file at offset p. The two children of the new record's node will be set to left_child and right_child.

void
Table::write_record (off_t &p, const TableRecord &rec,
                     off_t left_child, off_t right_child)
{

Write the record header.

   write_int (p, rec.id);
   write_int (p, left_child);
   write_int (p, right_child);

Write the values for the fields.

   for (size_t i = 0; i < fields.size(); ++i)
   {
      const TableField &fld = rec.fields[i];
      if (fld.type ≠ fields[i])
         throw ExceptionTable ("wrong type of value used for a field");

      if (fld.type == 'i')
         write_int (p, fld.v_int);
      else if (fld.type == 's')
      {
         if (i + 1 < fields.size())
         {
            size_t sz = fld.v_string->size();
            if (sz > 0xFFFF)
               throw ExceptionTable ("string value too long for `s' field");
            write_char (p, sz & 0xFF);
            write_char (p, (sz >> 8) & 0xFF);
         }
         write_string (p, *fld.v_string);
      }
      else if (fld.type == 'I')
      {
         write_int (p, fld.v_int_a->size());

         for (size_t j = 0; j < fld.v_int_a->size(); ++j)
            write_int (p, (*fld.v_int_a)[j]);
      }
      else
         assert (false);
   }

Update the secondary index, if there is one.

   if (!index_filename.empty())
   {
      // TODO: fix it to work with integer fields.
      datum dnam, did;
      dnam.dsize = rec.fields[index_field].v_string->size();
      dnam.dptr = (char *) rec.fields[index_field].v_string->c_str();

      did.dsize = 4;
      did.dptr = (char *) &rec.id;

      if (gdbm_store (index_file, dnam, did, GDBM_REPLACE))
         throw ExceptionGDBMError ("error writing to secondary index file");
   }
}

This is for writting records into a GDBM file when we are using a string primary key.

void
Table::write_record (string &dat, const TableRecord &rec)
{
   // Write the values for the fields.
   for (size_t i = 0; i < fields.size(); ++i)
   {
      const TableField &fld = rec.fields[i];
      if (fld.type ≠ fields[i])
         throw ExceptionTable ("wrong type of value used for a field");

      if (fld.type == 'i')
         write_int (dat, fld.v_int);
      else if (fld.type == 's')
      {
         if (i + 1 < fields.size())
         {
            size_t sz = fld.v_string->size();
            if (sz > 0xFFFF)
               throw ExceptionTable ("string value too long for `s' field");
            dat += (char) (sz & 0xFF);
            dat += (char) ((sz >> 8) & 0xFF);
         }
         dat += *fld.v_string;
         dat += (char) 0;
      }
      else if (fld.type == 'I')
      {
         write_int (dat, fld.v_int_a->size());

         for (size_t j = 0; j < fld.v_int_a->size(); ++j)
            write_int (dat, (*fld.v_int_a)[j]);
      }
      else
         assert (false);
   }
}

Write the record rec onto the end of the table file.

void
Table::append_record (const TableRecord &rec,
                      off_t left_child, off_t right_child)
{
   assert (length == jiggle (length));

   off_t p = length;
   write_record (p, rec, left_child, right_child);

   length = jiggle (p);
}

Open the GDBM file used to store records for tables which use a string primary key.

void
Table::open_stringprim_db (GDBM_FILE &db, const string &filename,
                           bool readonly)
{
   string fname = filename;
   string::size_type p = filename.rfind ('.');
   if (p ≠ string::npos)
      fname = filename.substr (0, p);
   fname += ".gdbm";

   db = gdbm_open ((char *) fname.c_str(), 0,
                   (readonly ? GDBM_READER : GDBM_WRCREAT), 0666, 0);

   if (db == 0)
      throw ExceptionGDBMError ("can't open GDBM file for primary index");
}

Open the GDBM file used to store the secondary index.

string
Table::open_secondary_db (GDBM_FILE &db, size_t &field_num, const string &base,
                          const string &indices, const string &fields,
                          bool readonly)
{
   string path;
   string::size_type psl = base.rfind ('/');
   if (psl ≠ string::npos)
      path = base.substr (0, ++psl);

   char buf[256];
   if (sscanf (indices.c_str(), " %u : %255s", &field_num, buf) == 2)
   {
      if (field_num == 0 || --field_num ≥ fields.size())
         throw ExceptionTable ("field number of secondary index is 0 or too"
                               " big");

      string filename = path + make_index_filename (buf);
      db = gdbm_open ((char *) filename.c_str(), 0,
                      GDBM_WRCREAT, 0666, 0);

      if (db == 0)
         throw ExceptionGDBMError ("can't open GDBM file for secondary index");

      return filename;
   }
   else
      return "";
}

Create a new table file.

void
Table::create (const string &filename, const string &fieldtypes,
               PrimKeyType pk, const string &indices)
{
   string fields = fieldtypes;
   for (size_t i = 0; i < fields.size(); ++i)
   {
      if (fields[i] ≠ 'i' && fields[i] ≠ 's' && fields[i] ≠ 'I')
         throw ExceptionTable ("bad field type given in fields spec");
   }
   if (fields.empty())
      fields = 's';

   int tab = 1;
   if (!filename.empty())
   {
      tab = open (filename.c_str(), O_CREAT | O_EXCL | O_WRONLY, 0666);
      if (tab < 0)
         throw ExceptionSystemError ("can't create table file");
   }

   off_t offset = 0;
   write_string (tab, offset, MAGIC, MAGIC_LEN);
   write_string (tab, offset, "0", 2);
   write_string (tab, offset, fields.c_str(), fields.size() + 1);
   write_string (tab, offset, "\0\0", 2);
   write_string (tab, offset, indices.c_str(), indices.size() + 1);
   write_string (tab, offset, "\0", 1);
   pad_file (tab, offset);
   write_int (tab, offset, (int) pk);     // Primary key type.
   write_int (tab, offset, 0);            // Number of records.
   write_int (tab, offset, (pk == PRIMKEY_AUTO) ? KEY_MIDDLE : 0);
   write_int (tab, offset, 0);            // Offset to root node.

   if (pk == PRIMKEY_STRING)
   {
      GDBM_FILE db;
      open_stringprim_db (db, filename, false);
      gdbm_close (db);
   }

   GDBM_FILE db;
   size_t index_field;
   if (!open_secondary_db (db, index_field, filename, indices, fieldtypes,
                           false).empty())
      gdbm_close (db);

   close (tab);
}

Open a table file and read its header information.

Table::Table (const string &fname, bool ro)
   : filename (fname), readonly (ro)
{
   table_fd = open (filename.c_str(), O_RDWR);
   if (table_fd < 0)
      throw ExceptionSystemError ("can't open table file");
   off_t endoff = seek_carefully (table_fd, 0, SEEK_END);
   length = jiggle (endoff);     // Rude.

   read_header();

   if (primkey == PRIMKEY_STRING)
      open_stringprim_db (table_gdbm, filename, readonly);
}

Close a table file, making sure the header fields are updated.

Table::~Table ()
{
   sync();

   close (table_fd);
   if (primkey == PRIMKEY_STRING)
      gdbm_close (table_gdbm);

   if (!index_filename.empty())
      gdbm_close (index_file);
}

Adds a new record with the key id and stores vals as the values of its fields. If there is already a record with the key id then no change is made. In that case `false' is returned, otherwise `true'.

TODO: update this old hat comment.

bool
Table::insert (TableRecord &rec)
{
   if (readonly)
      throw ExceptionTable ("can't insert into read-only table");
   if (rec.fields.size() ≠ fields.size())
      throw ExceptionTable ("wrong number of values passed to Table::insert");
   if (primkey == PRIMKEY_STRING)
      throw ExceptionTable ("wrong version of `insert' for a string primary"
                            " key");

   rec.id = autogen_id (rec.id);

   if (root == 0)
   {
      root = length;
      append_record (rec);
      num_recs = 1;
      return true;
   }

   off_t p = root;
   unsigned int i;

   while (read_int (p, i), i ≠ rec.id)
   {
      unsigned int child;
      read_int (p, child);
      if (rec.id > i)
         read_int (p, child);

      if (child == 0)
      {
         p -= 4;
         write_int (p, jiggle (length));
         append_record (rec);
         ++num_recs;
         return true;
      }

      p = child;
   }

   return false;
}

Adds a new record with the key id and stores vals as the values of its fields. If there is already a record with the key id then no change is made. In that case `false' is returned, otherwise `true'.

TODO: update this old hat comment.

bool
Table::insert (const string &key, TableRecord &rec)
{
   if (readonly)
      throw ExceptionTable ("can't insert into read-only table");
   if (rec.fields.size() ≠ fields.size())
      throw ExceptionTable ("wrong number of values passed to Table::insert");
   if (primkey ≠ PRIMKEY_STRING)
      throw ExceptionTable ("wrong version of `insert' for an integer primary"
                            " key");

   datum dkey, drec;
   dkey.dsize = key.size();
   dkey.dptr = (char *) key.c_str();

   string d;
   write_record (d, rec);
   drec.dsize = d.size();
   drec.dptr = (char *) d.c_str();

   if (gdbm_store (table_gdbm, dkey, drec, GDBM_REPLACE))
      throw ExceptionGDBMError ("error writing to GDBM file");

   return true;
}

bool
Table::update (const TableRecord &rec)
{
   if (primkey == PRIMKEY_STRING)
      throw ExceptionTable ("wrong version of `update' used for a string"
                            " primary key");
   if (readonly)
      throw ExceptionTable ("can't update record in read-only table");

   if (root == 0)
      return false;

   off_t p = root, parent = 0;
   unsigned int i;

   while (read_int (p, i), i ≠ rec.id)
   {
      unsigned int child;
      if (rec.id > i)
         p += 4;
      read_int (p, child);

      if (child == 0)
         return false;

      parent = p - 4;
      p = child;
   }

   unsigned int left_child, right_child;
   read_int (p, left_child);
   read_int (p, right_child);
   size_t oldrecsz = find_record_size (p);
   size_t newrecsz = calc_record_size (rec);

   if (newrecsz ≤ oldrecsz)
   {
      p -= 12;
      write_record (p, rec, left_child, right_child);
   }
   else
   {
      if (parent == 0)
         root = length;
      else
         write_int (parent, length);
      append_record (rec, left_child, right_child);
   }

   return true;
}

bool
Table::update (const string &key, const TableRecord &rec)
{
   if (primkey ≠ PRIMKEY_STRING)
      throw ExceptionTable ("wrong version of `update' used for an integer"
                            " primary key");
   if (readonly)
      throw ExceptionTable ("can't update record in read-only table");

   datum dkey, drec;
   dkey.dsize = key.size();
   dkey.dptr = (char *) key.c_str();
   string srec;
   write_record (srec, rec);
   drec.dsize = srec.size();
   drec.dptr = (char *) srec.c_str();

   if (gdbm_store (table_gdbm, dkey, drec, GDBM_REPLACE))
      throw ExceptionGDBMError ("error writing value to GDBM file");

   return true;
}

bool
Table::exists (Key id) const
{
   if (root == 0)
      return false;

   off_t p = root;
   unsigned int i;

   while (read_int (p, i), i ≠ id)
   {
      unsigned int child;
      if (id > i)
         p += 4;
      read_int (p, child);

      if (child == 0)
         return false;

      p = child;
   }

   return true;
}

bool
Table::find (Key id, vector <TableField> &vals) const
{
   if (primkey == PRIMKEY_STRING)
      throw ExceptionTable ("find (int, ...) only works if the primary key"
                            " is an integer");

   if (root == 0)
      return false;

   off_t p = root;
   unsigned int i;

   while (read_int (p, i), i ≠ id)
   {
      unsigned int child;
      if (id > i)
         p += 4;
      read_int (p, child);

      if (child == 0)
         return false;

      p = child;
   }

   p += 8;        // Skip over child offsets.
   read_record (p, vals);

   return true;
}

bool
Table::find (const string &key, vector <TableField> &vals) const
{
   if (primkey ≠ PRIMKEY_STRING)
      throw ExceptionTable ("find (string, ...) only works if the primary key"
                            " is a string");

   datum dkey;
   dkey.dsize = key.size();
   dkey.dptr = (char *) key.c_str();

   datum drec = gdbm_fetch (table_gdbm, dkey);

   if (drec.dptr)
   {
      read_record (drec, vals);

      free (drec.dptr);
      return true;
   }
   else
      return false;
}

bool
Table::find (TableField seckey, TableRecord &rec) const
{
   if (find_id (seckey, rec.id))
      return find (rec.id, rec.fields);
   else
      return false;
}

bool
Table::find_id (TableField seckey, Key &id) const
{
   if (primkey == PRIMKEY_STRING)
      throw ExceptionTable ("find_id can only be used with integer primary"
                            " keys");
   if ((char) seckey.type ≠ fields[index_field])
      throw ExceptionTable ("wrong type of value used for secondary key");

   datum seckeyd, keyd;
   seckeyd.dptr = (char *) seckey.v_string->c_str();
   seckeyd.dsize = seckey.v_string->size();
   keyd = gdbm_fetch (index_file, seckeyd);

   if (keyd.dptr)
   {
      if (keyd.dsize ≠ 4)
         throw ExceptionTable ("bad size for key in secondary index");

      id = *(Key *) keyd.dptr;

      free (keyd.dptr);
      return true;
   }
   else
      return false;
}

void
Table::leftmost_stack (off_t p, Iterator &iter) const
{
   assert (primkey ≠ PRIMKEY_STRING);

   do
   {
      iter.stack->push_back (pair <off_t, bool> (p, true));
      p += 4;
      unsigned int child;
      read_int (p, child);
      p = child;
   }
   while (p);
}

bool
Table::iter_begin (Iterator &iter) const
{
   if (primkey == PRIMKEY_STRING)
   {
      iter.dat = gdbm_firstkey (table_gdbm);

      return iter.dat.dptr;
   }
   else
   {
      iter.stack = new vector <pair <off_t, bool> >;

      if (root == 0)
         return false;

      leftmost_stack (root, iter);

      return true;
   }
}

bool
Table::iter_next (Iterator &iter) const
{
   if (primkey == PRIMKEY_STRING)
   {
      datum next = gdbm_nextkey (table_gdbm, iter.dat);

      free (iter.dat.dptr);
      iter.dat = next;

      return iter.dat.dptr;
   }
   else
   {
      bool popped = false;

      while (!iter.stack->empty())
      {
         pair <off_t, bool> &back = iter.stack->back();

         if (back.second)
         {
            if (popped)
               return true;

            off_t off = back.first;    // Check right hand child.
            off += 8;
            unsigned int child;
            read_int (off, child);

            if (child ≠ 0)
            {
               back.second = false;
               leftmost_stack (child, iter);
               return true;
            }
            else
            {
               popped = true;
               iter.stack->pop_back();
            }
         }
         else
         {
            popped = true;
            iter.stack->pop_back();
         }
      }

      return false;
   }
}

void
Table::iter_deref (const Iterator &iter, TableRecord &rec) const
{
   if (primkey == PRIMKEY_STRING)
   {
      assert (iter.dat.dptr);

      datum val = gdbm_fetch (table_gdbm, iter.dat);
      assert (val.dptr);

      rec.id = 0;
      read_record (val, rec.fields);
      free (val.dptr);
   }
   else
   {
      assert (!iter.stack->empty());

      off_t p = iter.stack->back().first;
      read_int (p, rec.id);

      p += 8;                 // Skip child offsets.
      read_record (p, rec.fields);
   }
}

void
Table::iter_delete (Iterator &iter) const
{
   if (primkey == PRIMKEY_STRING)
   {
      if (iter.dat.dptr)
         free (iter.dat.dptr);
   }
   else
      delete iter.stack;
}

void
Table::sync () const
{
   if (!readonly)
   {
      seek_carefully (table_fd, header_offsets, SEEK_SET);

      write_carefully (table_fd, &num_recs, 4);
      write_carefully (table_fd, &next_id, 4);

      int x = root;
      write_carefully (table_fd, &x, 4);
   }
}

This doesn't work with string primary keys, of course.

void
Table::debugging_recurse (FILE *ous, unsigned int indentation,
                          off_t node) const
{
   for (unsigned int i = 0; i < indentation; ++i)
      fputc (' ', ous);

   if (node == 0)
   {
      fprintf (ous, "<empty>\n");
      return;
   }

   off_t p = node;
   unsigned int key, lchild, rchild;
   read_int (p, key);
   read_int (p, lchild);
   read_int (p, rchild);
   string rec;
   read_string (p, rec);

//   fprintf (ous, "%u @ 0x%X -> %s\n", (int) key, (int) node, rec.c_str());
   fprintf (ous, "%u @ 0x%X\n", (int) key, (int) node);

   debugging_recurse (ous, ++indentation, lchild);
   debugging_recurse (ous, indentation, rchild);
}

void
Table::debugging_tree (FILE *ous) const
{
   debugging_recurse (ous, 0, root);
}

}  // namespace libdoc