xref: /dports/www/mnogosearch/mnogosearch-3.4.1/src/dbmode-blob.c

/* Copyright (C) 2000-2015 Lavtech.com corp. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#include "udm_config.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>

#include "udm_common.h"
#include "udm_utils.h"
#include "udm_db.h"
#include "udm_db_int.h"
#include "udm_vars.h"
#include "udm_coords.h"
#include "udm_log.h"
#include "udm_hash.h"
#include "udm_word.h"
#include "udm_crc32.h"
#include "udm_http.h"
#include "udm_contentencoding.h"
#include "udm_parsehtml.h"
#include "udm_store.h"
#include "udm_textlist.h"
#include "udm_indexcache.h"
#include "udm_result.h"
#include "udm_agent.h"

#ifdef HAVE_SQL


#ifdef WIN32
#define UDM_DEFAULT_ZINT4   UDM_TRUE
#define UDM_DEFAULT_DEFLATE UDM_TRUE
#else
#define UDM_DEFAULT_ZINT4   UDM_FALSE
#define UDM_DEFAULT_DEFLATE UDM_FALSE
#endif


/********** Record encoding and compression ************/
/*
  Record compression types:
  - 0x01 = Deflate
  - 0x02 = Zint4
  - 0x03 = Zint4+Deflate
  - 0x04 = Single URL_ID range
*/

#define UDM_BLOB_COMP_NONE                            0x00
#define UDM_BLOB_COMP_DEFLATE                         0x01
#define UDM_BLOB_COMP_ZINT4                           0x02
#define UDM_BLOB_COMP_ZINT4_DEFLATE                   0x03
#define UDM_BLOB_COMP_SINGLE_RANGE                    0x04
#define UDM_BLOB_COMP_URLID_DELTA_VARIABLE            0x05
#define UDM_BLOB_COMP_URLID_DELTA_2BYTES              0x06
#define UDM_BLOB_COMP_URLID_DELTA_2BYTES_WITH_OFFSET  0x07
#define UDM_BLOB_COMP_URLID_DELTA_3BYTES              0x08
#define UDM_BLOB_COMP_URLID_DELTA_1BYTES_WITH_OFFSET  0x09
#define UDM_BLOB_COMP_URLID_RANGE_MULTI               0x0A


static inline int
UdmDSTRAppendCompressionType(UDM_DSTR *dstr, int compression_type)
{
  if (!UdmDSTRAppendINT4(dstr, 0xFFFFFFFF))
    return 0;
  return (compression_type <= UDM_BLOB_COMP_ZINT4_DEFLATE) ?
         UdmDSTRAppendINT4(dstr, compression_type) :
         UdmDSTRAppendINT2BE(dstr, compression_type);
}


/*******************************************/

/*
  If can do "indexer -Eblob" using RENAME TABLE.
*/
static int
UdmBlobCanDoRename(UDM_DB *db)
{
  return
    (UdmSQLDBFlags(db) & UDM_SQL_HAVE_RENAME) &&
    (UdmSQLDBFlags(db) & UDM_SQL_HAVE_CREATE_LIKE) &&
    /* PgSQL can do RENAME only when "DROP TABLE IF EXISTS" is supported */
    (UdmSQLDBType(db)!= UDM_DB_PGSQL || UdmSQLDBFlags(db) & UDM_SQL_HAVE_DROP_IF_EXISTS);
}


static udm_rc_t
UdmBlobCreateIndexRandomName(UDM_AGENT *A, UDM_DB *db, const char *table_name)
{
  char qbuf[128];
  /* Create index with an unique name */
  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
  {
    udm_snprintf(qbuf, sizeof(qbuf),
                 "ALTER TABLE %s ADD KEY (word)", table_name);
  }
  else
  {
    udm_snprintf(qbuf, sizeof(qbuf),
                "CREATE INDEX bdict_%d_%d ON %s (word)",
                 (int) time(0), (int) (UdmStartTimer() % 0xFFFF), table_name);
  }
  return UdmDBSQLQuery(A, db, NULL, qbuf);
}


static int
UdmBlobGetTable(UDM_AGENT *A, UDM_DB *db)
{
  UDM_SQLRES SQLRes;
  int rc;
  const char *val;

  return(1);

  if (UDM_OK != UdmDBSQLQuery(A, db, &SQLRes, "SELECT n FROM bdictsw"))
    return(1);

  if (! UdmSQLNumRows(&SQLRes) || ! (val = UdmSQLValue(&SQLRes, 0, 0))) rc = 2;
  else if (*val != '1') rc = 3;
  else rc = 4;

  UdmSQLFree(&SQLRes);
  return(rc);
}


/*
  This function returns "bdict" by default,
  or the "bdict" parameter from DBAddr, if exists.
*/
static const char *
UdmBlobGetTableNamePrefix(UDM_DB *db)
{
  return UdmVarListFindStr(UdmSQLDBVars(db), "bdict", "bdict");
}


/*
  This function is used when "indexer -Erewritelimit"
  or "indexer -Erewriteurl" is called.
*/
static size_t
UdmBlobGetTableForRewrite(UDM_AGENT *A, UDM_DB *db, char *dst, size_t dstlen)
{
  const char *prefix= UdmBlobGetTableNamePrefix(db);
  return udm_snprintf(dst, dstlen, "%s", prefix);
}


static size_t
UdmBlobGetRTable(UDM_AGENT *A, UDM_DB *db, char *dst, size_t dstlen)
{
  const char *prefix= UdmBlobGetTableNamePrefix(db);
  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
    return udm_snprintf(dst, dstlen, "%s", prefix);
  if (UdmBlobGetTable(A, db) == 3)
    return udm_snprintf(dst, dstlen, "%s00", prefix);
  return udm_snprintf(dst, dstlen, "%s", prefix);
}


static udm_rc_t
UdmBlobGetWTable(UDM_AGENT *A, UDM_DB *db, char *name, size_t namelen)
{
  udm_rc_t rc;

  if (UdmBlobCanDoRename(db))
  {
    if ((UDM_OK != (rc= UdmDBSQLDropTableIfExists(A, db, "bdict_tmp"))) ||
        (UDM_OK != (rc= UdmDBSQLCopyStructure(A, db, "bdict", "bdict_tmp"))) ||
        (UDM_OK != (rc= UdmBlobCreateIndexRandomName(A, db, "bdict_tmp"))))
      return rc;
    udm_snprintf(name, namelen, "bdict_tmp");
    return UDM_OK;
  }

  udm_snprintf(name, namelen, "%s", UdmBlobGetTableNamePrefix(db));
  if (UdmBlobGetTable(A, db) == 4)
    udm_snprintf(name, namelen, "%s00", UdmBlobGetTableNamePrefix(db));
  return UDM_OK;
}


static udm_rc_t
UdmBlobSetTable(UDM_AGENT *A, UDM_DB *db)
{
  char qbuf[128];
  udm_rc_t rc;
  int t, n;
  const char *table_name= UdmVarListFindBool(&A->Conf->Vars, "delta", UDM_FALSE) ?
                          "bdict_delta" :
                          UdmBlobGetTableNamePrefix(db);

  if (UdmBlobCanDoRename(db))
  {
    if (UDM_OK == (rc= UdmDBSQLDropTableIfExists(A, db, table_name)))
      rc= UdmDBSQLRenameTable(A, db, "bdict_tmp", table_name);
    return rc;
  }

  t= UdmBlobGetTable(A, db);
  if (t == 1) return(UDM_OK);
  else if (t == 4) n = 0;
  else n = 1;

  rc= UdmDBSQLQuery(A, db, NULL, "DELETE FROM bdictsw");
  if (rc != UDM_OK) return(UDM_OK);
  udm_snprintf(qbuf, sizeof(qbuf), "INSERT INTO bdictsw VALUES(%d)", n);
  rc = UdmDBSQLQuery(A, db, NULL, qbuf);
  if (rc != UDM_OK) return(UDM_OK);
  return(UDM_OK);
}


typedef struct udm_blob_cache_stat_st
{
  urlid_t cur_url_id;
  urlid_t min_url_id;
  urlid_t max_url_id;
  urlid_t max_url_id_delta;
  urlid_t range_min_url_id;
  urlid_t range_max_url_id;
  size_t ndistinct_url_ids_minus_one;
  size_t compression_type;
  size_t range_offset;
  size_t nranges;
} UDM_BLOB_CACHE_WORD_STAT;


static inline void
udm_put_int2(int i, unsigned char *dst)
{
  dst[0]= (unsigned char) (i & 0xFF);
  dst[1]= (unsigned char) (i >> 8);
}

static inline int
udm_get_int2(const unsigned char *src)
{
  return ((int) src[0]) + (((int) src[1]) << 8);
}


static inline void
udm_put_int3(int i, unsigned char *dst)
{
  dst[0]= (unsigned char) (i & 0xFF);
  dst[1]= (unsigned char) ((i >> 8) & 0xFF);
  dst[2]= (unsigned char) ((i >> 16) & 0xFF);
}

static inline int
udm_get_int3(const unsigned char *src)
{
  return ((int) src[0]) + (((int) src[1]) << 8) + (((int) src[2]) << 16);
}


/*
  Pack integer using various formats:
  0 = don't put anything
  1 = one byte
  2 = two bytes
  3 = three bytes
  4 = four bytes
  5 = variable length encoding

  dst must have enough space to store the packed integer.
*/
static inline size_t
udm_put_int_with_format(int i, unsigned char *dst, int format)
{
  if (format == 1)
  {
    UDM_ASSERT((unsigned int) i < 256);
    dst[0]= (char) (unsigned char) i;
    return 1;
  }
  else if (format == 2)
  {
    udm_put_int2(i, (unsigned char*) dst);
    return 2;
  }
  else if (format == 3)
  {
    udm_put_int3(i, (unsigned char*) dst);
    return 3;
  }
  else if (format == 4)
  {
    udm_put_int4(i, (unsigned char*) dst);
    return 4;
  }
  else if (format == 5)
  {
    return udm_coord_put(i, dst, dst + 4);
  }
  UDM_ASSERT(format == 0);
  return 0;
}


/*
  Unpack coords when only minpos, maxpos and seclen is of interest
*/
static UDM_COORD2 *
UdmBlobPackedCoordsUnpackMinMaxLen(const unsigned char *s,
                                   const unsigned char *e,
                                   size_t nrecs,
                                   UDM_COORD2 *C,
                                   UDM_COORD2 *Coord,
                                   const unsigned char **end,
                                   int save_section_size,
                                   UDM_SEARCHSECTION *Section)
{
  size_t crd, nbytes;
  if (save_section_size)
  {
    if (nrecs > 1)
    {
      s= udm_coord_sum(&crd, s, e, nrecs - 1); /* Sum middle coords */
      C->pos+= crd;
      Section->maxpos= C->pos;
    }
    else
      Section->maxpos= C->pos; /* One coord, minpos=maxpos */

    if ((nbytes= udm_coord_get(&crd, s, e))) /* Get seclen */
    {
      s+= nbytes;
      C->pos+= crd;
      Section->seclen= C->pos;
      Section->ncoords= nrecs;
      Coord+= nrecs;
    }
    else
    {
      Section->seclen= 0;
      Section->ncoords= 0;
    }
  }
  else
  {
    s= udm_coord_sum(&crd, s, e, nrecs);      /* Sum middle coords  */
    C->pos+= crd;
    Section->maxpos= C->pos;
    Section->seclen= 0;
    Section->ncoords= nrecs + 1;
    Coord+= nrecs + 1;
  }
  *end= s;
  return Coord;
}



static size_t
UdmBlobCoordsGetCompressionType(UDM_AGENT *A,
                                UDM_BLOB_CACHE_WORD_STAT *Stat,
                                const unsigned char *s,
                                size_t length)
{
  size_t header_size;
  bzero((void*) Stat, sizeof(*Stat));
  Stat->compression_type= (length > 10 && udm_get_int4(s) == 0xFFFFFFFF) ?
                           udm_get_int4(s + 4) : 0;
  header_size= (Stat->compression_type & 0xFFFF0000) ? 6 : 8;
  Stat->compression_type&= 0x0000FFFF;
  /*
  fprintf(stderr, "Hdr size=%d cmpr=%08X len=%d\n", header_size, Stat->compression_type, length);
  */
  if (Stat->compression_type == UDM_BLOB_COMP_SINGLE_RANGE)
  {
    Stat->min_url_id= udm_get_int4(s + header_size);
    Stat->max_url_id= udm_get_int4(s + header_size + 4);
    Stat->cur_url_id= Stat->min_url_id;
    UdmLog(A, UDM_LOG_DEBUG,
           "Single-URLID-Range compression: %d-%d (%d docs)",
           Stat->min_url_id, Stat->max_url_id,
           Stat->max_url_id - Stat->min_url_id + 1);
    return header_size + 8;
  }
  else if (Stat->compression_type == UDM_BLOB_COMP_URLID_DELTA_VARIABLE)
  {
    UdmLog(A, UDM_LOG_DEBUG, "URLID-Delta compression");
    return header_size;
  }
  else if (Stat->compression_type == UDM_BLOB_COMP_URLID_DELTA_2BYTES)
  {
    UdmLog(A, UDM_LOG_DEBUG, "URLID-Delta-2bytes compression");
    return header_size;
  }
  else if (Stat->compression_type == UDM_BLOB_COMP_URLID_DELTA_2BYTES_WITH_OFFSET)
  {
    Stat->cur_url_id= udm_get_int2(s + header_size) << 16;
    UdmLog(A, UDM_LOG_DEBUG,
           "URLID-Delta-2bytes-offset compression, offs=%d", Stat->cur_url_id);
    return header_size + 2;
  }
  else if (Stat->compression_type == UDM_BLOB_COMP_URLID_DELTA_1BYTES_WITH_OFFSET)
  {
    Stat->cur_url_id= udm_get_int4(s + header_size);
    UdmLog(A, UDM_LOG_DEBUG,
           "URLID-Delta-1byte-offset compression, offs=%d", Stat->cur_url_id);
    return header_size + 4;
  }
  else if (Stat->compression_type == UDM_BLOB_COMP_URLID_DELTA_3BYTES)
  {
    UdmLog(A, UDM_LOG_DEBUG, "URLID-Delta-2bytes compression");
    return header_size;
  }
  else if (Stat->compression_type == UDM_BLOB_COMP_URLID_RANGE_MULTI)
  {
    Stat->nranges= udm_get_int4(s + header_size);
    Stat->range_offset= header_size + 4;
    UdmLog(A, UDM_LOG_DEBUG,
           "URLID-Range-Multi compression, nranges=%d", (int) Stat->nranges);
    return header_size + 4 + 8 * Stat->nranges;
  }
  else if (Stat->compression_type != UDM_BLOB_COMP_NONE)
  {
    UdmLog(A, UDM_LOG_DEBUG,
           "Unknown coompression type: %08X", (int) Stat->compression_type);
  }
  return 0;
}


static size_t
UdmExpectedSectionCount(size_t compression_type, size_t length)
{
  switch (compression_type)
  {
    case UDM_BLOB_COMP_NONE:
      /*
        Shortest section with no compression is 6 bytes:
        - 4 bytes for URL id
        - 1 byte for "ncoords"
        - 1 byte for coord
      */
      return length / 6;
    case UDM_BLOB_COMP_URLID_DELTA_VARIABLE:
      /*
        - 1..4 bytes URL id delta
        - 1 byte for "ncoordss"
        - 1 byte for coord
      */
      return length / 3;

    case UDM_BLOB_COMP_SINGLE_RANGE:
    case UDM_BLOB_COMP_URLID_RANGE_MULTI:
      /*
        Shortest section with single-range compression is 2 byest:
        - 1 byte for ncoords
        - 1 byte for coord
      */
      return length / 2;
    case UDM_BLOB_COMP_URLID_DELTA_1BYTES_WITH_OFFSET:
      /* 1 byte url_id, 1 byte ncoorrds, 1 byte coords */
      return length / 3;
    case UDM_BLOB_COMP_URLID_DELTA_2BYTES:
    case UDM_BLOB_COMP_URLID_DELTA_2BYTES_WITH_OFFSET:
      /* 2 bytes url_id, 1 byte ncoords, 1 byte coord */
      return length / 4;
    case UDM_BLOB_COMP_URLID_DELTA_3BYTES:
      /* 3 bytes url_id, 1 bbyte ncoords, 1 byte coord */
      return length / 5;
  }

  UDM_ASSERT(0); /* Should not normally get here */
  return length;
}


static udm_rc_t
UdmSectionListBlobCoordsUnpack(UDM_AGENT *A,
                               UDM_SEARCHSECTIONLIST *SectionList,
                               UDM_URLID_LIST *urls,
                               UDM_SEARCHSECTION *SectionTemplate,
                               const unsigned char *s,
                               size_t length,
                               int save_section_size,
                               int need_coords)
{
  size_t ncoords= 0, nurls= urls->nurls;
  const unsigned char *s0= s;
  const unsigned char *e= s + length;
  const unsigned char *last_urlid_start= e - sizeof(urlid_t) - 1;
  UDM_COORD2 C, *Coord;
  unsigned char secno= SectionTemplate->secno;
  unsigned char wordnum= SectionTemplate->wordnum;
  unsigned char order= SectionTemplate->order;
  UDM_SEARCHSECTION *Section;
  UDM_BLOB_CACHE_WORD_STAT Stat;
  const unsigned char *range_ptr;
  int single_range, range_multi, compr_urlid_delta;
  int compr_urlid_delta_1bytes, compr_urlid_delta_2bytes, compr_urlid_delta_3bytes;
  size_t coords_alloced, sections_alloced;
  s+= UdmBlobCoordsGetCompressionType(A, &Stat, s, length);
  range_ptr= s0 + Stat.range_offset;

  UdmLog(A, UDM_LOG_DEBUG+1, "Secno=%d len=%d",
         SectionTemplate->secno, (int) length);
  single_range= (Stat.compression_type == UDM_BLOB_COMP_SINGLE_RANGE);
  range_multi= (Stat.compression_type == UDM_BLOB_COMP_URLID_RANGE_MULTI);
  compr_urlid_delta= (Stat.compression_type == UDM_BLOB_COMP_URLID_DELTA_VARIABLE);
  compr_urlid_delta_2bytes= (Stat.compression_type == UDM_BLOB_COMP_URLID_DELTA_2BYTES) ||
                            (Stat.compression_type == UDM_BLOB_COMP_URLID_DELTA_2BYTES_WITH_OFFSET);
  compr_urlid_delta_3bytes= Stat.compression_type == UDM_BLOB_COMP_URLID_DELTA_3BYTES;
  compr_urlid_delta_1bytes= Stat.compression_type == UDM_BLOB_COMP_URLID_DELTA_1BYTES_WITH_OFFSET;

  if (single_range || range_multi)
    last_urlid_start= e - 1; /* TODO: check other modes */

  coords_alloced= length;
  sections_alloced= UdmExpectedSectionCount(Stat.compression_type, length);
  UdmSearchSectionListAlloc(SectionList, coords_alloced, sections_alloced);
  Coord= SectionList->Coord;
  Section= SectionList->Section;

  /*
    A non-compressed chunk consists of:
    - sizeof(urlid_t)
    - at least one byte for length
  */
  for (C.order= order ; s < last_urlid_start; )
  {
    int active= 1;
    size_t nrecs;

    Section->Coord= Coord;
    Section->secno= secno;
    if (Stat.compression_type)
    {
      if (single_range)
      {
        Section->url_id= Stat.cur_url_id++;
      }
      else if (range_multi)
      {
        if (Stat.cur_url_id == Stat.range_max_url_id && Stat.nranges--)
        {
          Stat.range_min_url_id= udm_get_int4(range_ptr); range_ptr+= 4;
          Stat.range_max_url_id= udm_get_int4(range_ptr); range_ptr+= 4;
          Stat.cur_url_id= Stat.range_min_url_id;
          /*
          fprintf(stderr, "loading range: %d-%d nranges remain: %d\n",
                  Stat.range_min_url_id, Stat.range_max_url_id, Stat.nranges);
          */
        }
        else
        {
          Stat.cur_url_id++;
        }
        Section->url_id= Stat.cur_url_id;
      }
      else if (compr_urlid_delta)
      {
        size_t delta, nbytes= udm_coord_get(&delta, s, e);
        s+= nbytes;
        if (!nbytes)
          break;
        Stat.cur_url_id+= delta;
        Section->url_id= Stat.cur_url_id;
      }
      else if (compr_urlid_delta_1bytes)
      {
        size_t delta= (unsigned char) *s;
        s++;
        Stat.cur_url_id+= delta;
        Section->url_id= Stat.cur_url_id;
        /*
        fprintf(stderr, "url_id=%d delta=%d\n", Stat.cur_url_id, delta);
        */
      }
      else if (compr_urlid_delta_2bytes)
      {
        size_t delta= udm_get_int2(s);
        s+= 2;
        Stat.cur_url_id+= delta;
        Section->url_id= Stat.cur_url_id;
        /*
        fprintf(stderr, "url_id=%d delta=%d\n", Stat.cur_url_id, delta);
        */
      }
      else if (compr_urlid_delta_3bytes)
      {
        size_t delta= udm_get_int3(s);
        s+= 3;
        Stat.cur_url_id+= delta;
        Section->url_id= Stat.cur_url_id;
        /*
        fprintf(stderr, "url_id=%d delta=%d\n", Stat.cur_url_id, delta);
        */
      }
    }
    else
    {
      Section->url_id= (urlid_t) udm_get_int4(s);
      s+= 4;
    }
    Section->wordnum= wordnum;
    Section->order= order;

    if (nurls)
    {
      void *found= UdmBSearch(&Section->url_id, urls->urls, urls->nurls,
                              sizeof(urlid_t), (udm_qsort_cmp)UdmCmpURLID);
      if (found && urls->exclude)
        active= 0;
      if (!found && !urls->exclude)
        active= 0;
    }

    /* Get number of coords */
    if (*s < 128)
    {
      nrecs= *s++;
    }
    else
    {
      size_t nbytes= udm_coord_get(&nrecs, s, e);
      if (!nbytes) break;
      s+= nbytes;
    }

    if (!nrecs) /* extra safety */
      break;

    if (!active)
    {
      s= udm_coord_skip(s, e, nrecs);
      continue;
    }

    ncoords+= nrecs;

    if (save_section_size && nrecs > 1)
      ncoords--;

    Section->PackedCoord= s;

    /* Get first coord and put into S->minpos */
    if (*s < 128)
    {
      C.pos= *s++;
    }
    else
    {
      size_t crd;
      size_t nbytes= udm_coord_get(&crd, s, e);
      if (!nbytes) break;
      s+= nbytes;
      C.pos= crd;
    }

    Section->minpos= C.pos;

    /*
      If no coords anymore.
      Maybe the "section length" record didn't fit
      into "64K words per section" limit.
      Add section with seclen=0.
    */
    if (!--nrecs)
    {
      Section->seclen= C.pos;
      Section->ncoords= 1;
      Section->maxpos= C.pos;
      if (need_coords)
        *Coord= C;
      Coord++;
      Section++;
      continue;
    }

    if (!need_coords) /* Does not need coords, e.g. one word search */
    {
      Coord= UdmBlobPackedCoordsUnpackMinMaxLen(s, e, nrecs, &C, Coord, &s,
                                                save_section_size, Section);
      Section++;
      continue;
    }

    *Coord++= C; /* Add first coord */

    /* Unpack the other coordinates */
    Coord= UdmPackedCoordsToUnpackedCoords(s, e, nrecs, &C, Coord, &s);

    /* Set section length */
    nrecs= Coord - Section->Coord;
    if (save_section_size)
    {
      /*
        We need to check whether Coord > Coord0 in the above
        condition: URL could be skipped because of limit
      */
      Section->seclen= ((--Coord)->pos);
      Section->ncoords= nrecs - 1;
      Section->maxpos= Coord[-1].pos;;
    }
    else
    {
      Section->seclen= 0;
      Section->ncoords= nrecs;
      Section->maxpos= C.pos;
    }
    Section++;
  }

  SectionList->ncoords= ncoords;
  SectionList->nsections= Section - SectionList->Section;

  UDM_ASSERT(SectionList->ncoords <= coords_alloced);
  UDM_ASSERT(SectionList->nsections <= sections_alloced);

  return UDM_OK;
}


static udm_rc_t
UdmDeleteWordsFromURLBlob(UDM_AGENT *Indexer, UDM_DB *db, urlid_t url_id)
{
  return UDM_OK;
}


static udm_rc_t
UdmStoreWordsBlob(UDM_AGENT *Indexer, UDM_DB *db, UDM_DOCUMENT *Doc)
{
  return UDM_OK;
}


static const char*
UdmBlobModeInflateOrSelf(UDM_AGENT *A,
                         UDM_DSTR *buf, const char *name,
                         const char *src, size_t *len)
{
  int use_zint4;
  int use_deflate;
  /* fprintf(stderr, "here, len=%d src=%p\n", *len, src); */
  if (!src || *len < 8 ||
      (unsigned char)src[0] != 0xFF ||
      (unsigned char)src[1] != 0xFF ||
      (unsigned char)src[2] != 0xFF ||
      (unsigned char)src[3] != 0xFF ||
      (src[4] != 1 && src[4] != 2 && src[4] != 3) ||
      src[5] != 0x00 ||
      src[6] != 0x00 ||
      src[7] != 0x00)
    return src;
  use_zint4= src[4] == 2 || src[4] == 3;
  use_deflate= src[4] == 1 || src[4] == 3;
  src+= 8;
  *len-= 8;
  if (name)
    UdmLog(A, UDM_LOG_DEBUG, "Unpacking '%s'", name);
  if (use_deflate)
  {
    udm_timer_t ticks= UdmStartTimer();
    size_t len0= len[0];
    UdmLog(A,UDM_LOG_DEBUG, "Deflate header detected");
    UdmDSTRReset(buf);
    if (UDM_OK == UdmDSTRAppendInflate(buf, src, *len))
    {
      src= UdmDSTRPtr(buf);
      len[0]= UdmDSTRLength(buf);
      UdmLog(A, UDM_LOG_DEBUG, "%d to %d bytes inflated",
                                 (int) len0, (int) len[0]);
    }
    UdmLog(A, UDM_LOG_DEBUG, "Inflating done: %.2f", UdmStopTimer(&ticks));
  }
  if (*len >= 5 && use_zint4)
  {
    udm_timer_t ticks= UdmStartTimer();
    char *zint4_buf= (char*) UdmMalloc(*len);
    UdmLog(A, UDM_LOG_DEBUG, "zint4 header detected (zint4ed data length: %d)",
                             (int) (*len));
    if (! zint4_buf)
    {
      UdmLog(A, UDM_LOG_ERROR, "Malloc failed. Requested %u bytes",
                               (int) (*len));
      return(NULL);
    }
    memcpy(zint4_buf, src, *len);
    if (buf->size_alloced < *len * 7 && UdmDSTRRealloc(buf, *len * 7) != UDM_OK)
    {
      UdmFree(zint4_buf);
      UdmLog(A, UDM_LOG_ERROR, "UdmDSTRRealloc failed. Requested %u bytes",
             (int) (*len * 7));
      return(NULL);
    }
    *len= udm_dezint4(zint4_buf, (int4 *) buf->Val.str, *len) * 4;
    src= UdmDSTRPtr(buf);
    UdmFree(zint4_buf);
    UdmLog(A, UDM_LOG_ERROR, "dezint4ed data length: %d", (int) (*len));
    UdmLog(A, UDM_LOG_ERROR, "dezint4 done: %.2f", UdmStopTimer(&ticks));
  }
  return src;
}


static void
UdmBlobModeInflateOrAlloc(UDM_AGENT *A, UDM_DSTR *buf, const char *name,
                          UDM_STR *row, UDM_CONST_STR *dst)
{
  dst->str= UdmBlobModeInflateOrSelf(A, buf, name, row->str, &row->length);
  if (dst->str == row->str)
  {
    UdmDSTRRealloc(buf, row->length);
    memcpy(buf->Val.str, row->str, row->length);
    buf->Val.length= row->length;
    dst->str= UdmDSTRPtr(buf);
  }
  dst->length= row->length;
}


static udm_rc_t
UdmInflateBlobModeSQLRes(UDM_AGENT *A, UDM_SQLRES *src)
{
  UDM_DSTR ibuf;
  size_t row;
  UdmDSTRInit(&ibuf, 1024);
  for (row= 0; row < src->nRows; row++)
  {
    size_t len= UdmSQLLen(src, row, 1);
    const char *val= UdmSQLValue(src, row, 1);
    const char *iflt;
    iflt= UdmBlobModeInflateOrSelf(A, &ibuf, NULL, val, &len);
    if (iflt != val)
    {
      size_t offs= src->nCols*row + 1;
      UdmFree(src->Items[offs].str);
      src->Items[offs].str= (char*) UdmMalloc(len + 1);
      memcpy(src->Items[offs].str, iflt, len);
      src->Items[offs].length= len;
      src->Items[offs].str[len]= '\0';
    }
  }
  UdmDSTRFree(&ibuf);
  return UDM_OK;
}


static udm_rc_t
UdmAddCollationMatch(UDM_QUERY *Query, UDM_FINDWORD_ARGS *args,
                     const char *word, size_t count)
{
  /*
    If match is not full, then we don't know whether
    the word is a substring or a collation match.
    Let's assume it is a substring, to avoid long
    word lists in $(WE).
  */
  if (args->Word.Param.match_mode == UDM_MATCH_FULL ||
      args->Word.Param.match_mode == UDM_MATCH_RANGE)
  {
    UDM_WIDEWORD_PARAM Param= Query->Res.WWList.Word[args->Word.Param.order].Param;
    Param.origin= UDM_WORD_ORIGIN_COLLATION;
    Param.count= count;
    UdmWideWordListAddLike(&args->CollationMatches, &Param, word);
  }
  return UDM_OK;
}


static udm_rc_t
UdmBlobAddCoords(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                 UDM_FINDWORD_ARGS *args,
                 UDM_SQLRES *SQLRes)
{
  size_t numrows= UdmSQLNumRows(SQLRes);
  size_t i;
  char *wf= args->query_param.wf;
  UDM_URLID_LIST *urls= &args->urls;
  int need_coords= args->query_param.NewVersion ?
                   args->need_coords : (Query->Res.WWList.nwords > 1);
  udm_bool_t save_section_size= args->query_param.SaveSectionSize;
  UDM_SEARCHSECTION Section;

  bzero((void*) &Section, sizeof(Section));
  Section.wordnum= args->Word.Param.order & 0xFF;
  Section.order= Query->Res.WWList.Word[Section.wordnum].Param.order;

  for (i= 0; i < numrows; i++)
  {
    const unsigned char *s= (const unsigned char *)UdmSQLValue(SQLRes, i, 1);
    size_t length= UdmSQLLen(SQLRes, i, 1);
    unsigned char secno= UDM_ATOI(UdmSQLValue(SQLRes, i, 0));
    const char *cmatch= UdmSQLValue(SQLRes, i, 2);
    UDM_SEARCHSECTIONLIST SectionList;

    if (!wf[secno])
      continue;

    Section.secno= secno;

    UdmSectionListBlobCoordsUnpack(A, &SectionList,
                                   urls, &Section,
                                   s, length,
                                   save_section_size,
                                   need_coords);

#ifdef HAVE_DEBUG
    if (UdmVarListFindBool(UdmSQLDBVars(db), "DebugSectionList", UDM_FALSE))
      UdmSearchSectionListPrint(&SectionList);
#endif

    if (SectionList.nsections && SectionList.ncoords)
    {
      UdmSearchSectionListListAdd(&args->SearchSectionListList, &SectionList);
      args->Word.Param.count+= SectionList.ncoords;
      UdmAddCollationMatch(Query, args, cmatch, SectionList.ncoords);
    }
    else
    {
      UdmSearchSectionListFree(&SectionList);
    }
  }

  return UDM_OK;
}


static udm_rc_t
UdmFindWordBlobFromTable(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                         UDM_FINDWORD_ARGS *args, const char *table_name)
{
  char qbuf[4096];
  char secno[32]= "";
  char special[32]= "";
  udm_timer_t ticks;
  UDM_SQLRES SQLRes;
  udm_rc_t rc;

  if (args->urls.empty)
  {
    UdmLog(A, UDM_LOG_DEBUG,
           "Not searching '%s': Base URL limit is empty", table_name);
    return UDM_OK;
  }

  ticks= UdmStartTimer();
  UdmLog(A, UDM_LOG_DEBUG, "Start fetching");
  if (args->Word.Param.secno)
    udm_snprintf(secno, sizeof(secno), " AND secno=%d", (int) args->Word.Param.secno);
  /*
    When performing substring or number search,
    don't include special data, like '##last_mod_time' or '##rec_id'
  */
  if (args->Word.Param.match_mode != UDM_MATCH_FULL)
    udm_snprintf(special, sizeof(special), " AND word NOT LIKE '##%%'");
  udm_snprintf(qbuf, sizeof(qbuf),
               "SELECT secno,coords,word FROM %s WHERE %s%s%s",
               table_name, args->cmparg, secno, special);
  if (UDM_OK != (rc= UdmDBSQLQuery(A, db, &SQLRes, qbuf)))
    return rc;
  UdmLog(A, UDM_LOG_DEBUG, "%-30s%.2f", "Stop  fetching", UdmStopTimer(&ticks));

  ticks= UdmStartTimer();
  UdmLog(A, UDM_LOG_DEBUG, "Start BlobAddCoords");
  UdmInflateBlobModeSQLRes(A, &SQLRes);
  UdmBlobAddCoords(A, db, Query, args, &SQLRes);
  UdmLog(A, UDM_LOG_DEBUG, "%-30s%.2f", "Stop  BlobAddCoords", UdmStopTimer(&ticks));
  if (args->query_param.NewVersion)
  {
    UdmSQLResListAdd(&args->SQLResults, &SQLRes);
  }
  else
  {
    UdmSQLFree(&SQLRes);
  }
  return(UDM_OK);
}


static udm_rc_t
UdmFindWordBlobSimple(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                      UDM_FINDWORD_ARGS *args)
{
  char tablename[64];
  udm_rc_t rc;
  int delta= UdmVarListFindBool(&A->Conf->Vars, "UseDelta", UDM_FALSE);
  UdmBlobGetRTable(A, db, tablename, sizeof(tablename));
  if (UDM_OK != (rc= UdmFindWordBlobFromTable(A, db, Query, args, tablename)))
    return rc;
  if (delta && UDM_OK != (rc= UdmFindWordBlobFromTable(A, db, Query,
                                                       args, "bdict_delta")))
    return rc;
  return UDM_OK;
}


static udm_rc_t
UdmLoadSlowLimitWithSort(UDM_AGENT *A, UDM_DB *db, UDM_URLID_LIST *list, const char *q)
{
  udm_rc_t rc= UdmLoadSlowLimit(A, db, list, q);
  if (rc == UDM_OK)
    UdmURLIdListSort(list);
  return rc;
}


static udm_rc_t
UdmBlobLoadLiveUpdateLimitLoad(UDM_AGENT *A, UDM_DB *db, UDM_FINDWORD_ARGS *args)
{
  udm_rc_t rc;
  int ts= 0;
  udm_timer_t ticks;
  char qbuf[128];
  UDM_ASSERT(UdmSQLDBMode(db) == UDM_SQLDBMODE_BLOB);
  ticks= UdmStartTimer();
  UdmLog(A, UDM_LOG_DEBUG, "Start loading LiveUpdate url_id list");
  if (UDM_OK != (rc= UdmBlobReadTimestamp(A, db, &ts, 0)))
    return rc;
  args->live_updates_ts= ts;
  udm_snprintf(qbuf, sizeof(qbuf),
               "SELECT url_id FROM cachedcopy WHERE ts>=%d", ts);
  if (UDM_OK != (rc= UdmLoadSlowLimitWithSort(A, db,
                                              &args->live_update_deleted_urls,
                                              qbuf)))
    return rc;
  UdmLog(A, UDM_LOG_DEBUG,
         "Stop loading LiveUpdate url_id list: %.02f, %d updated docs found",
         UdmStopTimer(&ticks), (int) args->live_update_deleted_urls.nurls);
  args->live_update_deleted_urls.exclude= 1;
  UdmURLIdListCopy(&args->live_update_active_urls, &args->urls);
  UdmURLIdListMerge(&args->urls, &args->live_update_deleted_urls);
  return UDM_OK;
}


static udm_rc_t
UdmFindWordBlobLiveUpdates(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                           UDM_FINDWORD_ARGS *args)
{
  udm_rc_t rc;
  if (!(UDM_OK == (rc= UdmFindWordBlobSimple(A, db, Query, args))) ||
      !(UDM_OK == (rc= UdmFindWordRawBlobDelta(A, db, Query, args))))
    goto ret;

ret:
  return rc;
}


static udm_rc_t
UdmFindWordBlob(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                UDM_FINDWORD_ARGS *args)
{
  return args->live_updates ?
         UdmFindWordBlobLiveUpdates(A, db, Query, args) :
         UdmFindWordBlobSimple(A, db, Query, args);
}


udm_rc_t
UdmBlobReadTimestamp(UDM_AGENT *A, UDM_DB *db, int *ts, int def)
{
  udm_rc_t rc;
  char lname[]= "##ts";
  char qbuf[64], tablename[64];
  UDM_SQLRES SQLRes;

  UdmBlobGetRTable(A, db, tablename, sizeof(tablename));
  udm_snprintf(qbuf, sizeof(qbuf), "SELECT coords FROM %s WHERE word='%s'",
               tablename, lname);
  if (UDM_OK == (rc= UdmDBSQLQuery(A, db, &SQLRes, qbuf)) &&
      UdmSQLNumRows(&SQLRes) > 0)
    *ts= atoi(UdmSQLValue(&SQLRes, 0,0));
  else
    *ts= def;
  UdmSQLFree(&SQLRes);
  return rc;
}


static udm_rc_t
UdmBlobWriteWordPrepare(UDM_AGENT *A, UDM_DB *db, const char *table)
{
  udm_rc_t rc;
  char qbuf[128];
  const char *int_cast= UdmSQLDBType(db) == UDM_DB_PGSQL ? "::integer" : "";
  udm_snprintf(qbuf, sizeof(qbuf),
               "INSERT INTO %s (word,secno,coords) "
               "VALUES(%s, %s%s, %s)",
               table,
               UdmDBSQLParamPlaceHolder(db, 1),
               UdmDBSQLParamPlaceHolder(db, 2),
               int_cast,
               UdmDBSQLParamPlaceHolder(db, 3));
  rc= UdmDBSQLPrepare(A, db, qbuf);
  return rc;
}


#if 0
static udm_bool_t
check_well_formed_length_with_warn(UDM_AGENT *A,
                                   const char *str, size_t length)
{
  size_t wflength= A->Conf->lcs->cset->well_formed_length(A->Conf->lcs,
                                                          str, length,
                                                          UDM_RECODE_HTML);
  if (wflength < length)
  {
    UDM_DSTR tmp;
    UdmDSTRInit(&tmp, 128);
    UdmDSTRAppendHex(&tmp, str + wflength, length - wflength);
    UdmLog(A, UDM_LOG_DEBUG, "Not a well formed word: '%s'", tmp.Val.str);
    UdmDSTRFree(&tmp);
    return UDM_TRUE;
  }
  return UDM_FALSE;
}
#endif


static udm_rc_t
UdmBlobWriteWordUsingBind(UDM_AGENT *A, UDM_DB *db,  const char *table,
                          const char *word, uint4 secno,
                          const char *data, size_t len, UDM_DSTR *buf,
                          int auto_prepare)
{
  udm_rc_t rc;
  size_t wordlen= strlen(word);
  if ((auto_prepare &&
       UDM_OK != (rc= UdmBlobWriteWordPrepare(A, db, table))) ||
       UDM_OK != (rc= UdmDBSQLBindParameter(A, db, 1, word, (int) wordlen, UDM_SQLTYPE_VARCHAR)) ||
       UDM_OK != (rc= UdmDBSQLBindParameter(A, db, 2, &secno, (int) sizeof(secno), UDM_SQLTYPE_INT32)) ||
       UDM_OK != (rc= UdmDBSQLBindParameter(A, db, 3, data, (int) len, UDM_SQLTYPE_LONGVARBINARY) ) ||
       UDM_OK != (rc= UdmDBSQLExecute(A, db)) ||
       (auto_prepare && UDM_OK != (rc= UdmDBSQLStmtFree(A, db))))
    return rc;

  return UDM_OK;
}


static void
UdmDSTREncodeForDB(UDM_AGENT *A, UDM_DB *db, UDM_DSTR *buf,
                   const char *src, size_t length)
{
  if (UdmSQLDBType(db) == UDM_DB_PGSQL)
  {
    char *s= buf->Val.str + buf->Val.length;
    buf->Val.length+= UdmDBSQLBinEscStr(A, db, s, buf->size_alloced, src, length);
  }
  else
    UdmDSTRAppendHex(buf, src, length);
}


static udm_rc_t
UdmBlobWriteWordUsingEncoding(UDM_AGENT *A, UDM_DB *db,  const char *table,
                              const char *word, size_t secno,
                              const char *data, size_t len, UDM_DSTR *buf,
                              int auto_prepare)
{
  udm_rc_t rc;
  size_t escape_factor= UdmSQLDBType(db) == UDM_DB_PGSQL ? 5 : 2;
  const char *pf= UdmSQLDBType(db) == UDM_DB_PGSQL ? "'" : "0x";
  const char *sf= UdmSQLDBType(db) == UDM_DB_PGSQL ? "'" : "";
  const char *E= (UdmSQLDBDriver(db) == UDM_DBAPI_PGSQL && UdmSQLDBVersion(db) >= 80101) ? "E" : "";
  size_t nbytes= 100 + len * escape_factor + 1;

  if (UdmSQLDBFlags(db) & UDM_SQL_HAVE_STDHEX) /* X'AABBCC' syntax */
  {
    pf= "X'";
    sf= "'";
  }
  else if (UdmSQLDBFlags(db) & UDM_SQL_HAVE_BLOB_AS_HEX) /* 'AABBCC' syntax */
  {
    pf= "'";
    sf= "'";
  }

  UdmDSTRReset(buf);

  if (UdmDSTRAlloc(buf, nbytes))
  {
    UdmLog(A, UDM_LOG_ERROR,
           "BlobWriteWordUsingEncoding: DSTRAlloc(%d) failed: "
           "word='%s' secno=%d length=%d",
           (int) nbytes, word, (int) secno, (int) len);
    return UDM_OK; /* Skip this word - try to continue */
  }
  UdmDSTRAppendf(buf, "INSERT INTO %s VALUES('%s', %d, %s%s",
                 table, word, (int) secno, E, pf);
  UdmDSTREncodeForDB(A, db, buf, data, len);
  UdmDSTRAppendf(buf, "%s)", sf);
  if (UDM_OK != (rc= UdmDBSQLQuery(A, db, NULL, UdmDSTRPtr(buf))))
    return rc;

  UdmDSTRReset(buf);

  return UDM_OK;
}


static udm_rc_t
UdmBlobWriteWordUsingMultiInsert(UDM_AGENT *A, UDM_DB *db,  const char *table,
                                 const char *word, size_t secno,
                                 const char *data, size_t len, UDM_DSTR *buf,
                                 int auto_prepare)
{
  const char *comma= ",";
  size_t escape_factor= 2;
  size_t nbytes= UdmDSTRLength(buf) + 100 + len * escape_factor + 1;

  if (UdmDSTRRealloc(buf, nbytes))
  {
    UdmLog(A, UDM_LOG_ERROR, "DSTRAlloc(%d) failed: word='%s' secno=%d len=%d",
            (int) nbytes, word, (int) secno, (int) len);
    return UDM_ERROR;
  }

  if (!UdmDSTRLength(buf))
  {
    UdmDSTRAppendf(buf, "INSERT INTO %s VALUES ", table);
    comma= "";
  }

  UdmDSTRAppendf(buf, "%s('%s',%d,0x", comma, word, (int) secno);
  UdmDSTRAppendHex(buf, data, len);
  UdmDSTRAppendf(buf, ")");
  return UDM_OK;
}


static udm_rc_t
UdmBlobWriteWord(UDM_AGENT *A, UDM_DB *db, const char *table,
                 const char *word, size_t secno,
                 const char *data, size_t len, UDM_DSTR *buf,
                 int auto_prepare, int use_multi_insert)
{
  udm_rc_t rc;
  int use_bind= UdmSQLDBFlags(db) & UDM_SQL_HAVE_BIND_BINARY;

  rc= use_multi_insert ?
      UdmBlobWriteWordUsingMultiInsert(A, db, table, word, secno, data, len,
                                       buf, auto_prepare) :
      use_bind ?
      UdmBlobWriteWordUsingBind(A, db, table, word, secno, data, len,
                                buf, auto_prepare):
      UdmBlobWriteWordUsingEncoding(A, db, table, word, secno, data, len,
                                    buf, auto_prepare);
  return rc;
}


static udm_rc_t
UdmBlobWriteWordCmpr(UDM_AGENT *A, UDM_DB *db, const char *table,
                     const char *word, size_t secno,
                     const char *data, size_t len,
                     UDM_DSTR *buf, UDM_DSTR *z,
                     int use_zint4,
                     int auto_prepare,
                     int allow_multi_insert)
{
#ifdef HAVE_ZLIB
  if (z && len > 256)
  {
    UdmDSTRReset(z);
    UdmDSTRRealloc(z, len + 8 + 1); /* 8 for two INTs */
    /* Append Format version */
#if 0
    if (use_zint4)
    {
      udm_rc_t dummy;
      /* Something is wrong here: why UdmDeflate? */
      UdmDSTRAppendCompressionType(z, UDM_BLOB_COMP_ZINT4_DEFLATE);
      UdmDSTRAppendDeflate(z, data + 8, len - 8);
    }
    else
#endif
    {
      UdmDSTRAppendCompressionType(z, UDM_BLOB_COMP_DEFLATE);
      UdmDSTRAppendDeflate(z, data, len);
    }
    if (UdmDSTRLength(z) < len)
    {
      data= UdmDSTRPtr(z);
      len= UdmDSTRLength(z);
    }
  }
#endif
  return UdmBlobWriteWord(A, db, table, word, secno, data, len, buf,
                          auto_prepare, allow_multi_insert);
}


/*
  The word must not require escaping!
*/
static udm_rc_t
UdmBlobDeleteWordFromTable(UDM_AGENT *A, UDM_DB *db,
                           const char *table, const char *word)
{
  char qbuf[64];
  udm_snprintf(qbuf, sizeof(qbuf),
               "DELETE FROM %s WHERE word='%s'", table, word);
  return UdmDBSQLQuery(A, db, NULL, qbuf);
}


static udm_rc_t
UdmBlobRewriteIntWord(UDM_AGENT *A, UDM_DB *db, UDM_DSTR *buf,
                      const char *table, const char *name, int value,
                      udm_bool_t rewrite)
{
  udm_rc_t rc;
  char data[64];
  size_t size_data;
  UdmLog(A, UDM_LOG_DEBUG, "Writing '%s'", name);
  if (rewrite &&
      UDM_OK != (rc= UdmBlobDeleteWordFromTable(A, db, table, name)))
    return rc;
  size_data= udm_snprintf(data, sizeof(data), "%d", value);
  return UdmBlobWriteWord(A, db, table, name, 0, data, size_data, buf, 1, 0);
}


static udm_rc_t
UdmBlobWriteTimestamp(UDM_AGENT *A, UDM_DB *db,
                      const char *table, udm_bool_t rewrite)
{
  udm_rc_t rc;
  UDM_DSTR buf;
  UdmDSTRInit(&buf, 128);
  if (UDM_OK != (rc= UdmBlobRewriteIntWord(A, db, &buf, table, "##ts",
                                           (int) time(0), rewrite)))
    goto ex;
  rc= UdmBlobRewriteIntWord(A, db, &buf, table, "##version",
                            UDM_VERSION_ID, rewrite);
ex:
  UdmDSTRFree(&buf);
  return rc;
}



static udm_rc_t
UdmTruncateDictBlob(UDM_AGENT *Indexer, UDM_DB *db)
{
  return UdmDBSQLTableTruncateOrDelete(Indexer, db, "bdict");
}

/************************************************/

typedef struct
{
  UDM_DSTR buf;
  UDM_DSTR compress;
  udm_bool_t use_deflate;
} UDM_WRITE_HELPER;


static udm_rc_t
UdmWriteHelperInit(UDM_WRITE_HELPER *Helper, UDM_DB *db)
{
  bzero((void*) Helper, sizeof(*Helper));
#ifdef HAVE_ZLIB
  Helper->use_deflate= UdmVarListFindBool(UdmSQLDBVars(db), "deflate", UDM_FALSE);
#endif
  if (UDM_OK != UdmDSTRInit(&Helper->buf, 8 * 1024))
    return UDM_ERROR;
  if (UDM_OK != UdmDSTRInit(&Helper->compress, 8 * 1024))
  {
    UdmDSTRFree(&Helper->buf);
    return UDM_ERROR;
  }
  return UDM_OK;
}


static void
UdmWriteHelperFree(UDM_WRITE_HELPER *Helper)
{
  UdmDSTRFree(&Helper->buf);
  UdmDSTRFree(&Helper->compress);
}


static udm_rc_t
UdmWriteWordWithHelper(UDM_AGENT *A, UDM_DB *db, const char *table,
                       const char *word, const UDM_DSTR *data,
                       UDM_WRITE_HELPER *Helper, int use_zint4)
{
  return UdmBlobWriteWordCmpr(A, db, table, word, /*secno*/0,
                              UdmDSTRPtr(data), UdmDSTRLength(data),
                              &Helper->buf,
                              Helper->use_deflate ? &Helper->compress : NULL,
                              use_zint4, 1, 0);
}
/*****************************************************/

/*
  Write limits, but don't COMMIT and don't write timestamp
*/
static udm_rc_t
UdmBlobWriteLimitsInternal(UDM_AGENT *A, UDM_DB *db,
                           const char *table, UDM_WRITE_HELPER *Helper)
{
  UDM_VARLIST *Vars= &A->Conf->Vars;
  UDM_DSTR l;
  udm_rc_t rc= UDM_OK;
  size_t nvar;

  UdmDSTRInit(&l, 8192);
  for (nvar= 0; nvar < Vars->nvars; nvar++)
  {
    UDM_VAR *Var= UdmVarListFindByIndex(Vars, nvar);
    size_t i, ndocs;
    char qbuf[128];
    char lname[64];
    UDM_URLID_LIST list;
    UDM_URL_INT4_LIST UserScore;
    int is_score= 0;
    udm_timer_t ticks;

    if (!strncasecmp(UdmVarName(Var), "Limit.", 6))
      udm_snprintf(lname, sizeof(lname), "##limit#%s", UdmVarName(Var) + 6);
    else if (!strncasecmp(UdmVarName(Var), "Order.", 6))
      udm_snprintf(lname, sizeof(lname), "##order#%s", UdmVarName(Var) + 6);
    else if ((is_score= !strncasecmp(UdmVarName(Var), "Score.", 6)))
      udm_snprintf(lname, sizeof(lname), "##score#%s", UdmVarName(Var) + 6);
    else
      continue;
    UdmLog(A, UDM_LOG_DEBUG, "Writing '%s'", lname);

    bzero((void*) &list, sizeof(list));
    bzero((void*) &UserScore, sizeof(UserScore));

    if (UDM_OK != (rc= is_score ?
                       UdmUserScoreListLoad(A, db, &UserScore, UdmVarStr(Var)) :
                       UdmLoadSlowLimit(A, db, &list, UdmVarStr(Var))))
      goto ret;

    ticks= UdmStartTimer();

    if (!strncasecmp(UdmVarName(Var), "Limit.", 6))
      UdmURLIdListSort(&list);

    UdmDSTRReset(&Helper->buf);
    UdmDSTRReset(&l);
    ndocs= is_score ? UserScore.nitems : list.nurls;
    for (i= 0; i < ndocs; i++)
    {
      if (is_score)
      {
        UDM_URL_INT4 *item= &UserScore.Item[i];
        char ch= item->param;
        UdmDSTRAppendINT4(&l, item->url_id);
        UdmDSTRAppend(&l, &ch, 1);
      }
      else
      {
        /* Limit */
        UdmDSTRAppendINT4(&l, list.urls[i]);
      }
    }

    udm_snprintf(qbuf, sizeof(qbuf), "DELETE FROM %s WHERE word=('%s')", table, lname);
    if (UDM_OK != (rc= UdmDBSQLQuery(A, db, NULL, qbuf)))
      goto ret;

    if (UdmDSTRLength(&l))
    {
      if (UDM_OK != (rc= UdmWriteWordWithHelper(A, db, table, lname, &l,
                                                Helper, 0)))
        goto ret;
    }

    UDM_FREE(list.urls);
    UDM_FREE(UserScore.Item);
    UdmLog(A, UDM_LOG_DEBUG, "%d documents written to '%s': %.2f",
                              (int) ndocs, lname, UdmStopTimer(&ticks));
  }
ret:
  UdmDSTRFree(&l);
  return rc;
}


/*******************************************/

static udm_rc_t
UdmURLDataListStorePopularityBdict(UDM_AGENT *A, UDM_DB *db, const char *table,
                                   UDM_URLDATALIST *List,
                                   UDM_WRITE_HELPER *Helper)
{
  udm_rc_t rc= UDM_OK;
  UDM_DSTR pop;

  UdmLog(A, UDM_LOG_DEBUG, "Writing '##pop'");

  /* TODO34: add "rewrite" parameter */
  if (UDM_OK != (rc= UdmBlobDeleteWordFromTable(A, db, table, "##pop")))
    return rc;

  if (UDM_OK != (rc= UdmDSTRInit(&pop, 8192)))
    return UDM_ERROR;

  if (UDM_OK != (rc= UdmURLDataListPackPopularity(A, List, &pop)))
  {
    UdmLog(A, UDM_LOG_ERROR, "Packing popularity failed");
    goto ex;
  }

  if (UdmDSTRLength(&pop))
    rc= UdmWriteWordWithHelper(A, db, table, "##pop", &pop, Helper, 0);
ex:
  UdmDSTRFree(&pop);
  return rc;
}


static udm_rc_t
UdmURLDataListStorePopularityTable(UDM_AGENT *A, UDM_DB *db,
                                   UDM_URLDATALIST *List)
{
  udm_timer_t timer= UdmStartTimer();
  char qbuf[128];
  size_t i;
  const char *table= UdmVarListFindStr(&A->Conf->Vars, "SQLExportPopularityTable", NULL);

  if (!table)
    return UDM_OK;

  /* TODO34: allow only safe table names? */
  UdmLog(A, UDM_LOG_DEBUG, "Writing popularity table '%s'", table);
  if (UDM_OK != UdmDBSQLTableTruncateOrDelete(A, db, table))
    return UDM_ERROR;
  udm_snprintf(qbuf, sizeof(qbuf), "%s WRITE", table);
  if (UDM_OK != UdmDBSQLLockOrBegin(A, db, qbuf))
    return UDM_ERROR;
  for (i= 0; i < List->nitems; i++)
  {
    UDM_URLDATA *Item= &List->Item[i];
    udm_snprintf(qbuf, sizeof(qbuf),
                 "INSERT INTO %s (url_id,inlinks,poprank) VALUES (%d,%d,%f)",
                 table, Item->url_id, Item->per_site, Item->pop_rank);
    if (UDM_OK != UdmDBSQLQuery(A, db, NULL, qbuf))
      return UDM_ERROR;
  }
  if (UDM_OK != UdmDBSQLUnlockOrCommit(A, db))
    return UDM_ERROR;
  UdmLog(A, UDM_LOG_DEBUG, "Writing popularity table done: %.2f", UdmStopTimer(&timer));
  return UDM_OK;
}


static udm_rc_t
UdmWritePopularityBdictAndTable(UDM_AGENT *A, UDM_DB *db,
                                UDM_URLDATALIST *URLData,
                                const char *table,
                                UDM_WRITE_HELPER *Helper)
{
  udm_rc_t rc;
  /* Write pop_rank only if we have some non-empty value */
  if (UDM_OK != (rc= UdmURLDataListStorePopularityBdict(A, db, table, URLData,
                                                        Helper)))
    return rc;
  if (UDM_OK != (rc= UdmURLDataListStorePopularityTable(A, db, URLData)))
    return rc;
  return UDM_OK;
}


static udm_rc_t
UdmBlobWriteURLData(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                    const char *table,
                    UDM_WRITE_HELPER *Helper)
{
  udm_rc_t rc= UDM_OK;
  int use_zint4= UdmVarListFindBool(UdmSQLDBVars(db), "zint4", UDM_DEFAULT_ZINT4);
  size_t i;
  UDM_DSTR url_id, site, last_mod_time;
  UDM_URLDATALIST *List= &Query->URLData;

  UdmDSTRInit(&url_id, 8192);
  UdmDSTRInit(&site, 1024);
  UdmDSTRInit(&last_mod_time, 8192);

  for (i= 0; i < List->nitems; i++)
  {
    UDM_URLDATA *Item= &List->Item[i];
    UdmDSTRAppendINT4(&url_id, Item->url_id);
    UdmDSTRAppendINT4(&last_mod_time, Item->last_mod_time);
  }

  UdmLog(A, UDM_LOG_DEBUG, "Writing '##rec_id'");
  if (UDM_OK != (rc= UdmWriteWordWithHelper(A, db, table, "##rec_id",
                                            &url_id, Helper, use_zint4)))
    goto ex;

  UdmLog(A, UDM_LOG_DEBUG, "Writing '##last_mod_time'");
  if (UDM_OK != (rc= UdmWriteWordWithHelper(A, db, table, "##last_mod_time",
                                            &last_mod_time, Helper, 0)))
    goto ex;

  if (UDM_OK != (rc= UdmURLDataListPackSite(List, &site)))
    goto ex;

  UdmLog(A, UDM_LOG_DEBUG, "Writing '##site'");
  if (UDM_OK != (rc= UdmWriteWordWithHelper(A, db, table, "##site",
                                            &site, Helper, 0)))
    goto ex;

ex:
  UdmDSTRFree(&url_id);
  UdmDSTRFree(&site);
  UdmDSTRFree(&last_mod_time);
  return rc;
}


/*
  Unpack URL data from all packed records:
  '##rec_id'
  '##last_mod_time'
*/
static size_t
UdmURLDataUnpackFull(UDM_URLDATALIST *DataList, size_t nrows,
                     const char *rec_id_str,
                     const char *last_mod_time_str)
{
  size_t i, j;
  for (j= 0, i= 0; i < nrows; i++)
  {
    urlid_t rec_id= udm_get_int4(rec_id_str);
    rec_id_str+= 4;

    if (rec_id == DataList->Item[j].url_id)
    {
      UDM_URLDATA *D= &DataList->Item[j];
      if (last_mod_time_str)
        D->last_mod_time= udm_get_int4(&last_mod_time_str[i*4]);
      j++;
      if (j == DataList->nitems)
        break;
    }
  }
  return j;
}


/*
  Unpack rec_id from '##rec_id' record.
*/
static size_t
UdmURLDataUnpackRecID(UDM_AGENT *A, UDM_URLDATALIST *DataList, size_t nrows,
                      const char *rec_id_str)
{
  /* Need only rec_id */
  UDM_URLDATA *Data= DataList->Item;
  size_t j, i, skip= 0, ncoords= DataList->nitems;

  for (j = 0, i = 0; i < nrows; i++)
  {
    urlid_t rec_id= udm_get_int4(rec_id_str);
    while (rec_id > Data[j].url_id && j < ncoords)
    {
      skip++;
      j++;
    }

    if (rec_id == Data[j].url_id)
    {
      j++;
      if (j == ncoords) break;
    }
    rec_id_str+= 4;
  }
  if (j < ncoords)
  {
    skip+= (ncoords - j);
    UdmLog(A, UDM_LOG_DEBUG,
           "Warning: %d out of %d coords didn't have URL data",
           (int) skip, (int) DataList->nitems);
    j= DataList->nitems;
  }
  return j;
}


static size_t
UdmDSTRAppendWithComma(UDM_DSTR *dstr, const char *s, size_t length)
{
  if (UdmDSTRLength(dstr))
    UdmDSTRAppend(dstr, ",", 1);
  return UdmDSTRAppend(dstr, s, length);
}


udm_rc_t
UdmLoadURLDataFromBdict(UDM_AGENT *A, UDM_DB *db,
                        UDM_URLDATALIST *DataList,
                        int flags)
{
  udm_rc_t rc;
  char qbuf[4*1024], table[64];
  UDM_SQLRES SQLres;
  UDM_DSTR inbuf, rec_id_buf, site_buf, pop_rank_buf, last_mod_time_buf;
  UDM_STR row[2];
  udm_timer_t ticks= UdmStartTimer();
  int need_pop_rank= (flags & UDM_URLDATA_POP);
  int need_last_mod_time= (flags & UDM_URLDATA_LM);
  int need_site= (flags & (UDM_URLDATA_SITE | UDM_URLDATA_SITE_RANK));
  int need_id= need_last_mod_time; /* TODO34: pack last_mode_time toghether with id? */
  UDM_CONST_STR site, id, pop, lm;

  if (!flags)
    return UDM_NOTARGET;

  UdmConstStrInit(&site);
  UdmConstStrInit(&id);
  UdmConstStrInit(&pop);
  UdmConstStrInit(&lm);

  UdmBlobGetRTable(A, db, table, sizeof(table));

  UdmDSTRInit(&inbuf, 64);
  if (need_id)
    UdmDSTRAppendWithComma(&inbuf, UDM_CSTR_WITH_LEN("'##rec_id'"));
  if (need_pop_rank)
    UdmDSTRAppendWithComma(&inbuf, UDM_CSTR_WITH_LEN("'##pop'"));
  if (need_last_mod_time)
    UdmDSTRAppendWithComma(&inbuf, UDM_CSTR_WITH_LEN("'##last_mod_time'"));
  if (need_site)
    UdmDSTRAppendWithComma(&inbuf, UDM_CSTR_WITH_LEN("'##site'"));

  /* Check that DataList is not empty and is sorted by url_id */
  UDM_ASSERT(DataList->nitems);
  UDM_ASSERT(DataList->Item[0].url_id <= DataList->Item[DataList->nitems-1].url_id);

  UdmLog(A,UDM_LOG_DEBUG,"Loading URL data from bdict");
  udm_snprintf(qbuf, sizeof(qbuf),
               "SELECT word,coords FROM %s WHERE word IN (%s)",
               table, UdmDSTRPtr(&inbuf));
  UdmDSTRFree(&inbuf);

  if (UDM_OK != (rc= UdmDBSQLExecDirect(A, db, &SQLres, qbuf)))
  {
    UdmLog(A,UDM_LOG_DEBUG,"Couldn't run a query on bdict");
    return(rc);
  }

  UdmDSTRInit(&rec_id_buf, 4096);
  UdmDSTRInit(&site_buf, 4096);
  UdmDSTRInit(&pop_rank_buf, 4096);
  UdmDSTRInit(&last_mod_time_buf, 4096);

  while (UdmDBSQLFetchRow(A, db, &SQLres, row) == UDM_OK)
  {
    if (!strcmp(row[0].str, "##rec_id"))
      UdmBlobModeInflateOrAlloc(A, &rec_id_buf, "##rec_id", &row[1], &id);
    else if (!strcmp(row[0].str, "##site"))
      UdmBlobModeInflateOrAlloc(A, &site_buf, "##site", &row[1], &site);
    else if (!strcmp(row[0].str, "##last_mod_time"))
      UdmBlobModeInflateOrAlloc(A, &last_mod_time_buf, "##last_mod_time",
                                &row[1], &lm);
    else if (!strcmp(row[0].str, "##pop"))
      UdmBlobModeInflateOrAlloc(A, &pop_rank_buf, "##pop", &row[1], &pop);
  }

  UdmLog(A, UDM_LOG_DEBUG, "Fetch from bdict done:        %.2f", UdmStopTimer(&ticks));

  if (need_pop_rank)
  {
    if (pop.str)
    {
      ticks= UdmStartTimer();
      UdmURLDataListUnpackPopularity(A, DataList, &pop);
      UdmLog(A, UDM_LOG_DEBUG, "Unpacking popularity done:    %.02f", UdmStopTimer(&ticks));
    }
    else
    {
      /*
        All pop_rank values were 0 at "indexer -Eblob" time.
        Use 0 as pop_rank values.
      */
      UdmLog(A, UDM_LOG_DEBUG, "Warning: s=P is requested, but '##pop' record not found");
      UdmLog(A, UDM_LOG_DEBUG, "Perhaps you forgot to run 'indexer -n0 -R' before running 'indexer --index'");
      need_pop_rank= 0;
    }
  }

  if (need_site)
  {
    if (site.str)
    {
      ticks= UdmStartTimer();
      UdmURLDataListUnpackSite(A, DataList, &site);
      UdmLog(A, UDM_LOG_DEBUG, "Unpacking site done: %.02f", UdmStopTimer(&ticks));
    }
    else
    {
      UdmLog(A, UDM_LOG_DEBUG, "No '##site' record found");
    }
  }

  if (!need_id && !need_last_mod_time)
  {
    rc= UDM_OK;
    goto ret;
  }

  if (id.str && id.length &&
      (lm.str || ! need_last_mod_time))
  {
    size_t j, nrows= id.length / 4;

    ticks= UdmStartTimer();
    UdmLog(A, UDM_LOG_DEBUG, "Unpacking URL Data %d rows", (int) nrows);
    if (need_last_mod_time)
    {
      /* Need pop_rank or last_mod_time */
      j= UdmURLDataUnpackFull(DataList, nrows, id.str,
                              need_last_mod_time ? lm.str : NULL);
    }
    else
    {
      /* Need only rec_id */
      j= UdmURLDataUnpackRecID(A, DataList, nrows, id.str);
    }

    UdmLog(A, UDM_LOG_DEBUG, "Unpacking URL Data done: %.02f", UdmStopTimer(&ticks));

    if (j != DataList->nitems)
    {
      UdmLog(A,UDM_LOG_DEBUG, "Expected to load %d URLs, loaded %d",
                              (int) DataList->nitems, (int) j);
      UdmLog(A,UDM_LOG_DEBUG,"Couldn't load URL data from bdict");
      goto load_from_url;
    }
  }
  else
  {
    UdmLog(A,UDM_LOG_DEBUG,"There is no URL data in bdict");
  }


load_from_url:
  rc= UDM_NOTARGET;

ret:
  UdmSQLFree(&SQLres);
  UdmDSTRFree(&rec_id_buf);
  UdmDSTRFree(&site_buf);
  UdmDSTRFree(&pop_rank_buf);
  UdmDSTRFree(&last_mod_time_buf);
  return rc;
}



/*******************************************/

static udm_rc_t
UdmRewriteURL(UDM_AGENT *Indexer, UDM_DB *db, UDM_QUERY *Query)
{
  udm_rc_t rc;
  int tr= (UdmSQLDBFlags(db) & UDM_SQL_HAVE_TRANSACT) ? 1 : 0;
  char tablename[64];
  UDM_WRITE_HELPER Helper;

  if (UDM_OK != (rc= UdmWriteHelperInit(&Helper, db)))
    return rc;
  UdmBlobGetTableForRewrite(Indexer, db, tablename, sizeof(tablename));
  if ((tr && UDM_OK != (rc= UdmDBSQLBegin(Indexer, db))) ||
      UDM_OK != (rc= UdmBlobWriteTimestamp(Indexer, db, tablename, UDM_TRUE)) ||
      UDM_OK != (rc= UdmBlobWriteURLData(Indexer, db, Query, tablename, &Helper)) ||
      UDM_OK != (rc= UdmBlobWriteLimitsInternal(Indexer, db, tablename, &Helper)) ||
      (tr && UDM_OK != (rc= UdmDBSQLCommit(Indexer, db))))
  {
  }
  UdmWriteHelperFree(&Helper);
  return rc;
}


/*
  Write limits with COMMIT and timestamp
*/
static udm_rc_t
UdmBlobWriteLimits(UDM_AGENT *A, UDM_DB *db, const char *table,
                   UDM_WRITE_HELPER *Helper)
{
  udm_rc_t rc;
  if (UDM_OK != (rc= UdmDBSQLBegin(A, db)) ||
      UDM_OK != (rc= UdmBlobWriteLimitsInternal(A, db, table, Helper)) ||
      UDM_OK != (rc= UdmBlobWriteTimestamp(A, db, table, UDM_TRUE)) ||
      UDM_OK != (rc= UdmDBSQLCommit(A, db)))
    return rc;
  return UDM_OK;
}


static udm_rc_t
UdmRewriteLimits(UDM_AGENT *Indexer, UDM_DB *db)
{
  udm_rc_t rc;
  char tablename[64];
  UDM_WRITE_HELPER Helper;
  if (UDM_OK != (rc= UdmWriteHelperInit(&Helper, db)))
    return rc;
  UdmBlobGetTableForRewrite(Indexer, db, tablename, sizeof(tablename));
  rc= UdmBlobWriteLimits(Indexer, db, tablename, &Helper);
  UdmWriteHelperFree(&Helper);
  return rc;
}




/************************************************/

#include "udm_doc.h"
#include "udm_parsehtml.h"
#include "udm_parsexml.h"
#include "udm_server.h"  /* UdmSpiderParamInit */
#include "udm_url.h"

/* TODO34: merge with word.c */
/*
static int wlcmp_sort(UDM_WORD *w1, UDM_WORD *w2)
{
  register int _;
  if ((_= strcmp(w1->word, w2->word)))
    return _;
  if (w1->secno != (int) w2->secno)
    return w1->secno < w2->secno ? -1 : 1;
  if (w1->pos != w2->pos)
    return w1->pos < w2->pos ? -1 : 1;
  return 0;
}
*/
static int wlcmp_search(UDM_WORD *w1, UDM_WORD *w2)
{
  register int _;
  if ((_= strcmp(w1->word, w2->word)))
    return _;
  if (w1->coord.secno != (int) w2->coord.secno)
    return w1->coord.secno < w2->coord.secno ? -1 : 1;
  return 0;
}
/*
static void
UdmWordListSort(UDM_WORDLIST *WL)
{
  if (WL->nwords)
    UdmSort(WL->Word, WL->nwords, sizeof(UDM_WORD), (udm_qsort_cmp) wlcmp_sort);
}
*/

static udm_rc_t
UdmWordList2InvertedIndexCache(UDM_AGENT *A, UDM_WORDLIST *WL,
                               UDM_CONSTWORD_HASH_DATA *data,
                               UDM_INVERTED_INDEX_CACHE *cache)
{
  size_t i;
  UDM_WORD *prev;
  if (!WL->nwords)
    return UDM_OK;
  for (prev= &WL->Word[0], i= 0; i <= WL->nwords; i++)
  {
    UDM_WORD *W= &WL->Word[i];
    if (i == WL->nwords || wlcmp_search(prev, W))
    {
      if (UDM_OK != UdmInvertedIndexCacheAdd(A, data, cache, prev, W - prev))
        return UDM_ERROR;
      prev= W;
    }
  }
  return UDM_OK;
}



static udm_rc_t
UdmConstWordListToPairWordList(UDM_INVERTED_INDEX_CACHE *cache,
                               UDM_WORDLIST *WL,
                               UDM_CONSTWORD_HASH_DATA *data,
                               UDM_INVERTED_INDEX_COORD_LIST *CL)
{
  size_t i;

  for (i= 1; i < CL->nitems; i++)
  {
    UDM_INVERTED_INDEX_COORD *C1= &CL->Item[i-1];
    UDM_INVERTED_INDEX_COORD *C2= &CL->Item[i];
    UDM_INVERTED_INDEX_WORD *W1= &cache->Words.list.Item[C1->word_id];
    UDM_INVERTED_INDEX_WORD *W2= &cache->Words.list.Item[C2->word_id];
    /*
      printf("[%d][%d][%d:%d][%d:%d] '%.*s'-'%.*s' (%d-%d)\n",
             i, (int) data->url_id,
             C1->secno, C1->pos, C2->secno, C2->pos,
             (int) W1->str.length, W1->str.str,
             (int) W2->str.length, W2->str.str,
             W1->last_url_id_count, W2->last_url_id_count);
    */
    /* TODO34: Check max_word_len and min_word_len */
    if (C1->coord.secno == C2->coord.secno &&
        C1->coord.pos + 1 == C2->coord.pos &&
        W1->last_url_id_count > cache->param.pair_limit &&
        W2->last_url_id_count > cache->param.pair_limit)
    {
      char word[256]= "";
      UDM_ASSERT(W1->last_url_id == W2->last_url_id);
      udm_snprintf(word, sizeof(word), "##%.*s-%.*s",
                   (int) W1->length, W1->str,
                   (int) W2->length, W2->str);
      /*
      fprintf(stderr, "[%d:%d:%d] %d-%d '%s'\n",
              data->url_id, C1->secno, C1->pos, W1->last_url_id_count, W2->last_url_id_count,
              word);
      */
      UdmWordListAddEx(WL, word, C1->coord.secno, C1->coord.pos, 0);
    }
  }
  return UDM_OK;
}


static udm_rc_t
UdmConvertPairs(UDM_AGENT *A,
                UDM_INVERTED_INDEX_CACHE *cache,
                UDM_INVERTED_INDEX_COORD_LIST *CL,
                UDM_CONSTWORD_HASH_DATA *hash_data)
{
  udm_rc_t rc= UDM_OK;
  UDM_WORDLIST WL;
  UdmWordListInit(&WL);
  UdmConstWordListToPairWordList(cache, &WL, hash_data, CL);
  if (WL.nwords)
    rc= UdmWordList2InvertedIndexCache(A, &WL, hash_data, cache);
  UdmWordListFree(&WL);
  return rc;
}


static udm_rc_t
UdmWordListSortAndConvert(UDM_AGENT *A,
                          UDM_CONSTWORDLIST *CWL,
                          UDM_INVERTED_INDEX_CACHE *cache,
                          urlid_t url_id,
                          UDM_CHARSET *doccs)
{
  udm_rc_t rc= UDM_OK;
  UDM_INVERTED_INDEX_COORD_LIST CL;
  UDM_CONSTWORD_HASH_DATA hash_data;
  udm_timer_t ticks= UdmStartTimer();

  if (!CWL->nitems)
    return UDM_OK;

  UDM_GETLOCK(A, UDM_LOCK_DOC_CACHE);
  bzero(&CL, sizeof(CL));
  UdmConvInit(&hash_data.cnv, doccs, A->Conf->lcs);
  hash_data.url_id= url_id;
  hash_data.cache= cache;
  cache->Words.hash.user_data= &hash_data;
  cache->Stats.conv+= UdmStartTimer() - ticks;

  ticks= UdmStartTimer();
  rc= UdmConstWordListToInvertedIndexCoordList(cache, &CL, CWL);
  cache->Stats.prepare_words+= UdmStartTimer() - ticks;
  UDM_RELEASELOCK(A, UDM_LOCK_DOC_CACHE);

  if (rc != UDM_OK)
    goto ret;

  UDM_GETLOCK(A, UDM_LOCK_DOC_CACHE);
  if (hash_data.cache->param.pair_limit)
  {
    ticks= UdmStartTimer();
    rc= UdmConvertPairs(A, cache, &CL, &hash_data);
    cache->Stats.pairs+=  UdmStartTimer() - ticks;
  }
  UDM_RELEASELOCK(A, UDM_LOCK_DOC_CACHE);

  ticks= UdmStartTimer();
  UdmInvertedIndexCoordListSort(&CL);

  UDM_GETLOCK(A, UDM_LOCK_DOC_CACHE);
  cache->Stats.sort_wordlist+= UdmStartTimer() - ticks;
  ticks= UdmStartTimer();
  rc= UdmInvertedIndexCoordList2InvertedIndexCache(A, &CL, CWL, &hash_data, cache);
  cache->Stats.conv+= UdmStartTimer() - ticks;
  UDM_RELEASELOCK(A, UDM_LOCK_DOC_CACHE);

  if (rc != UDM_OK)
    goto ret;

ret:
  ticks= UdmStartTimer();
  UdmInvertedIndexCoordListFree(&CL);
  cache->Stats.conv+= UdmStartTimer() - ticks;
  return rc;
}


static udm_rc_t
UdmInvertedIndexCacheDocParse(UDM_AGENT *Indexer,
                              urlid_t url_id,
                              UDM_DOCUMENT *Doc,
                              UDM_INVERTED_INDEX_CACHE *cache)
{
  UDM_CONSTWORDLIST CWL;
  UDM_CHARSET *doccs= UdmDocDetectCachedCharset(Indexer, Doc, url_id);
  UDM_CHARSET *metacs;
  udm_rc_t rc;

  UdmConstWordListInit(&CWL);
  UdmDocToConstWordList(Indexer, Doc, &CWL, doccs,
                        cache->param.cnvflags,
                        cache->param.aggregate_section_flags,
                        &cache->Stats.parse,
                        &cache->Stats.prepare_words);
  metacs= UdmVarListFindCharset(&Doc->Sections, "Strong-Meta-Charset", doccs);
  rc= UdmWordListSortAndConvert(Indexer, &CWL, cache, url_id, metacs);
  UdmConstWordListFree(&CWL);
  return rc;
}



#define UDM_BLOB2_MULTI_INSERT_ALLOC_SIZE  32*1024*1024
#define UDM_BLOB2_MULTI_INSERT_FLUSH_SIZE  32*1024*1024
#define UDM_BLOB2_MILTI_INSERT_WORD_SIZE   1024

static udm_rc_t
UdmSendMultiInsertQuery(UDM_AGENT *A, UDM_DB *db,
                        UDM_INVERTED_INDEX_STATS *Stats, UDM_DSTR *multi)
{
  udm_timer_t ticks= UdmStartTimer();
  udm_rc_t rc= UdmDBSQLQuery(A, db, NULL, UdmDSTRPtr(multi));
  Stats->send_multi+= UdmStartTimer() - ticks;
  return rc;
}


/* TODO34: add compression */
static udm_rc_t
UdmInvertedIndexCacheStoreWord(UDM_AGENT *Indexer, UDM_DB *db,
                               UDM_INVERTED_INDEX_STATS *Stats,
                               const char *wtable,
                               UDM_DSTR *dstr,
                               UDM_DSTR *multi,
                               UDM_INVERTED_INDEX_CACHE_ITEM *Items,
                               size_t nitems)
{
  size_t i;
  udm_timer_t ticks= UdmStartTimer();

  UdmDSTRReset(dstr);
  for (i= 0; i < nitems; i++)
  {
    UDM_INVERTED_INDEX_CACHE_ITEM *Item= &Items[i];
    uint4 wordlen= strlen(Item->ptr) + 1;
    if (!UdmDSTRAppendINT4(dstr, Item->url_id) ||
        !UdmDSTRAppend(dstr, Item->ptr + wordlen,
                             Item->length - wordlen))
    {
      UdmLog(Indexer, UDM_LOG_ERROR, "UdmDSTRAppend failed");
      return UDM_ERROR;
    }
  }
  Stats->pack+= UdmStartTimer() - ticks;

  /*
  fprintf(stderr, "secno=%d ndocs=%-6d ncoords=%-6d datalen=%-6d %s\n",
          Items->secno, (int) nitems, (int) ncoords, (int) dstr->size_data, Items->word);
  */
  /*
    TODO: Multi-insert for MyODBC. Needs proper escaping.
  */
  if (UdmSQLDBDriver(db) == UDM_DBAPI_MYSQL &&
      UdmDSTRLength(dstr) < UDM_BLOB2_MILTI_INSERT_WORD_SIZE)
  {
    ticks= UdmStartTimer();
    if (!UdmDSTRLength(multi))
      UdmDSTRAppendf(multi, "INSERT INTO %s VALUES('", wtable);
    else
      UdmDSTRAppend(multi, UDM_CSTR_WITH_LEN(",('"));
    UdmDSTRAppend(multi, Items->ptr, strlen(Items->ptr));
    UdmDSTRAppendf(multi,"',%d,", Items->secno);
    if (0)
    {
      UdmDSTRAppend(multi, "0x", 2);
      UdmDSTRAppendHex(multi, UdmDSTRPtr(dstr), UdmDSTRLength(dstr));
    }
    else
    {
      size_t esclen;
      UdmDSTRAppend(multi, "'", 1);
      if (UDM_OK != UdmDSTRRealloc(multi, UdmDSTRLength(multi) +
                                          2 * UdmDSTRLength(dstr)))
        return UDM_ERROR;
      esclen= UdmDBSQLEscStr(Indexer, db, multi->Val.str + multi->Val.length,
                             UdmDSTRPtr(dstr), UdmDSTRLength(dstr));
      multi->Val.length+= esclen;
      UdmDSTRAppend(multi, "'", 1);
    }
    UdmDSTRAppend(multi, UDM_CSTR_WITH_LEN(")"));
    Stats->send_multi+= UdmStartTimer() - ticks;
    if (UdmDSTRLength(multi) > UDM_BLOB2_MULTI_INSERT_FLUSH_SIZE)
    {
      if (UDM_OK != UdmSendMultiInsertQuery(Indexer, db, Stats, multi))
        return UDM_ERROR;
      UdmDSTRReset(multi);
    }
  }
  else
  {
    ticks= UdmStartTimer();
    if (UDM_OK != UdmBlobWriteWordUsingBind(Indexer, db, wtable,
                                            Items->ptr, Items->secno,
                                            UdmDSTRPtr(dstr),
                                            UdmDSTRLength(dstr),
                                            NULL, 0))
    {
      UdmLog(Indexer, UDM_LOG_ERROR, "%s", UdmDBSQLError(db));
      return UDM_ERROR;
    }
    Stats->send+= UdmStartTimer() - ticks;
  }

  return UDM_OK;
}


static udm_rc_t
UdmInvertedIndexCacheStoreUsingBind(UDM_AGENT *Indexer, UDM_DB *db,
                                    UDM_INVERTED_INDEX_STATS *Stats,
                                    const char *wtable,
                                    UDM_INVERTED_INDEX_CACHE_PART *cache,
                                    UDM_DSTR *dstr, UDM_DSTR *multi)
{
  UDM_INVERTED_INDEX_CACHE_ITEM *prev;
  udm_rc_t rc;
  size_t i;

  if (cache->nitems == 0)
    return UDM_OK;

  if (UDM_OK != (rc= UdmBlobWriteWordPrepare(Indexer, db, wtable)))
    return rc;

  for (prev= &cache->Item[0], i= 0; i <= cache->nitems; i++)
  {
    UDM_INVERTED_INDEX_CACHE_ITEM *Item= &cache->Item[i];
    if (i == cache->nitems ||
        strcmp(prev->ptr, Item->ptr) ||
        prev->secno != Item->secno)
    {
      if (UDM_OK != (rc= UdmInvertedIndexCacheStoreWord(Indexer, db,
                                                        Stats,
                                                        wtable,
                                                        dstr, multi,
                                                        prev, Item - prev)))
        break;
      prev= Item;
    }
  }
  UdmDBSQLStmtFree(Indexer, db);
  if (UdmDSTRLength(multi) &&
      UDM_OK != UdmSendMultiInsertQuery(Indexer, db, Stats, multi))
    return UDM_ERROR;
  return rc;
}


/**
  Store a sorted part
*/
static udm_rc_t
UdmInvertedIndexCachePartStore(UDM_AGENT *Indexer, UDM_DB *db,
                               UDM_INVERTED_INDEX_STATS *Stats,
                               const char *wtable,
                               UDM_INVERTED_INDEX_CACHE_PART *part)
{
  udm_rc_t rc= UDM_OK;
  UDM_DSTR dstr, multi;
  UdmDSTRInit(&dstr, 1024);
  UdmDSTRInit(&multi, UDM_BLOB2_MULTI_INSERT_ALLOC_SIZE);
  UdmLog(Indexer, UDM_LOG_DEBUG, "Writing word index cache part");
  rc= UdmInvertedIndexCacheStoreUsingBind(Indexer, db, Stats,
                                          wtable, part, &dstr, &multi);
  UdmDSTRFree(&multi);
  UdmDSTRFree(&dstr);
  return rc;
}


/**
  Sort one cache part
*/
static void
UdmInvertedIndexCacheSortPart(UDM_AGENT *A,
                              UDM_INVERTED_INDEX_CACHE *cache,
                              size_t partno)
{
  udm_timer_t ticks= UdmStartTimer();
  UDM_INVERTED_INDEX_CACHE_PART *part= &cache->Item[partno];
  UdmLog(A, UDM_LOG_DEBUG,
         "Sorting word index cache part%02d: %d records",
         (int) partno, (int) part->nitems);
  UdmInvertedIndexCachePartSort(part);
  cache->Stats.sort+= UdmStartTimer() - ticks;
}


typedef struct
{
  size_t partno;
  UDM_INVERTED_INDEX_CACHE *cache;
} UDM_INDEXCACHESORT_SHARE;


typedef struct
{
  UDM_AGENT Agent;
  void *thd;
  UDM_INDEXCACHESORT_SHARE *share;
} UDM_INDEXCACHESORT_PARAM;



static
void *IndexCacheSortThread(void *arg)
{
  UDM_INDEXCACHESORT_PARAM *param= (UDM_INDEXCACHESORT_PARAM *) arg;
  udm_bool_t done= UDM_FALSE;
  for ( ; ; )
  {
    size_t partno;
    UDM_GETLOCK(&param->Agent, UDM_LOCK_DOC_CACHE);
    partno= param->share->partno;
    if (param->share->partno < param->share->cache->nitems)
      param->share->partno++;
    else
      done= UDM_TRUE;
    UDM_RELEASELOCK(&param->Agent, UDM_LOCK_DOC_CACHE);
    if (done)
      break;
    UdmInvertedIndexCacheSortPart(&param->Agent, param->share->cache, partno);
  }
  return NULL;
}


static udm_rc_t
UdmInvertedIndexCacheSortThreaded(UDM_AGENT *A,
                                  UDM_INVERTED_INDEX_CACHE *cache,
                                  size_t nthreads)
{
  size_t i;
  udm_timer_t ticks= UdmStartTimer();
  UDM_INDEXCACHESORT_SHARE share;
  UDM_INDEXCACHESORT_PARAM params[UDM_INDEXER_THREADS_MAX];

  share.cache= cache;
  share.partno= 0;
  if (nthreads > cache->nitems)
    nthreads= cache->nitems;

  UdmLog(A, UDM_LOG_DEBUG, "Sorting cache");
  for (i= 0; i < nthreads; i++)
  {
    UDM_INDEXCACHESORT_PARAM *param= &params[i];
    UdmAgentInit(&param->Agent, A->Conf, i + 1);
    param->thd= NULL;
    param->share= &share;
    A->Conf->THDHandler.ThreadCreate(&param->thd, IndexCacheSortThread, (void*) param);
  }
  for (i= 0; i < nthreads; i++)
  {
    void *thd= params[i].thd;
    A->Conf->THDHandler.ThreadJoin(thd);
  }

  UdmLog(A, UDM_LOG_DEBUG, "Sorting threads done: %.2f", UdmStopTimer(&ticks));
  for (i= 0 ; i < nthreads; i++)
  {
    UDM_INDEXCACHESORT_PARAM *param= &params[i];
    UdmAgentFree(&param->Agent);
  }
  return UDM_OK;
}


/**
  Sort all cache parts
*/
static void
UdmInvertedIndexCacheSort(UDM_AGENT *A, UDM_INVERTED_INDEX_CACHE *cache)
{
  size_t nthreads= UdmVarListFindInt(&A->Conf->Vars, "IndexerThreads", 0);
  if (nthreads && A->Conf->THDHandler.ThreadCreate)
  {
    UdmInvertedIndexCacheSortThreaded(A, cache, nthreads);
  }
  else
  {
    size_t i;
    for (i= 0; i < cache->nitems; i++)
      UdmInvertedIndexCacheSortPart(A, cache, i);
  }
}


static udm_rc_t
UdmInvertedIndexCacheStore(UDM_AGENT *Indexer,
                           UDM_DB *db, const char *wtable,
                           UDM_INVERTED_INDEX_CACHE *cache)
{
  int tr= (UdmSQLDBFlags(db) & UDM_SQL_HAVE_TRANSACT) ? 1 : 0;
  size_t i;
  UdmLog(Indexer, UDM_LOG_EXTRA,
         "Storing word index cache: %lld bytes",
          (unsigned long long) UdmInvertedIndexCacheEstimateUsedMemory(cache));
  UdmInvertedIndexCacheSort(Indexer, cache);
  for (i= 0; i < cache->nitems; i++)
  {
    udm_rc_t rc;
    UDM_INVERTED_INDEX_CACHE_PART *Item= &cache->Item[i];
    if (!Item->nitems)
      continue;
    if (tr && (UDM_OK != (rc= UdmDBSQLBegin(Indexer, db))))
      return rc;
    if (UDM_OK != (rc= UdmInvertedIndexCachePartStore(Indexer, db,
                                                      &cache->Stats,
                                                      wtable, Item)))
      return rc;
    if (tr && (UDM_OK != (rc= UdmDBSQLCommit(Indexer, db))))
      return rc;
  }
  return UDM_OK;
}


static void
UdmInvertedIndexCacheReportStatistics(UDM_AGENT *Indexer,
                                      UDM_INVERTED_INDEX_STATS *Stats)
{
  UdmLog(Indexer, UDM_LOG_INFO, "Indexing statistics:");
  UdmLog(Indexer, UDM_LOG_INFO, "- Loading cached copies:      %.2f (%llu bytes)", (double) Stats->load / 1000, Stats->bytes_loaded);
  if (Stats->unpack_cached_copy)
    UdmLog(Indexer, UDM_LOG_INFO, "- Unpacking cached copies:    %.2f", (double) Stats->unpack_cached_copy / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Parsing documents:          %.2f", (double) Stats->parse / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Breaking sections to words: %.2f", (double) Stats->prepare_words / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Sorting word list:          %.2f", (double) Stats->sort_wordlist / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Groupping words:            %.2f", (double) Stats->conv / 1000);
  if (Stats->pairs)
    UdmLog(Indexer, UDM_LOG_INFO, "- Making pairs:               %.2f", (double) Stats->pairs / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Sorting words: %.2f", (double) Stats->sort / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Packing words: %.2f", (double) Stats->pack / 1000);
  UdmLog(Indexer, UDM_LOG_INFO, "- Sending words: %.2f", (double) (Stats->send + Stats->send_multi) / 1000);
  if (Stats->send_multi)
  {
    UdmLog(Indexer, UDM_LOG_EXTRA, "  +- Sending words: %.2f (MultiInsert)", (double) Stats->send_multi / 1000);
    UdmLog(Indexer, UDM_LOG_EXTRA, "  +- Sending words: %.2f (SQLExec)", (double) Stats->send / 1000);
  }
}


static udm_rc_t
UdmSQLFetchRowCachedCopy(UDM_AGENT *A, UDM_DB *db,
                         UDM_INVERTED_INDEX_STATS *Stats,
                         UDM_SQLRES *SQLRes, UDM_STR *row)
{
  udm_timer_t ticks= UdmStartTimer();
  udm_rc_t rc= UdmDBSQLFetchRow(A, db, SQLRes, row);
  Stats->load+= UdmStartTimer() - ticks;
  return rc;
}


static udm_rc_t
UdmParseRowCachedCopy(UDM_AGENT *Indexer, UDM_DB *db,
                      UDM_INVERTED_INDEX_CACHE *cache,
                      UDM_STR *row, UDM_URLDATALIST *URLList,
                      size_t max_doc_size)
{
  udm_rc_t rc;
  UDM_DOCUMENT Doc;
  urlid_t url_id= atoi(row[0].str);
  UDM_URLDATA *urldata;
  if (!(urldata= UdmURLDataListSearch(URLList, url_id)))
  {
    UdmLog(Indexer, UDM_LOG_DEBUG, "UdmInvertedIndexCreate: url_id=%d not found", url_id);
    return UDM_OK;
  }
  urldata->score++;
  UdmDocInit(&Doc);
  Doc.lcs= Indexer->Conf->lcs;
  if (UDM_OK != (rc= UdmDocSetFromCachedHTTPResponse(&Doc,
                                                     row[1].str, row[1].length,
                                                     max_doc_size,
                                                     &cache->Stats.unpack_cached_copy)))
  {
    UdmLog(Indexer, UDM_LOG_DEBUG,
           "UdmDocSetFromCachedHTTPResponse() failed: url_id=%d len=%d",
           url_id, (int) row[1].length);
    rc= UDM_OK;
    goto end;
  }
  cache->Stats.bytes_loaded+= UdmHTTPBufSize(&Doc.Buf);
  if (UDM_OK != (rc= UdmInvertedIndexCacheDocParse(Indexer, urldata->url_id,
                                                   &Doc, cache)))
    UdmLog(Indexer, UDM_LOG_DEBUG, "UdmInvertedIndexCacheDocParse() failed");

end:
  UdmDocFree(&Doc);
  return rc;
}


/***** Cache for Cached copies *********************************************/
typedef struct
{
  UDM_STR content;
  urlid_t url_id;
} UDM_DOCCACHEITEM;


static udm_rc_t
UdmDocCacheItemInit(UDM_DOCCACHEITEM *Item,
                    const UDM_CONST_STR *content,
                    urlid_t url_id)
{
  UDM_ASSERT(content->length > 0);
  if (!(Item->content.str= UdmMalloc(content->length)))
    return UDM_ERROR;
  Item->content.length= content->length;
  memcpy(Item->content.str, content->str, content->length);
  Item->url_id= url_id;
  return UDM_OK;
}


static void
UdmDocCacheItemFree(UDM_DOCCACHEITEM *Item)
{
  UdmFree(Item->content.str);
}


typedef struct
{
  size_t nitems;
  size_t mitems;
  UDM_DOCCACHEITEM *Item;
} UDM_DOCCACHELIST;


static void
UdmDocCacheListInit(UDM_DOCCACHELIST *List)
{
  bzero((void*) List, sizeof(*List));
}


static udm_rc_t
UdmDocCacheListRealloc(UDM_DOCCACHELIST *List)
{
  if (List->nitems >= List->mitems)
  {
    size_t mitems2= List->nitems + 256;
    if (!(List->Item= UdmRealloc(List->Item, mitems2 * sizeof(List->Item[0]))))
      return UDM_ERROR;
    List->mitems= mitems2;
  }
  return UDM_OK;
}


static udm_rc_t
UdmDocCacheListAdd(UDM_DOCCACHELIST *List, UDM_DOCCACHEITEM *Item)
{
  if (UDM_OK != UdmDocCacheListRealloc(List))
    return UDM_ERROR;
  List->Item[List->nitems++]= *Item;
  return UDM_OK;
}


static void
UdmDocCacheListFree(UDM_DOCCACHELIST *List)
{
  size_t i;
  for (i= 0; i < List->nitems; i++)
    UdmDocCacheItemFree(&List->Item[i]);
  UdmFree(List->Item);
}


typedef struct
{
  UDM_DOCCACHELIST DocList;
  udm_uint8 nbytes;
  size_t cur_item;
} UDM_DOCCACHE;


static void UdmDocCacheInit(UDM_DOCCACHE *Cache)
{
  UdmDocCacheListInit(&Cache->DocList);
  Cache->nbytes= 0;
  Cache->cur_item= 0;
}


static void UdmDocCacheFree(UDM_DOCCACHE *Cache)
{
  UdmDocCacheListFree(&Cache->DocList);
}


static void UdmDocCacheReset(UDM_DOCCACHE *Cache)
{
  UdmDocCacheListFree(&Cache->DocList);
  UdmDocCacheListInit(&Cache->DocList);
  Cache->nbytes= 0;
  Cache->cur_item= 0;
}

/**********************************************************************/

static udm_rc_t
UdmRowCachedCopyToDocCache(UDM_AGENT *Indexer, UDM_DB *db,
                           UDM_INVERTED_INDEX_STATS *Stats,
                           UDM_STR *row,
                           UDM_DOCCACHE *DocCache,
                           UDM_URLDATALIST *URLList)
{
  urlid_t url_id= atoi(row[0].str);
  UDM_URLDATA *urldata;
  UDM_CONST_STR content;
  UDM_DOCCACHEITEM DocCacheItem;
  if (!(urldata= UdmURLDataListSearch(URLList, url_id)))
  {
    UdmLog(Indexer, UDM_LOG_DEBUG, "UdmInvertedIndexCreate: url_id=%d not found", url_id);
    return UDM_OK;
  }
  urldata->score++;
  UdmConstStrSet(&content, row[1].str, row[1].length);
  if (UDM_OK != UdmDocCacheItemInit(&DocCacheItem, &content, url_id))
    return UDM_ERROR;
  if (UdmDocCacheListAdd(&DocCache->DocList, &DocCacheItem))
    return UDM_ERROR;
  DocCache->nbytes+= content.length;
  return UDM_OK;
}


static udm_rc_t
UdmCreateContentQueryUsingURLIdLoop(UDM_DSTR *query, const char *fmt,
                                    UDM_URLDATALIST *URLList, size_t offs,
                                    size_t nrows,
                                    const UDM_CONST_STR ihint)
{
  UdmDSTRReset(query);
  for ( ; *fmt; )
  {
    if (!strncasecmp(fmt, UDM_CSTR_WITH_LEN("${where}")))
    {
      size_t j;
      fmt+= 8;
      UdmDSTRAppend(query, UDM_CSTR_WITH_LEN(" WHERE url_id IN("));
      for (j=0; j < nrows; j++)
      {
        if (j > 0)
          UdmDSTRAppend(query, ",", 1);
        UdmDSTRAppendf(query, "%d", URLList->Item[offs + j].url_id);
      }
      UdmDSTRAppend(query, UDM_CSTR_WITH_LEN(")"));
    }
    else if (!strncasecmp(fmt, UDM_CSTR_WITH_LEN("${ihint}")))
    {
      fmt+= 8;
      UdmDSTRAppend(query, ihint.str, ihint.length);
    }
    else
      UdmDSTRAppend(query, fmt++, 1);
  }
  return UDM_OK;
}


#define DEFAULT_CONTENT_FMT "SELECT url_id,content FROM cachedcopy${ihint}${where}"


static size_t
in_limit(UDM_DB *db)
{
  if (UdmSQLDBType(db)== UDM_DB_ORACLE8)
    return 1000; /* TODO34: move to UDM_SQLDB_DRIVER */
  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
    return 10*1024;
  if (UdmSQLDBType(db) == UDM_DB_IBASE)
    return 1500;
  return 2000;
}


static udm_rc_t
UdmDocCacheIndexOneDocument(UDM_AGENT *Indexer,
                            UDM_DOCCACHE *Docs,
                            UDM_INVERTED_INDEX_CACHE *cache,
                            size_t max_doc_size)
{
  udm_rc_t rc;
  UDM_DOCCACHEITEM *Doc;
  UDM_DOCUMENT Doc2;

  UDM_GETLOCK(Indexer, UDM_LOCK_DOC_CACHE);
  if (Docs->cur_item < Docs->DocList.nitems)
  {
    Doc= &Docs->DocList.Item[Docs->cur_item];
    Docs->cur_item++;
  }
  else
  {
    Doc= NULL;
  }
  UDM_RELEASELOCK(Indexer, UDM_LOCK_DOC_CACHE);
  if (!Doc)
    return UDM_NOTARGET;

  UdmDocInit(&Doc2);
  Doc2.lcs= Indexer->Conf->lcs;
  if (UDM_OK != (rc= UdmDocSetFromCachedHTTPResponse(&Doc2,
                                                     Doc->content.str,
                                                     Doc->content.length,
                                                     max_doc_size,
                                                     &cache->Stats.unpack_cached_copy)))
  {
    UdmLog(Indexer, UDM_LOG_DEBUG,
           "UdmDocSetFromCachedHTTPResponse() failed: url_id=%d len=%d",
           Doc->url_id, (int) Doc->content.length);
    rc= UDM_OK;
    goto end;
  }
  cache->Stats.bytes_loaded+= UdmHTTPBufSize(&Doc2.Buf);

  if (UDM_OK != (rc= UdmInvertedIndexCacheDocParse(Indexer, Doc->url_id,
                                                   &Doc2, cache)))
    UdmLog(Indexer, UDM_LOG_DEBUG, "UdmInvertedIndexCacheDocParse() failed");

end:
  UdmDocFree(&Doc2);
  return rc;
}


static udm_rc_t
UdmInvertedIndexCacheParse(UDM_AGENT *Indexer,
                           UDM_DOCCACHE *Docs,
                           UDM_INVERTED_INDEX_CACHE *cache,
                           size_t max_doc_size)
{
  int rc;
  while (UDM_OK == (rc= UdmDocCacheIndexOneDocument(Indexer, Docs, cache,
                                                    max_doc_size)))
  { }
  return rc == UDM_NOTARGET ? UDM_OK : rc;
}


typedef struct indexer_param_st
{
  UDM_AGENT Agent;
  UDM_DOCCACHE *Docs;
  UDM_INVERTED_INDEX_CACHE *cache;
  size_t max_doc_size;
  udm_rc_t rc;
  void *thd;
} UDM_INDEXER_PARAM;


static
void *IndexerThread(void *arg)
{
  UDM_INDEXER_PARAM *param= (UDM_INDEXER_PARAM *) arg;
  param->rc= UdmInvertedIndexCacheParse(&param->Agent, param->Docs,
                                        param->cache,
                                        param->max_doc_size);
  return NULL;
}


static udm_rc_t
UdmInvertedIndexCacheParseThreaded(UDM_AGENT *A,
                                   UDM_DOCCACHE *Docs,
                                   UDM_INVERTED_INDEX_CACHE *cache,
                                   size_t nthreads)
{
  size_t i;
  udm_timer_t ticks= UdmStartTimer();
  size_t max_doc_size= UdmVarListFindInt(&A->Conf->Vars, "MaxDocSize", UDM_MAXDOCSIZE);
  UDM_INDEXER_PARAM params[UDM_INDEXER_THREADS_MAX];

  UdmLog(A, UDM_LOG_DEBUG, "   Indexing %d docs", (int) Docs->DocList.nitems);
  if (!Docs->DocList.nitems)
    return UDM_OK;
  if (nthreads > Docs->DocList.nitems)
    nthreads= Docs->DocList.nitems;
  for (i= 0; i < nthreads; i++)
  {
    UDM_INDEXER_PARAM *param= &params[i];
    UdmAgentInit(&param->Agent, A->Conf, i + 1);
    param->Docs= Docs;
    param->cache= cache;
    param->thd= NULL;
    param->max_doc_size= max_doc_size;
    A->Conf->THDHandler.ThreadCreate(&param->thd, IndexerThread, (void*) param);
  }
  for (i= 0; i < nthreads; i++)
  {
    void *thd= params[i].thd;
    A->Conf->THDHandler.ThreadJoin(thd);
  }

  UdmLog(A, UDM_LOG_DEBUG, "   Threads finished: %.2f", UdmStopTimer(&ticks));
  for (i= 0 ; i < nthreads; i++)
  {
    UDM_INDEXER_PARAM *param= &params[i];
    UdmAgentFree(&param->Agent);
  }
  return UDM_OK;
}


static udm_rc_t
UdmIndertedIndexExecContentQuery(UDM_AGENT *Indexer, UDM_DB *db,
                                 UDM_INVERTED_INDEX_CACHE *cache,
                                 UDM_QUERY *Query,
                                 const char *qbuf)
{
  udm_rc_t rc= UDM_OK;
  udm_timer_t ticks1= UdmStartTimer();
  size_t max_doc_size= UdmVarListFindInt(&Indexer->Conf->Vars, "MaxDocSize", UDM_MAXDOCSIZE);
  size_t nthreads= UdmVarListFindInt(&Indexer->Conf->Vars, "IndexerThreads", 0);
  UDM_SQLRES SQLRes;
  UDM_STR row[2];
  UDM_DOCCACHE DocCache;

  if (UDM_OK != (rc= UdmDBSQLExecDirect(Indexer, db, &SQLRes, qbuf)))
    goto end;
  cache->Stats.load+= UdmStartTimer() - ticks1;

  UdmDocCacheInit(&DocCache);
  while (UDM_OK == UdmSQLFetchRowCachedCopy(Indexer, db, &cache->Stats,
                                            &SQLRes, row))
  {
    if (!nthreads || !Indexer->Conf->THDHandler.ThreadCreate)
    {
      if (UDM_OK != (rc= UdmParseRowCachedCopy(Indexer, db, cache,
                                               row, &Query->URLData,
                                               max_doc_size)))
        break;
    }
    else
    {
      if (UDM_OK != (rc= UdmRowCachedCopyToDocCache(Indexer, db, &cache->Stats,
                                                    row, &DocCache,
                                                    &Query->URLData)))
        break;
      /*fprintf(stderr, "DOCCACHE=%.3f\n", (double) DocCache.nbytes / 1024 / 1024);*/
      if (DocCache.nbytes >= 32 * 1024 * 1024)
      {
        UdmInvertedIndexCacheParseThreaded(Indexer, &DocCache,
                                           cache, nthreads);
        UdmDocCacheReset(&DocCache);
      }
    }
  }
  UdmSQLFree(&SQLRes);
  if (DocCache.DocList.nitems)
    UdmInvertedIndexCacheParseThreaded(Indexer, &DocCache, cache, nthreads);
  UdmDocCacheFree(&DocCache);

end:
  return rc;
}


static udm_rc_t
UdmInvertedIndexCreateUsingURLIdLoop(UDM_AGENT *Indexer,
                                     UDM_DB *db,
                                     UDM_INVERTED_INDEX_CACHE *cache,
                                     const char *wtable,
                                     UDM_QUERY *Query)
{
  udm_rc_t rc= UDM_OK;
  UDM_DSTR query;
  UDM_CONST_STR ihint;
  UDM_URLDATALIST *URLList= &Query->URLData;
  size_t i;
  size_t ndocs_at_time= in_limit(db);
  udm_uint8 cache_size_limit= (udm_uint8) UdmVarListFindUnsigned(&Indexer->Conf->Vars,
                                                                 "IndexCacheSize",
                                                                 128*1024*1024);
  UdmConstStrInit(&ihint);
  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
  {
    UDM_SQLRES SQLRes;
    UdmDBSQLQuery(Indexer, db, &SQLRes, "SHOW CREATE TABLE cachedcopy");
    if (UdmSQLNumCols(&SQLRes) == 2 && UdmSQLNumRows(&SQLRes) == 1)
    {
      if (strstr(UdmSQLValue(&SQLRes, 0, 1), "PARTITION BY HASH (url_id"))
        UdmConstStrSet(&ihint, UDM_CSTR_WITH_LEN(" IGNORE INDEX (url_id)"));
    }
    UdmSQLFree(&SQLRes);
  }
  UdmDSTRInit(&query, ndocs_at_time * 8);
  for (i= 0; i < URLList->nitems; )
  {
    udm_uint8 cache_size;
    size_t nrows= UDM_MIN(ndocs_at_time, URLList->nitems - i);
    UdmLog(Indexer, UDM_LOG_DEBUG, "-- IDs %d-%d",
           URLList->Item[i].url_id, URLList->Item[i+nrows-1].url_id);
    if (UDM_OK != (rc= UdmCreateContentQueryUsingURLIdLoop(&query,
                                                           DEFAULT_CONTENT_FMT,
                                                           URLList, i, nrows,
                                                           ihint)))
      goto end;
    if (UDM_OK != (rc= UdmIndertedIndexExecContentQuery(Indexer, db,
                                                        cache, Query,
                                                        UdmDSTRPtr(&query))))
      goto end;
    i+= nrows;

    if ((cache_size= UdmInvertedIndexCacheEstimateUsedMemory(cache)) >
        cache_size_limit)
    {
      rc= UdmInvertedIndexCacheStore(Indexer, db, wtable, cache);
      UdmInvertedIndexCacheReset(cache);
    }
  }

  rc= UdmInvertedIndexCacheStore(Indexer, db, wtable, cache);
end:
  UdmDSTRFree(&query);
  return rc;
}


static udm_rc_t
UdmInvertedIndexCreateUsingGenericLoop(UDM_AGENT *Indexer,
                                       UDM_DB *db,
                                       UDM_INVERTED_INDEX_CACHE *cache,
                                       const char *wtable,
                                       UDM_QUERY *Query,
                                       const char *field,
                                       size_t from, size_t to)
{
  udm_rc_t rc= UDM_OK;
  size_t i;
  udm_uint8 cache_size_limit= (udm_uint8) UdmVarListFindInt(&Indexer->Conf->Vars,
                                                            "IndexCacheSize",
                                                            128*1024*1024);
  for (i= from; i < to; i++)
  {
    char qbuf[256];
    udm_uint8 cache_size;
    if (from + 1 != to)
      UdmLog(Indexer, UDM_LOG_DEBUG, "-- Part %d", (int) i);
    udm_snprintf(qbuf, sizeof(qbuf),
                 "SELECT url_id,content FROM cachedcopy WHERE %s=%d",
                 field, (int) i);
    if (UDM_OK != (rc= UdmIndertedIndexExecContentQuery(Indexer, db,
                                                        cache, Query,
                                                        qbuf)))
      goto end;
    if ((cache_size= UdmInvertedIndexCacheEstimateUsedMemory(cache)) >
        cache_size_limit)
    {
      rc= UdmInvertedIndexCacheStore(Indexer, db, wtable, cache);
      UdmInvertedIndexCacheReset(cache);
    }
  }

  rc= UdmInvertedIndexCacheStore(Indexer, db, wtable, cache);
end:
  return rc;
}


/*
  TODO34: Character set detection (crawl or index time?)
  TODO34: Segmenting (crawl or index time)?
  TODO34: CRC32 (crawl or index time?)
  TODO34: what to do with urlinfo sections?
  TODO34: Raw sections
  TODO34: Pairs
  TODO34: remove HAVE_ZLIB
  TODO34: In case of 'Mime application/msword "text/plain; charset=utf-8"',
          the charset= part is stored twice in urlinfob.
*/
static udm_rc_t
UdmInvertedIndexCreate(UDM_AGENT *Indexer, UDM_DB *db, const char *wtable,
                       UDM_QUERY *Query)
{
  udm_timer_t ticks= UdmStartTimer();
  udm_rc_t rc= UDM_OK;
  UDM_INVERTED_INDEX_CACHE cache;

  UdmInvertedIndexCacheInit(&cache, Indexer->Conf);
  if (UDM_OK != UdmInvertedIndexCacheAllocParts(Indexer, &cache,
                                                INVERTED_INDEX_CACHE_PARTS))
    return UDM_ERROR;
  UdmLog(Indexer, UDM_LOG_INFO, "Indexing document contents");

  if (1)
    rc= UdmInvertedIndexCreateUsingURLIdLoop(Indexer, db, &cache, wtable, Query);
  else
    rc= UdmInvertedIndexCreateUsingGenericLoop(Indexer, db, &cache, wtable,
                                               Query, "0", 0, 1);

  UdmLog(Indexer, UDM_LOG_INFO, "Freeing cache");
  {
    udm_timer_t ticks1= UdmStartTimer();
    UdmInvertedIndexCacheFree(&cache);
    UdmLog(Indexer, UDM_LOG_INFO, "Freeing cache done: %.2f", UdmStopTimer(&ticks1));
  }
  UdmLog(Indexer, UDM_LOG_INFO, "Indexing document contents done: %.2f", UdmStopTimer(&ticks));
  UdmInvertedIndexCacheReportStatistics(Indexer, &cache.Stats);
  return rc;
}


typedef struct
{
  size_t outgoing_link_count;
  size_t incoming_link_count;
  double popularity0;
  double popularity1;
  double server_weight;
  urlid_t url_id;
  uint4 score;
} UDM_URLPOPINFO;


typedef struct
{
  size_t mitems;
  size_t nitems;
  UDM_URLPOPINFO *Item;
} UDM_URLPOPINFOLIST;


static void
UdmURLPopInfoListInit(UDM_URLPOPINFOLIST *List)
{
  bzero((void*) List, sizeof(*List));
}


static udm_rc_t
UdmURLPopInfoListAlloc(UDM_URLPOPINFOLIST *List, size_t mitems)
{
  if (!(List->Item= (UDM_URLPOPINFO *) UdmMalloc(mitems * sizeof(UDM_URLPOPINFO))))
    return UDM_ERROR;
  List->mitems= mitems;
  return UDM_OK;
}


static void
UdmURLPopInfoListFree(UDM_URLPOPINFOLIST *List)
{
  UdmFree(List->Item);
}


static int
cmp_data_urlpopinfo(UDM_URLPOPINFO *d1, UDM_URLPOPINFO *d2)
{
  if (d1->url_id > d2->url_id) return 1;
  if (d1->url_id < d2->url_id) return -1;
  return 0;
}


static UDM_URLPOPINFO *
UdmURLPopInfoListSearch(UDM_URLPOPINFOLIST *List, urlid_t id)
{
  UDM_URLPOPINFO d;
  void *found;
  if (!List->nitems)
    return 0;
  d.url_id= id;
  found= UdmBSearch(&d, List->Item, List->nitems, sizeof(UDM_URLPOPINFO),
                    (udm_qsort_cmp) cmp_data_urlpopinfo);
  return (UDM_URLPOPINFO*) found;
}


/*
  Init URLPopInfoList from a sorted URLDataList.
*/
static udm_rc_t
UdmURLPopInfoListInitFromURLDataList(UDM_AGENT *A,
                                     UDM_URLPOPINFOLIST *URLPopInfoList,
                                     UDM_URLDATALIST *URLDataList)
{
  size_t i;
  double r= 1e0 / (double) URLDataList->nitems;
  if (UDM_OK != UdmURLPopInfoListAlloc(URLPopInfoList, URLDataList->nitems))
    return UDM_ERROR;
  for (i= 0 ; i < URLDataList->nitems; i++)
  {
    UDM_URLPOPINFO *dst= &URLPopInfoList->Item[i];
    UDM_URLDATA *src= &URLDataList->Item[i];
    UDM_SERVER *Server= UdmServerFind(A, &A->Conf->Servers, src->url, NULL);
    dst->url_id= src->url_id;
    dst->score= src->score;
    dst->popularity0= r;
    dst->server_weight= (Server ? Server->weight : 1);
    dst->popularity1= 0;
    dst->outgoing_link_count= 0;
    dst->incoming_link_count= 0;
    /*fprintf(stderr, "[%.5f]%s\n", dst->popularity0, Server ?  Server->Match.pattern : NULL);*/
  }
  URLPopInfoList->nitems= URLDataList->nitems;
  return UDM_OK;
}


typedef struct
{
  UDM_URLPOPINFO *from;
  UDM_URLPOPINFO *to;
} UDM_LINKINFO;


typedef struct
{
  size_t nitems;
  size_t mitems;
  UDM_LINKINFO *Item;
} UDM_LINKINFOLIST;

static void
UdmLinkInfoListInit(UDM_LINKINFOLIST *List)
{
  bzero((void*) List, sizeof(*List));
}

static void
UdmLinkInfoListFree(UDM_LINKINFOLIST *List)
{
  UdmFree(List->Item);
}

static udm_rc_t
UdmLinkInfoListAlloc(UDM_LINKINFOLIST *List, size_t mitems)
{
  if (!(List->Item= (UDM_LINKINFO*) UdmMalloc(mitems * sizeof(UDM_LINKINFO))))
    return UDM_ERROR;
  List->mitems= mitems;
  return UDM_OK;
}

static void
UdmLinkInfoListAdd(UDM_LINKINFOLIST *List, const UDM_LINKINFO *Item)
{
  UDM_ASSERT(List->nitems < List->mitems);
  List->Item[List->nitems++]= *Item;
}


typedef struct
{
  size_t mitems;
  size_t nitems;
  UDM_LINKINFOLIST *Item;
} UDM_LINKINFOLISTLIST;


static void
UdmLinkInfoListListInit(UDM_LINKINFOLISTLIST *List)
{
  bzero((void*) List, sizeof(*List));
}

static void
UdmLinkInfoListListFree(UDM_LINKINFOLISTLIST *List)
{
  size_t i;
  for (i= 0; i < List->nitems; i++)
    UdmLinkInfoListFree(&List->Item[i]);
  UdmFree(List->Item);
}

static udm_rc_t
UdmLinkInfoListListRealloc(UDM_LINKINFOLISTLIST *List)
{
  if (List->nitems >= List->mitems)
  {
    size_t mitems= List->mitems + 256;
    size_t nbytes= mitems * sizeof(UDM_LINKINFOLIST);
    if (!(List->Item= (UDM_LINKINFOLIST*) UdmRealloc(List->Item, nbytes)))
      return UDM_ERROR;
    List->mitems= mitems;
  }
  return UDM_OK;
}

static udm_rc_t
UdmLinkInfoListListAdd(UDM_LINKINFOLISTLIST *List, const UDM_LINKINFOLIST *Item)
{
  if (UDM_OK != UdmLinkInfoListListRealloc(List))
    return UDM_ERROR;
  List->Item[List->nitems++]= *Item;
  return UDM_OK;
}


typedef struct
{
  UDM_CONST_STR text;
  UDM_LINKINFO link;
} UDM_LINKTEXT;

typedef struct
{
  size_t nitems;
  size_t mitems;
  UDM_LINKTEXT *Item;
} UDM_LINKTEXTLIST;

static int
linktextcmp(const UDM_LINKTEXT *a, const UDM_LINKTEXT *b)
{
  if (a->link.to->url_id != b->link.to->url_id)
    return a->link.to->url_id < b->link.to->url_id ? -1 : 1;
  return 0;
}

static void
UdmLinkTextListInit(UDM_LINKTEXTLIST *List)
{
  bzero((void*) List, sizeof(*List));
}

static void
UdmLinkTextListFree(UDM_LINKTEXTLIST *List)
{
  UdmFree(List->Item);
}

static udm_rc_t
UdmLinkTextListAlloc(UDM_LINKTEXTLIST *List, size_t nitems)
{
  if (!(List->Item= (UDM_LINKTEXT *) UdmMalloc(nitems * sizeof(UDM_LINKTEXT))))
    return UDM_ERROR;
  List->mitems= 0;
  return UDM_OK;
}

typedef struct
{
  UDM_CONST_STR url;
  udmcrc32_t hash;
  urlid_t id;
} UDM_URLIDHASH;


static void
UdmURLIdHashItemInitFromURLData(UDM_URLIDHASH *dst, UDM_URLDATA *src)
{
  UdmConstStrSetStr(&dst->url, src->url);
  dst->hash= UdmStrCRC32(src->url);
  dst->id= src->url_id;
}


static void
UdmURLIdHashItemInit(UDM_URLIDHASH *dst, const char *url)
{
  UdmConstStrSetStr(&dst->url, url);
  dst->hash= UdmStrCRC32(url);
  dst->id= 0;
}


static udm_rc_t
UdmURLIdHashStore(UDM_HASH *hash, void *ofs, void *item)
{
  memcpy(ofs, item, sizeof(UDM_URLIDHASH));
  return UDM_OK;
}


static udm_rc_t
UdmURLIdHashJoin(UDM_HASH *hash, void *ofs, void *b)
{
  return UDM_OK;
}


static int
UdmConstStrEQ(const UDM_CONST_STR *s1, const UDM_CONST_STR *s2)
{
  if (s1->length != s2->length)
    return 1;
  return memcmp(s1->str, s2->str, s1->length);
}


/*
  Returns 0 (on equal)
  Returns 1 (on non equal).
  Note: not siutable for sorting!
*/
static int
UdmURLIdHashCmp(UDM_HASH *hash, void *w1, void *w2)
{
  if (((UDM_URLIDHASH*)w1)->hash != ((UDM_URLIDHASH*)w2)->hash)
    return 1;
  return UdmConstStrEQ(&(((UDM_URLIDHASH*)w1)->url),
                       &(((UDM_URLIDHASH*)w2)->url));
}


static udmcrc32_t
UdmURLIdHashCalculateHash(UDM_HASH *hash, const void *item)
{
  return ((const UDM_URLIDHASH*)item)->hash;
}


static UDM_HASH_HANDLER UdmURLIdHashHandler=
{
  UdmURLIdHashStore,         /* store  */
  UdmURLIdHashJoin,          /* join   */
  UdmURLIdHashCmp,           /* cmp    */
  UdmURLIdHashCalculateHash, /* keykey */
  UdmURLIdHashCalculateHash  /* reckey */
};


static udm_rc_t
UdmURLIdHashInitFromURLDataList(UDM_AGENT *Indexer,
                                UDM_HASH *urlidhash,
                                const UDM_URLDATALIST *URLList)
{
  size_t i;
  if (UDM_OK != UdmHashInit(urlidhash, &UdmURLIdHashHandler,
                            NULL, URLList->nitems + URLList->nitems/10 + 100,
                            sizeof(UDM_URLIDHASH)))
  {
    UdmLog(Indexer, UDM_LOG_ERROR, "UdmHashInit failed");
    return UDM_ERROR;
  }
  for (i= 0; i < URLList->nitems; i++)
  {
    UDM_URLIDHASH item;
    UdmURLIdHashItemInitFromURLData(&item, &URLList->Item[i]);
    if (NULL == UdmHashPut(urlidhash, &item))
    {
      UdmLog(Indexer, UDM_LOG_ERROR, "UdmHashPut failed");
      return UDM_ERROR;
    }
  }
  return UDM_OK;
}


static udm_rc_t
UdmInvertedIndexAddURLText(UDM_AGENT *Indexer,
                             UDM_DB *db,
                             const char *wtable,
                             UDM_URLDATALIST *URLList)
{
  udm_rc_t rc= UDM_OK;
  size_t i;
  UDM_INVERTED_INDEX_CACHE cache;

  if (!UdmVarListFindByPrefix(&Indexer->Conf->Sections, "url.", 4))
    return UDM_OK;

  UdmLog(Indexer, UDM_LOG_INFO, "Indexing URL text");
  UdmInvertedIndexCacheInit(&cache, Indexer->Conf);
  if (UDM_OK != UdmInvertedIndexCacheAllocParts(Indexer, &cache,
                                                INVERTED_INDEX_CACHE_PARTS))
    return UDM_ERROR;

  for (i=0; i < URLList->nitems && rc == UDM_OK; i++)
  {
    UDM_URLDATA *Item= &URLList->Item[i];
    UDM_CONSTWORDLIST CWL;
    UDM_DOCUMENT Doc;
    if (!Item->score)
      continue; /* Does not have urlinfob record, e.g. HrefOnly */
    UdmConstWordListInit(&CWL);
    UdmDocInit(&Doc);
    UdmURLParse(&Doc.CurURL, Item->url);
    UdmVarListAddLst(&Doc.Sections, &Indexer->Conf->Sections, NULL, "*");
    UdmParseURLText(Indexer, &Doc);
    /* TODO34: RemoteCharset, RemoteFileNameCharset */
    UdmTextListToConstWordList(&Doc.TextList,
                               Indexer->Conf->unidata, Indexer->Conf->lcs,
                               UDM_RECODE_HTML, &CWL);
    rc= UdmWordListSortAndConvert(Indexer, &CWL, &cache,
                                  Item->url_id, Indexer->Conf->lcs);
    UdmConstWordListFree(&CWL);
    UdmDocFree(&Doc);
  }
  if (rc == UDM_OK)
    rc= UdmInvertedIndexCacheStore(Indexer, db, wtable, &cache);

  UdmInvertedIndexCacheFree(&cache);
  return rc;
}


static udm_rc_t
UdmLinkTextListToInvertedIndexCache(UDM_AGENT *Indexer,
                                    UDM_LINKTEXTLIST *LinkTextList,
                                    UDM_INVERTED_INDEX_CACHE *cache,
                                    udm_secno_t ilinktext)
{
  size_t i;
  udm_rc_t rc= UDM_OK;
  UDM_WORD_SCANNER scanner;
  UDM_CONSTWORDLIST CWL;

  UdmWordScannerInit(&scanner, Indexer->Conf->unidata, Indexer->Conf->lcs);
  UdmConstWordListInit(&CWL);

  /* Sort Linktext items by url_id */
  UdmSort(LinkTextList->Item, LinkTextList->nitems, sizeof(UDM_LINKTEXT), (udm_qsort_cmp) linktextcmp);
  for (i= 0; i < LinkTextList->nitems; i++)
  {
    UDM_LINKTEXT *Item= &LinkTextList->Item[i];
    UDM_URLPOPINFO *popinfo= Item->link.to;
    if (!popinfo->score) /* URL does not have urlinfob record. e.g. HrefOnly */
      continue;
    UdmConstWordListAddString(&scanner,
                              UDM_RECODE_HTML, &CWL, /* TODO34: StripAccents */
                              ilinktext,
                              Item->text.str, Item->text.length);
    CWL.wordpos[ilinktext]+= 8; /* TODO34: check overflow */
    if (i + 1 == LinkTextList->nitems ||
        LinkTextList->Item[i+1].link.to->url_id != Item->link.to->url_id)
    {
      if (UDM_OK != (rc= UdmWordListSortAndConvert(Indexer, &CWL, cache,
                                                   Item->link.to->url_id,
                                                   Indexer->Conf->lcs)))
        goto ex;
      CWL.nitems= 0;
      CWL.wordpos[ilinktext]= 0;
    }
  }
ex:
  UdmConstWordListFree(&CWL);
  return rc;
}


/* Order of the columns */
#define LINKTEXT_ID   0
#define LINKTEXT_URL  1
#define LINKTEXT_TEXT 2

static udm_rc_t
UdmInvertedIndexProcessLinksResult(UDM_AGENT *Indexer,
                                   UDM_HASH *urlidhash,
                                   UDM_URLPOPINFOLIST *URLPopInfoList,
                                   UDM_INVERTED_INDEX_CACHE *cache,
                                   UDM_LINKINFOLISTLIST *LinkInfoListList,
                                   UDM_SQLRES *SQLRes,
                                   udm_secno_t ilinktext)
{
  udm_rc_t rc;
  size_t i, nrows;
  UDM_LINKTEXTLIST LinkTextList;
  UDM_LINKINFOLIST LinkInfoList;

  if (!(nrows= UdmSQLNumRows(SQLRes)))
    return UDM_OK;

  UdmLinkInfoListInit(&LinkInfoList);
  UdmLinkTextListInit(&LinkTextList);

  if (UDM_OK != (rc= UdmLinkTextListAlloc(&LinkTextList, nrows)))
    return rc;

  if (UDM_OK != (rc= UdmLinkInfoListAlloc(&LinkInfoList, nrows)))
    goto ex;

  for (i=0; i < nrows; i++)
  {
    UDM_LINKTEXT *Item= &LinkTextList.Item[LinkTextList.nitems];
    const char *url= UdmSQLValue(SQLRes, i, LINKTEXT_URL);
    urlid_t fromid= atoi(UdmSQLValue(SQLRes, i, LINKTEXT_ID));
    UDM_URLIDHASH *u, search_item;
    UdmURLIdHashItemInit(&search_item, url);
    if (!(u= (UDM_URLIDHASH*) UdmHashFind(urlidhash, &search_item)) || !u->id)
      continue;
    if (!(Item->link.to= UdmURLPopInfoListSearch(URLPopInfoList, u->id)))
    {
      UdmLog(Indexer, UDM_LOG_WARN,
             "UdmInvertedIndexAddLinkText: to_id=%d not found", u->id);
      continue;
    }
    if (!(Item->link.from= UdmURLPopInfoListSearch(URLPopInfoList, fromid)))
    {
      UdmLog(Indexer, UDM_LOG_WARN,
             "UdmInvertedIndexAddLinkText: from_id=%s not found",
             UdmSQLValue(SQLRes, i, LINKTEXT_ID));
      continue;
    }
    if (cache)
    {
      Item->text.str= UdmSQLValue(SQLRes, i, LINKTEXT_TEXT);
      Item->text.length= UdmSQLLen(SQLRes, i, LINKTEXT_TEXT);
      LinkTextList.nitems++;
    }
    UdmLinkInfoListAdd(&LinkInfoList, &Item->link);
  }
  if (!LinkInfoList.nitems)
  {
    /* It will not be added into LinkInfoListList, so free it now */
    UdmLinkInfoListFree(&LinkInfoList);
    goto ex;
  }

  if (UDM_OK != UdmLinkInfoListListAdd(LinkInfoListList, &LinkInfoList))
    return UDM_ERROR;

  if (LinkTextList.nitems)
    rc= UdmLinkTextListToInvertedIndexCache(Indexer, &LinkTextList, cache,
                                            ilinktext);
ex:
  UdmLinkTextListFree(&LinkTextList);
  return rc;
}


static void
UdmLinkInfoListCalcOutgoingLinks(UDM_LINKINFOLIST *List)
{
  size_t i;
  for (i= 0; i < List->nitems; i++)
  {
    UDM_LINKINFO *Item= &List->Item[i];
    Item->from->outgoing_link_count++;
    /*fprintf(stderr, "%d->%d\n", Item->from->url_id, Item->to->url_id);*/
  }
}


static void
UdmLinkInfoListCalcIncomingLinks(UDM_LINKINFOLIST *List)
{
  size_t i;
  for (i= 0; i < List->nitems; i++)
  {
    UDM_LINKINFO *Item= &List->Item[i];
    Item->to->incoming_link_count++;
  }
}


static void
UdmLinkInfoListCalcSumWeightsIncomingLinks2(UDM_LINKINFOLIST *List)
{
  size_t i;
  for (i= 0; i < List->nitems; i++)
  {
    UDM_LINKINFO *Item= &List->Item[i];
    /*Item->to->incoming_link_count++;*/
    if (Item->from->outgoing_link_count)
      Item->to->popularity1+= (Item->from->popularity0 /
                               (double) Item->from->outgoing_link_count) *
                              Item->from->server_weight;
    /*fprintf(stderr, "%d->%d\n", Item->from->url_id, Item->to->url_id);*/
  }
}


static void
UdmURLDataListNormalizePopRank(UDM_URLPOPINFOLIST *URLPopInfoList,
                               UDM_URLDATALIST *URLDataList)
{
  size_t i;
  UDM_ASSERT(URLDataList->nitems == URLPopInfoList->nitems);
  for (i= 0; i < URLPopInfoList->nitems; i++)
  {
    UDM_URLPOPINFO *src= &URLPopInfoList->Item[i];
    UDM_URLDATA *dst= &URLDataList->Item[i];
    double tmp= src->popularity0 * 1000000000;
    UDM_ASSERT(src->url_id == dst->url_id);
    if (tmp < 1)
      tmp= 1;
    dst->pop_rank= log(tmp)/log(1000000000);
    dst->per_site= (uint4) src->incoming_link_count;
    /*fprintf(stderr, "[%d] %f %f %s\n", dst->url_id, src->popularity0, dst->pop_rank, dst->url);*/
  }
}


static udm_rc_t
UdmLinkInfoListListCalcPopRankOnce(UDM_URLPOPINFOLIST *URLPopInfoList,
                                  UDM_LINKINFOLISTLIST *LinkInfoListList)
{
  size_t i;
  UDM_ASSERT(URLPopInfoList->nitems);
  for (i= 0; i < LinkInfoListList->nitems; i++)
    UdmLinkInfoListCalcSumWeightsIncomingLinks2(&LinkInfoListList->Item[i]);
  for (i= 0; i < URLPopInfoList->nitems; i++)
  {
    UDM_URLPOPINFO *Item= &URLPopInfoList->Item[i];
    Item->popularity0= (0.15 / URLPopInfoList->nitems +
                        0.85 * Item->popularity1);
    /*
    fprintf(stderr, "[%d] norm=%.6f old=%.6f new=%.6f ilinks=%d olinks=%d Wsrv=%.2f\n",
            Item->url_id, 0.15 / URLPopInfoList->nitems,
            Item->popularity1, Item->popularity0,
            (int) Item->incoming_link_count,
            (int) Item->outgoing_link_count,
            Item->server_weight);
    */
    Item->popularity1= 0; /* Prepare for the next iteration */
  }
  return UDM_OK;
}


static size_t
UdmLinkInfoListListCalcTotalLinkCount(UDM_LINKINFOLISTLIST *LinkInfoListList)
{
  size_t i, count;
  for (count=0, i= 0; i < LinkInfoListList->nitems; i++)
    count+= LinkInfoListList->Item[i].nitems;
  return count;
}


static udm_rc_t
UdmLinkInfoListListCalcPopRank(UDM_AGENT *Indexer,
                               UDM_URLPOPINFOLIST *URLPopInfoList,
                               UDM_LINKINFOLISTLIST *LinkInfoListList)
{
  size_t i;
  if (!URLPopInfoList->nitems)
    return UDM_OK;
  UdmLog(Indexer, UDM_LOG_INFO,
         "Calculating popularity: %d documents, %d links",
         (int) URLPopInfoList->nitems,
         (int) UdmLinkInfoListListCalcTotalLinkCount(LinkInfoListList));
  for (i= 0; i < LinkInfoListList->nitems; i++)
  {
    UdmLinkInfoListCalcOutgoingLinks(&LinkInfoListList->Item[i]);
    UdmLinkInfoListCalcIncomingLinks(&LinkInfoListList->Item[i]);
  }
  for (i= 0; i < 3; i++)
  {
    udm_rc_t rc;
    if (UDM_OK != (rc= UdmLinkInfoListListCalcPopRankOnce(URLPopInfoList,
                                                          LinkInfoListList)))
      return rc;
  }
  return UDM_OK;
}


typedef struct
{
  udm_secno_t ilinktext;
  udm_bool_t  use_popularity;
} UDM_ADD_LINKS_PARAM;


/**
  Link processing parameters for "indexer --rewritepop"
*/
static const UDM_ADD_LINKS_PARAM links_param_for_popularity=
{
  0,       /* Don't touch indexer link words*/
  UDM_TRUE /* Force UsePopularity if explicitly asked for */
};


static void
UdmAddLinksParamInitFromEnv(UDM_ADD_LINKS_PARAM *param, const UDM_ENV *Env)
{
  const UDM_VAR *ilinktext= UdmVarListFind(&Env->Sections, "ilinktext");
  param->ilinktext= ilinktext ? UdmVarSecno(ilinktext) : 0;
  param->use_popularity= UdmVarListFindBool(&Env->Vars, "UsePopularity", UDM_TRUE);
}


static udm_rc_t
UdmLoadRedirectLinks(UDM_AGENT *Indexer, UDM_DB *db,
                     UDM_URLDATALIST *URLList,
                     UDM_HASH *urlidhash)
{
  udm_timer_t ticks= UdmStartTimer();
  char qbuf[128];
  UDM_SQLRES SQLRes;
  udm_rc_t rc;
  size_t i;

  UdmLog(Indexer, UDM_LOG_INFO, "Loading redirects");
  udm_snprintf(qbuf, sizeof(qbuf), "SELECT url_id,url FROM redirect");
  if (UDM_OK != (rc= UdmDBSQLQuery(Indexer, db, &SQLRes, qbuf)))
   return rc;

  for (i= 0; i < UdmSQLNumRows(&SQLRes); i++)
  {
    /*
      There is a redirect from (src_id,src_url) to (dst_id,dst_url).
      We associate src_url with dst_id, so all links coming to src_url
      are considered as belonging to dst_id instead.
    */
    urlid_t srcid= atoi(UdmSQLValue(&SQLRes, i, 0));
    UDM_URLIDHASH *src, *dst, key;
    UDM_URLDATA *srcdata= UdmURLDataListSearch(URLList, srcid);
    if (!srcdata)
    {
      /* TODO34: src can be not in URLData, if "indexer -s200 --index" is give */
      continue;
    }

    UdmURLIdHashItemInit(&key, srcdata->url);
    if (!(src= (UDM_URLIDHASH*) UdmHashFind(urlidhash, &key)) || !src->id)
      continue; /* The link source is out of the indexing space */

    UdmURLIdHashItemInit(&key, UdmSQLValue(&SQLRes, i, 1));
    if (!(dst= (UDM_URLIDHASH*) UdmHashFind(urlidhash, &key)) || !dst->id)
      continue; /* The link destination is out of the indexing space */

    /*printf("FROM [%d:%s]\nTO   [%d:%s]\n",
             src->id, src->url.str, dst->id, dst->url.str);*/
    src->id= dst->id;
  }
  UdmSQLFree(&SQLRes);
  UdmLog(Indexer, UDM_LOG_INFO,
         "Loading redirects done: %d links, %.2f sec",
         (int) UdmSQLNumRows(&SQLRes), UdmStopTimer(&ticks));
  return UDM_OK;
}


/**
  Calculate popularity and add link words into the index cache.
  cache can be NULL, which means don't add link words and
  calculate popularity only.
*/
static udm_rc_t
UdmInvertedIndexAddLinkText(UDM_AGENT *Indexer,
                            UDM_DB *db,
                            UDM_INVERTED_INDEX_CACHE *cache,
                            const char *wtable,
                            UDM_URLDATALIST *URLList,
                            const UDM_ADD_LINKS_PARAM param)
{
  int i;
  udm_rc_t rc= UDM_OK;
  udm_timer_t ticks;
  UDM_HASH urlidhash;
  UDM_LINKINFOLISTLIST LinkInfoListList;
  UDM_URLPOPINFOLIST URLPopInfoList;

  UdmURLPopInfoListInit(&URLPopInfoList);
  UdmLinkInfoListListInit(&LinkInfoListList);
  if (UDM_OK != (rc= UdmURLPopInfoListInitFromURLDataList(Indexer,
                                                          &URLPopInfoList,
                                                          URLList)))
    goto ex;

  if (UDM_OK != (rc= UdmURLIdHashInitFromURLDataList(Indexer,
                                                     &urlidhash, URLList)))
    goto ex;

  if (UdmVarListFindBool(&Indexer->Conf->Vars, "ResolveRedirect", UDM_TRUE) &&
      UDM_OK != (rc= UdmLoadRedirectLinks(Indexer, db, URLList, &urlidhash)))
    goto ex;

  ticks= UdmStartTimer();
  UdmLog(Indexer, UDM_LOG_INFO, "Loading links");
  for (i= 0; i < 256; i++)
  {
    char qbuf[128];
    UDM_SQLRES SQLRes;
    if ((i % 16) == 0)
      UdmLog(Indexer, UDM_LOG_EXTRA, "- Loading links (part %d..%d)", i, i + 16 - 1);
    /*
      Add the "ilinktext" column only if cache is not NULL. Otherwise,
      we're running "indexer --rewritepop", so ilinktext is not needed.
    */
    udm_snprintf(qbuf, sizeof(qbuf),
                 "SELECT url_id,url%s FROM links WHERE seed=%d",
                 cache ? ",linktext" : "", i);
    if (UDM_OK != (rc= UdmDBSQLQuery(Indexer, db, &SQLRes, qbuf)))
      goto ex;

    rc= UdmInvertedIndexProcessLinksResult(Indexer, &urlidhash, &URLPopInfoList,
                                            cache, &LinkInfoListList,
                                            &SQLRes, param.ilinktext);
    UdmSQLFree(&SQLRes);
    if (rc != UDM_OK)
      goto ex;
  }
  UdmLog(Indexer, UDM_LOG_INFO,
         "Loading links done: %.2f sec", UdmStopTimer(&ticks));

  rc= UdmLinkInfoListListCalcPopRank(Indexer, &URLPopInfoList, &LinkInfoListList);
  UdmURLDataListNormalizePopRank(&URLPopInfoList, URLList);

ex:
  UdmURLPopInfoListFree(&URLPopInfoList);
  UdmLinkInfoListListFree(&LinkInfoListList);
  UdmHashFree(&urlidhash);
  return rc;
}


static udm_rc_t
UdmInvertedIndexRewritePopularityOneDB(UDM_AGENT *A,
                                       UDM_DB *db,
                                       UDM_QUERY *Query,
                                       const char *wtable)
{
  udm_rc_t rc;
  udm_bool_t tr= UDM_TEST(UdmSQLDBFlags(db) & UDM_SQL_HAVE_TRANSACT);
  UDM_WRITE_HELPER Helper;

  if (UDM_OK != (rc= UdmWriteHelperInit(&Helper, db)))
    goto ex2;

  if (UDM_OK != (rc= UdmInvertedIndexAddLinkText(A, db, NULL, wtable,
                                                 &Query->URLData,
                                                 links_param_for_popularity)))
    goto ex;

  if (tr && UDM_OK != (rc= UdmDBSQLBegin(A, db)))
    goto ex;

  if (UDM_OK != (rc= UdmWritePopularityBdictAndTable(A, db, &Query->URLData,
                                                     wtable, &Helper)))
    goto ex;

  if (tr && UDM_OK != (rc= UdmDBSQLCommit(A, db)))
    goto ex;

ex:
  UdmWriteHelperFree(&Helper);
ex2:
  UdmURLDataListFree(&Query->URLData);
  return rc;
}


static udm_rc_t
UdmRewritePopularity(UDM_AGENT *Indexer, UDM_DB *db, UDM_QUERY *Query)
{

  char tablename[64];
  UdmBlobGetTableForRewrite(Indexer, db, tablename, sizeof(tablename));
  return UdmInvertedIndexRewritePopularityOneDB(Indexer, db, Query, tablename);
}


/**
  Load links from the database.
  Add link words into the cache, and calculate popularity.
  Store cached link words into the index.
  TODO34: don't calculate popularity if param.use_popularity is UDM_FALSE.
  TODO34: add tests for all "UsePopularity" and "Section ilinktext" combinations
          for "indexer --index" and "indexer --rewritepop"
*/
static udm_rc_t
UdmInvertedIndexAddLinks(UDM_AGENT *Indexer, UDM_DB *db,
                         const char *wtable,
                         UDM_URLDATALIST *URLList,
                         UDM_ADD_LINKS_PARAM param)
{
  udm_rc_t rc;
  UDM_INVERTED_INDEX_CACHE cache;
  UdmInvertedIndexCacheInit(&cache, Indexer->Conf);
  if (UDM_OK != (rc= UdmInvertedIndexCacheAllocParts(Indexer, &cache,
                                                     INVERTED_INDEX_CACHE_PARTS)))
    return rc;

  if (UDM_OK != (rc= UdmInvertedIndexAddLinkText(Indexer, db, &cache,
                                                 wtable, URLList, param)))
    goto ex;

  rc= UdmInvertedIndexCacheStore(Indexer, db, wtable, &cache);

ex:
  UdmInvertedIndexCacheFree(&cache);
  return rc;
}

/***************************************************************/
static udm_rc_t
UdmCreateInvertedWordIndexFromCachedCopySQL(UDM_AGENT *Indexer,
                                            UDM_DB *db,
                                            UDM_QUERY *Query)
{
  udm_rc_t rc;
  char buf[128], wtable[64];
  int tr= (UdmSQLDBFlags(db) & UDM_SQL_HAVE_TRANSACT) ? 1 : 0;
  int tr_truncate= tr && (UdmSQLDBType(db) != UDM_DB_SYBASE);
  udm_timer_t ticks;
  udm_bool_t disable_keys= UdmVarListFindBool(UdmSQLDBVars(db), "DisableKeys", UDM_TRUE);
  UDM_WRITE_HELPER Helper;
  UDM_ADD_LINKS_PARAM param;

  UdmAddLinksParamInitFromEnv(&param, Indexer->Conf);
  if (UDM_OK != (rc= UdmWriteHelperInit(&Helper, db)))
    return UDM_ERROR;

  UdmLog(Indexer, UDM_LOG_DEBUG, Helper.use_deflate ? "Using deflate" : "Not using deflate");

  /* Get table to write to */
  if (UDM_OK != (rc= UdmBlobGetWTable(Indexer, db, wtable, sizeof(wtable))))
    goto ret2;
  /* Lock tables for MySQL */
  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
  {
    if (UdmSQLDBVersion(db) >= 40000 && disable_keys)
    {
      sprintf(buf, "ALTER TABLE %s DISABLE KEYS", wtable);
      if (UDM_OK != UdmDBSQLQuery(Indexer, db, NULL, buf))
        goto ret;
    }
    udm_snprintf(buf, sizeof(buf),
                 "LOCK TABLES "
                 "url READ,urlinfo READ,cachedcopy READ,"
                 "redirect READ,links READ,"
                 "%s WRITE",
                 wtable);
    if (UDM_OK != (rc= UdmDBSQLQuery(Indexer, db, NULL, buf)))
      goto ret2;
  }

  /* Delete old words from bdict */
  if ((tr_truncate && UDM_OK != (rc= UdmDBSQLBegin(Indexer, db))) ||
      UDM_OK != (rc= UdmDBSQLTableTruncateOrDelete(Indexer, db, wtable)) ||
      (tr_truncate && UDM_OK != (rc= UdmDBSQLCommit(Indexer, db))))
    goto ret;

  /* Convert words */

  ticks= UdmStartTimer();
  if (UDM_OK != (rc= UdmInvertedIndexCreate(Indexer, db, wtable, Query)))
    goto ret;
  if (UDM_OK != (rc= UdmInvertedIndexAddURLText(Indexer, db, wtable, &Query->URLData)))
    goto ret;
  if (param.ilinktext || param.use_popularity)
  {
    rc= param.ilinktext ?
        UdmInvertedIndexAddLinks(Indexer, db, wtable, &Query->URLData, param):
        UdmInvertedIndexAddLinkText(Indexer, db, NULL, wtable, &Query->URLData, param);
    if (rc != UDM_OK)
      goto ret;
  }
  UdmLog(Indexer, UDM_LOG_DEBUG,
         "UdmInvertedIndexCreate done: %.2f sec", UdmStopTimer(&ticks));

  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
  {
    ticks= UdmStartTimer();
    if (UDM_OK != (rc= UdmDBSQLQuery(Indexer, db, NULL, "UNLOCK TABLES")))
      goto ret2;
    udm_snprintf(buf, sizeof(buf), "LOCK TABLES %s WRITE", wtable);
    if (UDM_OK != (rc= UdmDBSQLQuery(Indexer, db, NULL, buf)))
      goto ret2;
    UdmLog(Indexer, UDM_LOG_DEBUG,
           "Unlocking tables: %.2f sec", UdmStopTimer(&ticks));
  }

  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
  {
    if (UdmSQLDBVersion(db) >= 40000 && disable_keys)
    {
      ticks= UdmStartTimer();
      UdmLog(Indexer, UDM_LOG_INFO, "Enabling SQL indexes");
      sprintf(buf, "ALTER TABLE %s ENABLE KEYS", wtable);
      UdmDBSQLQuery(Indexer, db, NULL, buf);
      UdmLog(Indexer, UDM_LOG_INFO,
             "Enabling SQL indexes done, %.2f sec", UdmStopTimer(&ticks));
    }
  }

  /* Put timestamp: note, the indexes must be already enabled here! */
  if ((tr && (UDM_OK != (rc= UdmDBSQLBegin(Indexer, db)))) ||
      (UDM_OK != (rc= UdmBlobWriteTimestamp(Indexer, db, wtable, UDM_FALSE))) ||
      (tr && (UDM_OK != (rc= UdmDBSQLCommit(Indexer, db)))))
    goto ret;

  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
    UdmDBSQLQuery(Indexer, db, NULL, "UNLOCK TABLES");

  /* Convert URL */
  ticks= UdmStartTimer();
  UdmLog(Indexer, UDM_LOG_ERROR, "Writing url data");
  if ((tr && UDM_OK != (rc= UdmDBSQLBegin(Indexer, db))) ||
      (param.use_popularity &&
       UDM_OK != (rc= UdmWritePopularityBdictAndTable(Indexer, db, &Query->URLData, wtable, &Helper))) ||
      UDM_OK != (rc= UdmBlobWriteURLData(Indexer, db, Query, wtable, &Helper)) ||
      UDM_OK != (rc= UdmBlobWriteLimitsInternal(Indexer, db, wtable, &Helper)) ||
      (tr && UDM_OK != (rc= UdmDBSQLCommit(Indexer, db))))
    goto ret2;

  UdmLog(Indexer, UDM_LOG_DEBUG,
         "Writing URL data done: %.2f sec", UdmStopTimer(&ticks));

  /* Switch to new table */
  UdmLog(Indexer, UDM_LOG_ERROR, "Rotating table");
  rc= UdmBlobSetTable(Indexer, db);
  goto ret2;

ret:
  if (UdmSQLDBType(db) == UDM_DB_MYSQL)
    UdmDBSQLQuery(Indexer, db, NULL, "UNLOCK TABLES");
ret2:
  UdmWriteHelperFree(&Helper);
  return rc;
}


/*
  Name is already escaped here, using UdmSQLEscStrSimple().
*/
static udm_rc_t
UdmBlobLoadFastURLLimitByFullName(UDM_AGENT *A,
                                  UDM_DB *db,
                                  const char *ename,
                                  UDM_URLID_LIST *buf)
{
  udm_rc_t rc= UDM_OK;
  UDM_SQLRES SQLRes;
  char qbuf[256], tablename[64], exclude;
  size_t nrows, nurls, i, row;

  exclude= buf->exclude;
  bzero((void*)buf, sizeof(*buf));
  buf->exclude= exclude;

  UdmBlobGetRTable(A, db, tablename, sizeof(tablename));
  udm_snprintf(qbuf, sizeof(qbuf),
               "SELECT coords FROM %s WHERE word LIKE '%s'", tablename, ename);
  if (UDM_OK != (rc= UdmDBSQLQuery(A, db, &SQLRes, qbuf)))
   goto ret;

  if (! (nrows= UdmSQLNumRows(&SQLRes)))
  {
    buf->empty= 1;
    goto ret;
  }
  nurls= 0;
  for (row= 0; row < nrows; row++)
    nurls+= UdmSQLLen(&SQLRes, row, 0) / 4;

  if (!(buf->urls= (urlid_t*) UdmMalloc(sizeof(urlid_t) * nurls)))
    goto ret;

  for (row= 0; row < nrows; row++)
  {
    const char *src= UdmSQLValue(&SQLRes, row, 0);
    nurls= UdmSQLLen(&SQLRes, row, 0) / 4;
    if (src && nurls)
      for (i = 0; i < nurls; i++, src+= 4)
        buf->urls[buf->nurls++]= (urlid_t) udm_get_int4(src);
  }
  UdmURLIdListSort(buf);

ret:
  UdmSQLFree(&SQLRes);
  return rc;
}


udm_rc_t
UdmBlobLoadFastURLLimit(UDM_AGENT *A, UDM_DB *db,
                        const char *name, UDM_URLID_LIST *buf)
{
  char ename[130], ename2[130];
  size_t namelen= strlen(name);
  if (namelen > 64)
    return UDM_OK;
  UdmDBSQLEscStrSimple(A, db, ename, name, namelen);
  udm_snprintf(ename2, sizeof(ename2), "##limit#%s", ename);
  return UdmBlobLoadFastURLLimitByFullName(A, db, ename2, buf);
}


static udm_rc_t
UdmBlobLoadFastOrderOrFastScore(UDM_AGENT *A, UDM_DB *db, UDM_SQLRES *SQLRes,
                                const char *prefix, const char *name)
{
  char qbuf[256], ename[256], tablename[64];
  size_t namelen= strlen(name);
  bzero((void*) SQLRes, sizeof(*SQLRes));
  if (namelen > 64)
    return UDM_OK;
  UdmDBSQLEscStrSimple(A, db, ename, name, namelen); /* Escape order name */
  UdmBlobGetRTable(A, db, tablename, sizeof(tablename));
  udm_snprintf(qbuf, sizeof(qbuf),
               "SELECT coords FROM %s WHERE word LIKE '##%s#%s'",
               tablename, prefix, ename);
  return UdmDBSQLQuery(A, db, SQLRes, qbuf);
}


static udm_rc_t
UdmBlobUnpackFastOrder(UDM_URL_INT4_LIST *List,
                       UDM_SQLRES *SQLRes,
                       size_t record_size)
{
  size_t nrows, nurls, row, param;
  udm_rc_t rc= UDM_OK;

  bzero((void*)List, sizeof(*List));

  if (!(nrows= UdmSQLNumRows(SQLRes)))
    goto ret;

  nurls= 0;
  for (row= 0; row < nrows; row++)
    nurls+= UdmSQLLen(SQLRes, row, 0) / record_size;

  if (!(List->Item= (UDM_URL_INT4*) UdmMalloc(sizeof(UDM_URL_INT4) * nurls)))
  {
    rc= UDM_ERROR;
    goto ret;
  }

  for (param= 0x7FFFFFFF, row= 0; row < nrows; row++)
  {
    const char *src= UdmSQLValue(SQLRes, row, 0);
    nurls= UdmSQLLen(SQLRes, row, 0) / record_size;
    if (src && nurls)
    {
      size_t i;
      for (i= 0; i < nurls; i++, src+= record_size)
      {
        UDM_URL_INT4 *Item= &List->Item[List->nitems++];
        Item->url_id= (urlid_t) udm_get_int4(src);
        if (record_size == 5)
          Item->param= src[4];
        else
          Item->param= --param;
      }
    }
  }
  if (List->nitems > 1)
    UdmSort(List->Item, List->nitems, sizeof(UDM_URL_INT4), (udm_qsort_cmp) UdmCmpURLID);
ret:
  return rc;
}


udm_rc_t
UdmBlobLoadFastOrder(UDM_AGENT *A, UDM_DB *db,
                     UDM_URL_INT4_LIST *List, const char *name)
{
  udm_rc_t rc= UDM_OK;
  UDM_SQLRES SQLRes;

  if (UDM_OK != (rc= UdmBlobLoadFastOrderOrFastScore(A, db, &SQLRes, "order", name)) ||
      UDM_OK != (rc= UdmBlobUnpackFastOrder(List, &SQLRes, 4)))
    goto ret;

ret:
  UdmSQLFree(&SQLRes);
  return rc;
}


udm_rc_t
UdmBlobLoadFastScore(UDM_AGENT *A, UDM_DB *db,
                     UDM_URL_INT4_LIST *List, const char *name)
{
  udm_rc_t rc= UDM_OK;
  UDM_SQLRES SQLRes;

  if (UDM_OK != (rc= UdmBlobLoadFastOrderOrFastScore(A, db, &SQLRes, "score", name)) ||
      UDM_OK != (rc= UdmBlobUnpackFastOrder(List, &SQLRes, 5)))
    goto ret;

ret:
  UdmSQLFree(&SQLRes);
  return rc;
}


static udm_rc_t
UdmWordStatCreateBlob(UDM_AGENT *A, UDM_DB *db)
{
  char qbuf[128], tablename[64], expr[64];
  UdmBlobGetTableForRewrite(A, db, tablename, sizeof(tablename));
  switch(UdmSQLDBType(db))
  {
    case UDM_DB_ORACLE8:
      udm_snprintf(expr, sizeof(expr), "lengthb(coords)");
      break;
    case UDM_DB_SQLITE3:
      udm_snprintf(expr, sizeof(expr), "length(coords)");
      break;
    case UDM_DB_MONETDB:
      /* Div by 2, to convert hex digits to real length */
      udm_snprintf(expr, sizeof(expr), "length(cast(coords as text))/2");
      break;
    case UDM_DB_MSSQL:
      udm_snprintf(expr, sizeof(expr), "datalength(coords)");
      break;
    default:
      udm_snprintf(expr, sizeof(expr), "octet_length(coords)");
  }
  udm_snprintf(qbuf, sizeof(qbuf),
               "SELECT word, sum(%s) FROM %s WHERE word NOT LIKE '##%%' GROUP BY word",
               expr, tablename);
  return UdmWordStatQuery(A, db, qbuf);
}


/*
  Dump word information to stdout.
*/
static udm_rc_t
UdmDumpWordInfoOneDocBlob(UDM_AGENT *A, UDM_DB *db, UDM_DOCUMENT *Doc)
{
  return UDM_OK;
}


static udm_rc_t
UdmBlobInitSearch(UDM_AGENT *A, UDM_DB *db, UDM_QUERY *Query,
                  UDM_FINDWORD_ARGS *args)
{
  udm_rc_t rc;
  if (!args->live_updates)
    return UDM_OK;
  if ((UDM_OK != (rc= UdmBlobLoadLiveUpdateLimitLoad(A, db, args))))
    return rc;
  return udm_dbmode_handler_rawblob.InitSearch(A, db, Query, args);
}


static udm_rc_t
UdmQueryActionBlob(UDM_AGENT *A, UDM_DB *db,
                   UDM_QUERY *Query, udm_querycmd_t cmd)
{
  switch (cmd)
  {
    case UDM_QUERYCMD_CLEAR:              return UdmTruncateDictBlob(A, db);
    case UDM_QUERYCMD_WORDSTAT:           return UdmWordStatCreateBlob(A, db);
    case UDM_QUERYCMD_REWRITE_URLDATA:    return UdmRewriteURL(A, db, Query);
    case UDM_QUERYCMD_REWRITE_LIMITS:     return UdmRewriteLimits(A, db);
    case UDM_QUERYCMD_REWRITE_POPULARITY: return UdmRewritePopularity(A, db, Query);
    case UDM_QUERYCMD_INDEX:
      return UdmCreateInvertedWordIndexFromCachedCopySQL(A, db, Query);
    default: break;
  }
  return UDM_NOTARGET;
}


const UDM_DBMODE_HANDLER udm_dbmode_handler_blob=
{
  "blob",
  UdmStoreWordsBlob,
  UdmQueryActionBlob,
  UdmDeleteWordsFromURLBlob,
  UdmFindWordBlob,
  UdmDumpWordInfoOneDocBlob,
  UdmBlobInitSearch,
};

#endif /* HAVE_SQL */