X-Git-Url: http://git.indexdata.com/?p=idzebra-moved-to-github.git;a=blobdiff_plain;f=isamb%2Fisamb.c;h=c732143457f9b8f1dc2eb9a1ecb3cb601cf7237e;hp=de0cac9c8613fef8ba02a894167d09fe15d023c2;hb=05259bf944ff61befc9bee89b7377bf365b25099;hpb=a0acd766ca9eac2c185da604cdc0b93275a57863

diff --git a/isamb/isamb.c b/isamb/isamb.c
index de0cac9..c732143 100644
--- a/isamb/isamb.c
+++ b/isamb/isamb.c
@@ -1,5 +1,5 @@
-/* $Id: isamb.c,v 1.25 2003-03-20 14:37:35 adam Exp $
-   Copyright (C) 1995,1996,1997,1998,1999,2000,2001,2002,2003
+/* $Id: isamb.c,v 1.30 2004-06-01 12:56:38 adam Exp $
+   Copyright (C) 1995,1996,1997,1998,1999,2000,2001,2002,2003,2004
    Index Data Aps
 
 This file is part of the Zebra server.
@@ -36,13 +36,24 @@ struct ISAMB_head {
 
 #define ISAMB_DATA_OFFSET 3
 
+/* maximum size of encoded buffer */
 #define DST_ITEM_MAX 256
 
-/* approx 2*4 K + max size of item */
-#define DST_BUF_SIZE 8448
+#define ISAMB_MAX_LEVEL 10
+/* approx 2*max page + max size of item */
+#define DST_BUF_SIZE 16840
 
 #define ISAMB_CACHE_ENTRY_SIZE 4096
 
+/* CAT_MAX: _must_ be power of 2 */
+#define CAT_MAX 4
+#define CAT_MASK (CAT_MAX-1)
+/* CAT_NO: <= CAT_MAX */
+#define CAT_NO 4
+
+/* ISAMB_PTR_CODEC=1 var, =0 fixed */
+#define ISAMB_PTR_CODEC  1
+
 struct ISAMB_cache_entry {
     ISAMB_P pos;
     unsigned char *buf;
@@ -51,7 +62,6 @@ struct ISAMB_cache_entry {
     struct ISAMB_cache_entry *next;
 };
 
-
 struct ISAMB_file {
     BFile bf;
     int head_dirty;
@@ -61,13 +71,17 @@ struct ISAMB_file {
 
 struct ISAMB_s {
     BFiles bfs;
-    ISAMC_M method;
+    ISAMC_M *method;
 
     struct ISAMB_file *file;
     int no_cat;
     int cache; /* 0=no cache, 1=use cache, -1=dummy isam (for testing only) */
     int log_io;        /* log level for bf_read/bf_write calls */
     int log_freelist;  /* log level for freelist handling */
+    int skipped_numbers; /* on a leaf node */
+    int returned_numbers; 
+    int skipped_nodes[ISAMB_MAX_LEVEL]; /* [0]=skipped leaves, 1=higher etc */
+    int accessed_nodes[ISAMB_MAX_LEVEL]; /* nodes we did not skip */
 };
 
 struct ISAMB_block {
@@ -88,36 +102,78 @@ struct ISAMB_PP_s {
     ISAMB isamb;
     ISAMB_P pos;
     int level;
+    int maxlevel; /* total depth */
     int total_size;
     int no_blocks;
+    int skipped_numbers; /* on a leaf node */
+    int returned_numbers; 
+    int skipped_nodes[ISAMB_MAX_LEVEL]; /* [0]=skipped leaves, 1=higher etc */
+    int accessed_nodes[ISAMB_MAX_LEVEL]; /* nodes we did not skip */
     struct ISAMB_block **block;
 };
 
-void encode_ptr (char **dst, int pos)
+#if ISAMB_PTR_CODEC
+static void encode_ptr (char **dst, unsigned pos)
 {
-    memcpy (*dst, &pos, sizeof(pos));
+    unsigned char *bp = (unsigned char*) *dst;
+
+    while (pos > 127)
+    {
+         *bp++ = 128 | (pos & 127);
+         pos = pos >> 7;
+    }
+    *bp++ = pos;
+    *dst = (char *) bp;
+}
+#else
+static void encode_ptr (char **dst, unsigned pos)
+{
+    memcpy(*dst, &pos, sizeof(pos));
     (*dst) += sizeof(pos);
 }
+#endif
+
+#if ISAMB_PTR_CODEC
+static void decode_ptr (char **src1, int *pos)
+{
+    unsigned char **src = (unsigned char **) src1;
+    unsigned d = 0;
+    unsigned char c;
+    unsigned r = 0;
 
-void decode_ptr (char **src, int *pos)
+    while (((c = *(*src)++) & 128))
+    {
+        d += ((c & 127) << r);
+	r += 7;
+    }
+    d += (c << r);
+    *pos = d;
+}
+#else
+static void decode_ptr (char **src, int *pos)
 {
-    memcpy (pos, *src, sizeof(*pos));
-    (*src) += sizeof(*pos);
+     memcpy (pos, *src, sizeof(*pos));
+     (*src) += sizeof(*pos);
 }
+#endif
 
-ISAMB isamb_open (BFiles bfs, const char *name, int writeflag, ISAMC_M method,
+ISAMB isamb_open (BFiles bfs, const char *name, int writeflag, ISAMC_M *method,
                   int cache)
 {
     ISAMB isamb = xmalloc (sizeof(*isamb));
     int i, b_size = 32;
 
     isamb->bfs = bfs;
-    isamb->method = (ISAMC_M) xmalloc (sizeof(*method));
+    isamb->method = (ISAMC_M *) xmalloc (sizeof(*method));
     memcpy (isamb->method, method, sizeof(*method));
-    isamb->no_cat = 4;
+    isamb->no_cat = CAT_NO;
     isamb->log_io = 0;
     isamb->log_freelist = 0;
     isamb->cache = cache;
+    isamb->skipped_numbers=0;
+    isamb->returned_numbers=0;
+    for (i=0;i<ISAMB_MAX_LEVEL;i++)
+      isamb->skipped_nodes[i]= isamb->accessed_nodes[i]=0;
 
     assert (cache == 0);
     isamb->file = xmalloc (sizeof(*isamb->file) * isamb->no_cat);
@@ -170,11 +226,11 @@ static void flush_blocks (ISAMB b, int cat)
 
 static int get_block (ISAMB b, ISAMC_P pos, char *userbuf, int wr)
 {
-    int cat = pos&3;
-    int off = ((pos/4) & 
+    int cat = pos&CAT_MASK;
+    int off = ((pos/CAT_MAX) & 
                (ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size - 1))
         * b->file[cat].head.block_size;
-    int norm = pos / (4*ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size);
+    int norm = pos / (CAT_MASK*ISAMB_CACHE_ENTRY_SIZE / b->file[cat].head.block_size);
     int no = 0;
     struct ISAMB_cache_entry **ce, *ce_this = 0, **ce_last = 0;
 
@@ -244,6 +300,11 @@ static int get_block (ISAMB b, ISAMC_P pos, char *userbuf, int wr)
 void isamb_close (ISAMB isamb)
 {
     int i;
+    for (i=0;isamb->accessed_nodes[i];i++)
+        logf(LOG_DEBUG,"isamb_close  level leaf-%d: %d read, %d skipped",
+             i, isamb->accessed_nodes[i], isamb->skipped_nodes[i]);
+    logf(LOG_DEBUG,"isamb_close returned %d values, skipped %d",
+         isamb->skipped_numbers, isamb->returned_numbers);
     for (i = 0; i<isamb->no_cat; i++)
     {
         flush_blocks (isamb, i);
@@ -261,22 +322,22 @@ void isamb_close (ISAMB isamb)
 
 struct ISAMB_block *open_block (ISAMB b, ISAMC_P pos)
 {
-    int cat = pos&3;
+    int cat = pos&CAT_MASK;
     struct ISAMB_block *p;
     if (!pos)
         return 0;
     p = xmalloc (sizeof(*p));
     p->pos = pos;
-    p->cat = pos & 3;
+    p->cat = pos & CAT_MASK;
     p->buf = xmalloc (b->file[cat].head.block_size);
 
     if (!get_block (b, pos, p->buf, 0))
     {
         yaz_log (b->log_io, "bf_read: open_block");
-        if (!bf_read (b->file[cat].bf, pos/4, 0, 0, p->buf))
+        if (!bf_read (b->file[cat].bf, pos/CAT_MAX, 0, 0, p->buf))
         {
-            yaz_log (LOG_FATAL, "read failure for pos=%ld block=%ld",
-                     (long) pos, (long) pos/4);
+            yaz_log (LOG_FATAL, "isamb: read fail for pos=%ld block=%ld",
+                     (long) pos, (long) pos/CAT_MAX);
             abort();
         }
     }
@@ -285,13 +346,14 @@ struct ISAMB_block *open_block (ISAMB b, ISAMC_P pos)
     p->size = (p->buf[1] + 256 * p->buf[2]) - ISAMB_DATA_OFFSET;
     if (p->size < 0)
     {
-        fprintf (stderr, "pos=%d\n", pos);
+        yaz_log (LOG_FATAL, "Bad block size %d in pos=%d\n", p->size, pos);
     }
     assert (p->size >= 0);
     p->offset = 0;
     p->dirty = 0;
     p->deleted = 0;
     p->decodeClientData = (*b->method->code_start)(ISAMC_DECODE);
+    yaz_log (LOG_DEBUG, "isamb_open_block: Opened block %d ofs=%d",pos, p->offset);
     return p;
 }
 
@@ -306,24 +368,24 @@ struct ISAMB_block *new_block (ISAMB b, int leaf, int cat)
     {
         int block_no;
         block_no = b->file[cat].head.last_block++;
-        p->pos = block_no * 4 + cat;
+        p->pos = block_no * CAT_MAX + cat;
     }
     else
     {
         p->pos = b->file[cat].head.free_list;
-        assert((p->pos & 3) == cat);
+        assert((p->pos & CAT_MASK) == cat);
         if (!get_block (b, p->pos, p->buf, 0))
         {
             yaz_log (b->log_io, "bf_read: new_block");
-            if (!bf_read (b->file[cat].bf, p->pos/4, 0, 0, p->buf))
+            if (!bf_read (b->file[cat].bf, p->pos/CAT_MAX, 0, 0, p->buf))
             {
-                yaz_log (LOG_FATAL, "read failure for pos=%ld block=%ld",
-                         (long) p->pos/4, (long) p->pos/4);
+                yaz_log (LOG_FATAL, "isamb: read fail for pos=%ld block=%ld",
+                         (long) p->pos/CAT_MAX, (long) p->pos/CAT_MAX);
                 abort ();
             }
         }
         yaz_log (b->log_freelist, "got block %d from freelist %d:%d", p->pos,
-                 cat, p->pos/4);
+                 cat, p->pos/CAT_MAX);
         memcpy (&b->file[cat].head.free_list, p->buf, sizeof(int));
     }
     p->cat = cat;
@@ -365,7 +427,7 @@ static void check_block (ISAMB b, struct ISAMB_block *p)
         int pos;
             
         decode_ptr (&src, &pos);
-        assert ((pos&3) == p->cat);
+        assert ((pos&CAT_MASK) == p->cat);
         while (src != endp)
         {
             int item_len;
@@ -373,7 +435,7 @@ static void check_block (ISAMB b, struct ISAMB_block *p)
             assert (item_len > 0 && item_len < 30);
             src += item_len;
             decode_ptr (&src, &pos);
-            assert ((pos&3) == p->cat);
+            assert ((pos&CAT_MASK) == p->cat);
         }
     }
 }
@@ -385,13 +447,13 @@ void close_block (ISAMB b, struct ISAMB_block *p)
     if (p->deleted)
     {
         yaz_log (b->log_freelist, "release block %d from freelist %d:%d",
-                 p->pos, p->cat, p->pos/4);
+                 p->pos, p->cat, p->pos/CAT_MAX);
         memcpy (p->buf, &b->file[p->cat].head.free_list, sizeof(int));
         b->file[p->cat].head.free_list = p->pos;
         if (!get_block (b, p->pos, p->buf, 1))
         {
             yaz_log (b->log_io, "bf_write: close_block (deleted)");
-            bf_write (b->file[p->cat].bf, p->pos/4, 0, 0, p->buf);
+            bf_write (b->file[p->cat].bf, p->pos/CAT_MAX, 0, 0, p->buf);
         }
     }
     else if (p->dirty)
@@ -405,7 +467,7 @@ void close_block (ISAMB b, struct ISAMB_block *p)
         if (!get_block (b, p->pos, p->buf, 1))
         {
             yaz_log (b->log_io, "bf_write: close_block");
-            bf_write (b->file[p->cat].bf, p->pos/4, 0, 0, p->buf);
+            bf_write (b->file[p->cat].bf, p->pos/CAT_MAX, 0, 0, p->buf);
         }
     }
     (*b->method->code_stop)(ISAMC_DECODE, p->decodeClientData);
@@ -415,14 +477,14 @@ void close_block (ISAMB b, struct ISAMB_block *p)
 
 int insert_sub (ISAMB b, struct ISAMB_block **p,
                 void *new_item, int *mode,
-                ISAMC_I stream,
+                ISAMC_I *stream,
                 struct ISAMB_block **sp,
                 void *sub_item, int *sub_size,
                 void *max_item);
 
 int insert_int (ISAMB b, struct ISAMB_block *p, void *lookahead_item,
                 int *mode,
-                ISAMC_I stream, struct ISAMB_block **sp,
+                ISAMC_I *stream, struct ISAMB_block **sp,
                 void *split_item, int *split_size, void *last_max_item)
 {
     char *startp = p->bytes;
@@ -531,7 +593,8 @@ int insert_int (ISAMB b, struct ISAMB_block *p, void *lookahead_item,
 
 
 int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
-                 int *lookahead_mode, ISAMC_I stream, struct ISAMB_block **sp2,
+                 int *lookahead_mode, ISAMC_I *stream,
+		 struct ISAMB_block **sp2,
                  void *sub_item, int *sub_size,
                  void *max_item)
 {
@@ -542,7 +605,7 @@ int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
     void *c1 = (*b->method->code_start)(ISAMC_DECODE);
     void *c2 = (*b->method->code_start)(ISAMC_ENCODE);
     int more = 1;
-    int quater = b->file[b->no_cat-1].head.block_max / 4;
+    int quater = b->file[b->no_cat-1].head.block_max / CAT_MAX;
     char *cut = dst_buf + quater * 2;
     char *maxp = dst_buf + b->file[b->no_cat-1].head.block_max;
     char *half1 = 0;
@@ -571,7 +634,11 @@ int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
             if (d > 0)
             {
                 dst_item = lookahead_item;
-                assert (*lookahead_mode);
+                if (!*lookahead_mode)
+                {
+                    yaz_log (LOG_WARN, "isamb: Inconsistent register (1)");
+                    assert (*lookahead_mode);
+                }
             }
             else
                 dst_item = file_item_buf;
@@ -657,7 +724,7 @@ int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
         }
         if (!*lookahead_mode)
         {
-            yaz_log (LOG_WARN, "Inconsistent register (2)");
+            yaz_log (LOG_WARN, "isamb: Inconsistent register (2)");
             abort();
         }
         else if (!half1 && dst > cut)   
@@ -752,7 +819,7 @@ int insert_leaf (ISAMB b, struct ISAMB_block **sp1, void *lookahead_item,
 
 int insert_sub (ISAMB b, struct ISAMB_block **p, void *new_item,
                 int *mode,
-                ISAMC_I stream,
+                ISAMC_I *stream,
                 struct ISAMB_block **sp,
                 void *sub_item, int *sub_size,
                 void *max_item)
@@ -765,7 +832,36 @@ int insert_sub (ISAMB b, struct ISAMB_block **p, void *new_item,
                            sub_size, max_item);
 }
 
-int isamb_merge (ISAMB b, ISAMC_P pos, ISAMC_I stream)
+int isamb_unlink (ISAMB b, ISAMC_P pos)
+{
+    struct ISAMB_block *p1;
+
+    if (!pos)
+	return 0;
+    p1 = open_block(b, pos);
+    p1->deleted = 1;
+    if (!p1->leaf)
+    {
+	int sub_p;
+	int item_len;
+	char *src = p1->bytes + p1->offset;
+
+	decode_ptr(&src, &sub_p);
+	isamb_unlink(b, sub_p);
+	
+	while (src != p1->bytes + p1->size)
+	{
+	    decode_ptr(&src, &item_len);
+	    src += item_len;
+	    decode_ptr(&src, &sub_p);
+	    isamb_unlink(b, sub_p);
+	}
+    }
+    close_block(b, p1);
+    return 0;
+}
+
+int isamb_merge (ISAMB b, ISAMC_P pos, ISAMC_I *stream)
 {
     char item_buf[DST_ITEM_MAX];
     char *item_ptr;
@@ -822,14 +918,20 @@ int isamb_merge (ISAMB b, ISAMC_P pos, ISAMC_I stream)
 ISAMB_PP isamb_pp_open_x (ISAMB isamb, ISAMB_P pos, int *level)
 {
     ISAMB_PP pp = xmalloc (sizeof(*pp));
+    int i;
 
     pp->isamb = isamb;
-    pp->block = xmalloc (10 * sizeof(*pp->block));
+    pp->block = xmalloc (ISAMB_MAX_LEVEL * sizeof(*pp->block));
 
     pp->pos = pos;
     pp->level = 0;
+    pp->maxlevel=0;
     pp->total_size = 0;
     pp->no_blocks = 0;
+    pp->skipped_numbers=0;
+    pp->returned_numbers=0;
+    for (i=0;i<ISAMB_MAX_LEVEL;i++)
+        pp->skipped_nodes[i] = pp->accessed_nodes[i]=0;
     while (1)
     {
         struct ISAMB_block *p = open_block (isamb, pos);
@@ -838,15 +940,17 @@ ISAMB_PP isamb_pp_open_x (ISAMB isamb, ISAMB_P pos, int *level)
 
         pp->total_size += p->size;
         pp->no_blocks++;
-
         if (p->leaf)
             break;
 
+                                        
         decode_ptr (&src, &pos);
         p->offset = src - p->bytes;
         pp->level++;
+        pp->accessed_nodes[pp->level]++; 
     }
     pp->block[pp->level+1] = 0;
+    pp->maxlevel=pp->level;
     if (level)
         *level = pp->level;
     return pp;
@@ -862,6 +966,19 @@ void isamb_pp_close_x (ISAMB_PP pp, int *size, int *blocks)
     int i;
     if (!pp)
         return;
+    logf(LOG_DEBUG,"isamb_pp_close lev=%d returned %d values, skipped %d",
+        pp->maxlevel, pp->skipped_numbers, pp->returned_numbers);
+    for (i=pp->maxlevel;i>=0;i--)
+        if ( pp->skipped_nodes[i] || pp->accessed_nodes[i])
+            logf(LOG_DEBUG,"isamb_pp_close  level leaf-%d: %d read, %d skipped", i,
+                 pp->accessed_nodes[i], pp->skipped_nodes[i]);
+    pp->isamb->skipped_numbers += pp->skipped_numbers;
+    pp->isamb->returned_numbers += pp->returned_numbers;
+    for (i=pp->maxlevel;i>=0;i--)
+    {
+        pp->isamb->accessed_nodes[i] += pp->accessed_nodes[i];
+        pp->isamb->skipped_nodes[i] += pp->skipped_nodes[i];
+    }
     if (size)
         *size = pp->total_size;
     if (blocks)
@@ -884,6 +1001,12 @@ void isamb_pp_close (ISAMB_PP pp)
     isamb_pp_close_x (pp, 0, 0);
 }
 
+
+#if 0
+/* Old isamb_pp_read that Adam wrote, kept as a reference in case we need to
+   debug the more complex pp_read that also forwards. May be deleted near end
+   of 2004, if it has not shown to be useful */
+
 int isamb_pp_read (ISAMB_PP pp, void *buf)
 {
     char *dst = buf;
@@ -903,7 +1026,7 @@ int isamb_pp_read (ISAMB_PP pp, void *buf)
             pp->block[pp->level] = 0;
             (pp->level)--;
             p = pp->block[pp->level];
-            assert (!p->leaf);  /* must be int */
+            assert (!p->leaf);  
         }
         src = p->bytes + p->offset;
         
@@ -922,7 +1045,7 @@ int isamb_pp_read (ISAMB_PP pp, void *buf)
             pp->total_size += p->size;
             pp->no_blocks++;
             
-            if (p->leaf) /* leaf */
+            if (p->leaf) 
             {
                 break;
             }
@@ -938,9 +1061,175 @@ int isamb_pp_read (ISAMB_PP pp, void *buf)
     (*pp->isamb->method->code_item)(ISAMC_DECODE, p->decodeClientData,
                                     &dst, &src);
     p->offset = src - (char*) p->bytes;
+    key_logdump_txt(LOG_DEBUG,buf, "isamb_pp_read returning 1");
     return 1;
 }
 
+#else
+int isamb_pp_read (ISAMB_PP pp, void *buf)
+{
+    return isamb_pp_forward(pp,buf,0);
+}
+#endif
+
+int isamb_pp_forward (ISAMB_PP pp, void *buf, const void *untilbuf)
+{
+    /* pseudocode:
+     *   while 1
+     *     while at end of node
+     *       climb higher. If out, return 0
+     *     while not on a leaf (and not at its end)
+     *       decode next
+     *       if cmp
+     *         descend to node
+     *     decode next
+     *     if cmp
+     *       return 1
+     */
+        /* 
+         * The upper nodes consist of a sequence of nodenumbers and keys
+         * When opening a block,  the first node number is read in, and
+         * offset points to the first key, which is the upper limit of keys
+         * in the node just read.
+         */
+    char *dst = buf;
+    char *src;
+    struct ISAMB_block *p = pp->block[pp->level];
+    int cmp;
+    int item_len;
+    int pos;
+    int nxtpos;
+    if (!p)
+        return 0;
+    logf(LOG_DEBUG,"isamb_pp_forward starting [%p] p=%d",pp,p->pos);
+    
+    (*pp->isamb->method->log_item)(LOG_DEBUG, untilbuf, "until");
+    (*pp->isamb->method->log_item)(LOG_DEBUG, buf, "buf");
+
+    while (1)
+    {
+        while ( p->offset == p->size) 
+        {  /* end of this block - climb higher */
+            logf(LOG_DEBUG,"isamb_pp_forward climbing from l=%d",
+                            pp->level);
+            if (pp->level == 0)
+            {
+                logf(LOG_DEBUG,"isamb_pp_forward returning 0 at root");
+                return 0; /* at end of the root, nothing left */
+            }
+            close_block(pp->isamb, pp->block[pp->level]);
+            pp->block[pp->level]=0;
+            (pp->level)--;
+            p=pp->block[pp->level];
+            logf(LOG_DEBUG,"isamb_pp_forward climbed to node %d off=%d",
+                            p->pos, p->offset);
+            assert(!p->leaf);
+            /* skip the child we have handled */
+            if (p->offset != p->size)
+            { 
+                src = p->bytes + p->offset;
+                decode_ptr(&src, &item_len);
+		
+		(*pp->isamb->method->log_item)(LOG_DEBUG, src,
+					       " isamb_pp_forward "
+					       "climb skipping old key");
+                src += item_len;
+                decode_ptr(&src,&pos);
+                p->offset = src - (char*) p->bytes;
+                break; /* even if this puts us at the end of the block, we need to */
+                       /* descend to the last pos. UGLY coding, clean up some */
+                       /* day */
+            }
+        }
+        if (!p->leaf)
+        { 
+            src = p->bytes + p->offset;
+            if (p->offset == p->size)
+                cmp=-2 ; /* descend to the last node, as we have no value to cmp */
+            else
+            {
+                decode_ptr(&src, &item_len);
+                logf(LOG_DEBUG,"isamb_pp_forward (B) on a high node. ofs=%d sz=%d nxtpos=%d ",
+                        p->offset,p->size,pos);
+
+
+		(*pp->isamb->method->log_item)(LOG_DEBUG, src, "");
+                if (untilbuf)
+                    cmp=(*pp->isamb->method->compare_item)(untilbuf,src);
+                else
+                    cmp=-2;
+                src += item_len;
+                decode_ptr(&src,&nxtpos);
+            }
+            if (cmp<2)
+            { 
+                logf(LOG_DEBUG,"isambb_pp_forward descending l=%d p=%d ",
+                            pp->level, pos);
+                ++(pp->level);
+                p = open_block(pp->isamb,pos);
+                pp->block[pp->level] = p ;
+                pp->total_size += p->size;
+                (pp->accessed_nodes[pp->maxlevel - pp->level])++;
+                pp->no_blocks++;
+                if ( !p->leaf)
+                { /* block starts with a pos */
+                    src = p->bytes + p->offset;
+                    decode_ptr(&src,&pos);
+                    p->offset=src-(char*) p->bytes;
+                    logf(LOG_DEBUG,"isamb_pp_forward: block %d starts with %d",
+                                    p->pos, pos);
+                } 
+            } /* descend to the node */
+            else
+            { /* skip the node */
+                p->offset = src - (char*) p->bytes;
+                pos=nxtpos;
+                (pp->skipped_nodes[pp->maxlevel - pp->level -1])++;
+                logf(LOG_DEBUG,
+                    "isamb_pp_forward: skipping block on level %d, noting on %d (%d)",
+                    pp->level, pp->maxlevel - pp->level-1 , 
+                    pp->skipped_nodes[pp->maxlevel - pp->level-1 ]);
+                /* 0 is always leafs, 1 is one level above leafs etc, no
+                 * matter how high tree */
+            }
+        } /* not on a leaf */
+        else
+        { /* on a leaf */
+            src = p->bytes + p->offset;
+            dst=buf;
+            (*pp->isamb->method->code_item)(ISAMC_DECODE, p->decodeClientData,
+                                            &dst, &src);
+            p->offset = src - (char*) p->bytes;
+            if (untilbuf)
+                cmp=(*pp->isamb->method->compare_item)(untilbuf,buf);
+            else
+                cmp=-2;
+            logf(LOG_DEBUG,"isamb_pp_forward on a leaf. cmp=%d", 
+                              cmp);
+	    (*pp->isamb->method->log_item)(LOG_DEBUG, buf, "");
+
+            if (cmp <2)
+            {
+                if (untilbuf)
+		{
+		    (*pp->isamb->method->log_item)(LOG_DEBUG, buf, 
+						   "isamb_pp_forward returning 1");
+		}
+                else
+		{
+		    (*pp->isamb->method->log_item)(LOG_DEBUG, buf, 
+						   "isamb_pp_read returning 1 (fwd)");
+		}
+                pp->returned_numbers++;
+                return 1;
+            }
+            else
+                pp->skipped_numbers++;
+        } /* leaf */
+    } /* main loop */
+}
+
+
 int isamb_pp_num (ISAMB_PP pp)
 {
     return 1;