fti-new.c

Go to the documentation of this file.

#include "server/postgres.h"
#include <ctype.h>
#include <wctype.h>
#include <wchar.h>
#include <iconv.h>
#include <errno.h>
#include "server/executor/spi.h"
#include "server/commands/trigger.h"
#include <stdlib.h>
#include <unistd.h>
#include <locale.h>
#include <time.h>


typedef struct SubStrRec {
  Oid      oid;
  wchar_t *text;
  struct SubStrRec *next;
} SubStrRec;

double codec_time=0;
double breakup_time=0;
double overall_time=0;
double ins_substr_time=0;
double lookup_substr_time=0;
double lookup_stopword_time=0;
double sql_time=0;

extern Datum fti(PG_FUNCTION_ARGS);
static wchar_t *breakup(wchar_t *, long);
static int stopword_cmp(const void *a, const void *b);
static bool is_stopword(wchar_t *);
static bool new_tuple = false;
static Oid lookupSubString(wchar_t *);
static void insertSubString(wchar_t *, Oid);
static bool ischar(const wint_t c);
static wchar_t *mb2wc(char *s, long *len);
static char *wc2mb(wchar_t *s, long *len);
static char *codec(char *s, int len, char *from, char *to, long *wclen);
static long mywcslen(wchar_t *s);
static void insert(char *column, HeapTuple t, Oid oid, char *relname);
static void del(Oid oid, char *relname);
wchar_t **stopwords=0L;
int    stopwordcount=0;
#define substrhashsize 10000
SubStrRec *substrhash[substrhashsize];

/***********************************************************************/
PG_FUNCTION_INFO_V1(fti);

Datum
fti(PG_FUNCTION_ARGS)
{
  int ret;
  if ((ret = SPI_connect()) < 0)
    elog(ERROR, "Full Text Indexing: SPI_connect failed, returned %d\n", ret);



//  nanosleep(100);
  clock_t tmptime=clock();
  stopwordcount=0;
  stopwords=NULL;
  memset(substrhash, 0, substrhashsize*sizeof(SubStrRec*));

  setlocale(LC_ALL, "de_DE@euro");
  elog(NOTICE, "====================== fti() neu =========================");
//  elog(NOTICE, "TESTTESTTESTTEST");
  TriggerData *trigdata = (TriggerData *) fcinfo->context;
  Trigger     *trigger;         /* to get trigger name */
  int          nargs;           /* # of arguments */
  char       **args;            /* arguments */
  char        *relname;         /* triggered relation name */
  Relation     rel;             /* triggered relation */
  HeapTuple    rettuple = NULL;
  TupleDesc    tupdesc;         /* tuple description */
  bool         isinsert = false;
  bool         isdelete = false;
  //  int          ret;
  Oid          oid;
  char         *column=NULL;
  
  /*
   * FILE                *debug;
   */
  
  /*
   * debug = fopen("/dev/xconsole", "w"); fprintf(debug, "FTI: entered
   * function\n"); fflush(debug);
   */
  if (!CALLED_AS_TRIGGER(fcinfo))
    elog(ERROR, "Full Text Indexing: not fired by trigger manager");
  if (TRIGGER_FIRED_FOR_STATEMENT(trigdata->tg_event))
    elog(ERROR, "Full Text Indexing: can't process STATEMENT events");
  if (TRIGGER_FIRED_BEFORE(trigdata->tg_event))
    elog(ERROR, "Full Text Indexing: must be fired AFTER event");
  
  if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
    isinsert = true;
  if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event)) {
    isdelete = true;
    isinsert = true;
  }
  if (TRIGGER_FIRED_BY_DELETE(trigdata->tg_event))
    isdelete = true;
    
  trigger = trigdata->tg_trigger;
  rel = trigdata->tg_relation;
  relname = SPI_getrelname(rel);
  rettuple = trigdata->tg_trigtuple;
  if (isdelete && isinsert) 
    rettuple = trigdata->tg_newtuple;
  
  //  if ((ret = SPI_connect()) < 0)
  //    elog(ERROR, "Full Text Indexing: SPI_connect failed, returned %d\n", ret);
  
  nargs = trigger->tgnargs;
  if (nargs != 2)
    elog(ERROR, "Full Text Indexing: trigger can only have 2 arguments");
    
  args = trigger->tgargs;
  tupdesc = rel->rd_att; 
  
  oid = HeapTupleGetOid(rettuple);
  if (!OidIsValid(oid))
    elog(ERROR, "Full Text Indexing: oid of current tuple is NULL");
  
  
  if(isdelete && isinsert) {
    Oid     *argtypes;
    int      colnum;
    
    argtypes = (Oid *) palloc(sizeof(Oid));
    argtypes[0] = VARCHAROID;  
    
    colnum = SPI_fnumber(tupdesc, args[0]);
    
    if (colnum == SPI_ERROR_NOATTRIBUTE)
      elog(ERROR, "Full Text Indexing: column '%s' not found", args[0]);
    column = SPI_getvalue(rettuple, tupdesc, colnum);
    
    char *buffer=SPI_getvalue(trigdata->tg_trigtuple, tupdesc, colnum);
    elog(NOTICE, "update");
    if(!column || (buffer && colnum && !strcmp(buffer, column)))
      return PointerGetDatum(rettuple);
  }
  
  if (isdelete) {
    Oid     *argtypes;
    int      nonum;
    
    argtypes = (Oid *) palloc(2 * sizeof(Oid));
    argtypes[0] = VARCHAROID; 
    argtypes[1] = INT4OID;

    nonum = SPI_fnumber(tupdesc, args[1]);
    if (nonum == SPI_ERROR_NOATTRIBUTE)
      elog(ERROR, "Full Text Indexing: column '%s' not found", args[1]);
    oid = strtoul(SPI_getvalue(rettuple, tupdesc, nonum), NULL, 10);
    char txt[100];
    sprintf(txt, "oid=%i", oid);
    elog(NOTICE, txt);
    del(oid, relname);
  }
  
  if (isinsert) {
    Oid     *argtypes;
    int      colnum, nonum;
    
    argtypes = (Oid *) palloc(2 * sizeof(Oid));
    argtypes[0] = VARCHAROID; 
    argtypes[1] = INT4OID;
    
    if(!column) {
      colnum = SPI_fnumber(tupdesc, args[0]);
      if (colnum == SPI_ERROR_NOATTRIBUTE)
        elog(ERROR, "Full Text Indexing: column '%s' not found", args[0]);
      column = SPI_getvalue(rettuple, tupdesc, colnum);
    }

    nonum = SPI_fnumber(tupdesc, args[1]);
    if (nonum == SPI_ERROR_NOATTRIBUTE)
      elog(ERROR, "Full Text Indexing: column '%s' not found", args[1]);
    
    oid = strtoul(SPI_getvalue(rettuple, tupdesc, nonum), NULL, 10);
    
    insert(column, rettuple, oid, relname);
  }
  
  
  overall_time+=clock()-tmptime;
  /*
  char txt[100];
  sprintf(txt, "\noverall: %f (100)\ncodec: %f (%f)\nbreakup %f (%f)\nlookup_stopword: %f (%f)\nlookup_substr: %f (%f)\nins_substr: %f (%f)\nsqltime: %f (%f)",
          overall_time,  
          codec_time, codec_time/overall_time*100, 
          breakup_time, breakup_time/overall_time*100, 
          lookup_stopword_time, lookup_stopword_time/overall_time*100, 
          lookup_substr_time, lookup_substr_time/overall_time*100, 
          ins_substr_time, ins_substr_time/overall_time*100,
          sql_time, sql_time/overall_time*100);
  elog(NOTICE, txt);
  */
  SPI_finish();  
  return PointerGetDatum(rettuple);
}


  
static void del(Oid oid, char *relname) {
  elog(NOTICE, "del()");
  char    query[8192];
  sprintf(query, "DELETE FROM search_index WHERE id=%i and tab=tabid('%s')", oid, relname);
  int ret = SPI_exec(query, 0);
  if (ret != SPI_OK_DELETE)
    elog(ERROR, "Full Text Indexing: error executing plan in delete");
}

static void insert(char *column, HeapTuple t, Oid oid, char *relname) {
  long strlen;
  char query[8192];
  wchar_t *substring;
  wchar_t *string;
  int ret;
  
  //  elog(NOTICE, "insert()");
  
  string=mb2wc(column, &strlen);
  if(string) {
    //    wchar_t   *buff=NULL;
    
    int i;
    for(i=0; i<strlen; i++) // mywcslen(string); i++)
      string[i] = towlower(string[i]);
    
    //    buff = palloc((strlen+1)*sizeof(wchar_t)); // ((mywcslen(string)+ 1)*sizeof(wchar_t));
    new_tuple = true;
    
    //    elog(NOTICE, "nach stopwords");
    
    ret = SPI_exec("select word from search_stopword order by word", 0);
    stopwordcount = SPI_processed;
    if ( ret == SPI_OK_SELECT && SPI_processed > 0 ) {
      TupleDesc tupdesc = SPI_tuptable->tupdesc;
      SPITupleTable *tuptable = SPI_tuptable;
      int j;
      stopwords=palloc(stopwordcount*sizeof(wchar_t*));
      
      for (j = 0; j < stopwordcount; j++) {
        HeapTuple tuple = tuptable->vals[j];
        wchar_t *s=mb2wc(SPI_getvalue(tuple, tupdesc, 1), NULL);
        stopwords[j]=s;
      }
    }
    
    
    while((substring = breakup(string, strlen))) {
      //      elog(NOTICE, "nach breakup()");
      Oid wordoid=InvalidOid;
      if(wcslen(substring)<4) continue;
      if(is_stopword(substring)) continue;
      if(lookupSubString(substring)!=InvalidOid) continue;
      //      elog(NOTICE, wc2mb(substring, NULL));
      
      // keyword already in database? wordoid = oid of the keyword or InvalidOid, if not in db
      sprintf(query, "select oid from search_wordlist where string='%s'", wc2mb(substring, NULL));
      ret=SPI_exec(query, 0);
      if(ret != SPI_OK_SELECT)
        elog(ERROR, "Full Text Indexing: error executing select from wordlist");
      
      if(SPI_processed==1) {
        bool null;
        wordoid=DatumGetObjectId(SPI_getbinval(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1, &null));
      }
      SPI_freetuptable(SPI_tuptable);
      
      // if keyword not in db...
      if(wordoid==InvalidOid) {

        // write keyword to db, oid result in wordoid
        sprintf(query, "insert into search_wordlist (string) values ('%s')", 
                wc2mb(substring, NULL));
        ret = SPI_exec(query, 0);
        if (ret != SPI_OK_INSERT)
          elog(ERROR, "Full Text Indexing: error executing plan in insert");
        wordoid=SPI_lastoid;

        SPI_freetuptable(SPI_tuptable);
        
        // write substrings to db
        int parent=wordoid;
        int id=0;
        if(wcslen(substring)>3) {
          for(i=1; i<wcslen(substring)-3; i++) {
            wchar_t *s=substring+i;

            // substring already in db? res to soid.
            Oid soid=InvalidOid;
            sprintf(query, "select oid from search_wordlist where string='%s'", wc2mb(s, NULL));
            ret=SPI_exec(query, 0);
            if(ret != SPI_OK_SELECT)
              elog(ERROR, "Full Text Indexing: error executing select from wordlist");
            if(SPI_processed==1) {
              bool null;
              soid=DatumGetObjectId(SPI_getbinval(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1, &null));
            }
            SPI_freetuptable(SPI_tuptable);
            
            // substring not in db => write to db
            if(soid==InvalidOid) {
              sprintf(query, "insert into search_wordlist (string) values (lower('%s'))", wc2mb(s, NULL));
              ret = SPI_exec(query, 0);
              if (ret != SPI_OK_INSERT)
                elog(ERROR, "Full Text Indexing: error executing plan in insert");
              id=SPI_lastoid;
              SPI_freetuptable(SPI_tuptable);
            }
            else id=soid;

            // insert substring-id as child to parent id.
            sprintf(query, "insert into search_parent(id, parent) values (%i, %i)", id, parent);
            ret = SPI_exec(query, 0);
            if (ret != SPI_OK_INSERT)
              elog(ERROR, "Full Text Indexing: error executing plan in insert");
            SPI_freetuptable(SPI_tuptable);

            // substring id becomes new parent id.
            parent=id;

            if(soid!=InvalidOid) break;
          }
        }
      }

      insertSubString(substring, wordoid);
      
      clock_t sqltime=clock();
      sprintf(query, "insert into search_index (string, id, tab) values (%u, %u, tabid('%s'))", wordoid, oid, relname);
      
      ret=SPI_exec(query, 0);
      if (ret != SPI_OK_INSERT)
        elog(ERROR, "Full Text Indexing: error executing plan in insert");
      SPI_freetuptable(SPI_tuptable);
      
      sql_time+=clock()-sqltime;
    }
    
    //    pfree(buff);
  }
}

static Oid lookupSubString(wchar_t *substring) {
  clock_t tmptime=clock();
  SubStrRec *r;
  int len=wcslen(substring);
  int h=0;
  int i;

  //  elog(NOTICE, "lookupSubString()");

  unsigned char *s=(unsigned char *)substring;
  for(i=0; i<len*sizeof(wchar_t); i++) h+=s[i];

  h=h%substrhashsize;
  r=substrhash[h];
  
  while(r && wcscmp(r->text, substring)!=0) {
    // elog(NOTICE, wc2mb(r->text, NULL));
    r=r->next;
  }

  lookup_substr_time+=clock()-tmptime;
  if(!r) return InvalidOid;
  else return r->oid;
}

void insertSubString(wchar_t *substring, Oid oid) {
  //  elog(NOTICE, "insertSubString()");

  clock_t tmptime=clock();
  int len=wcslen(substring);
  int h=0;
  int i;

  unsigned char *s=(unsigned char *)substring;
  for(i=0; i<len*sizeof(wchar_t); i++) h+=s[i];
  h=h%substrhashsize;

  SubStrRec *r=palloc(sizeof(SubStrRec));
  r->oid=oid;
  //  elog(NOTICE, wc2mb(substring, NULL));
  r->text=palloc((len+2)*sizeof(wchar_t));
  wcscpy(r->text, substring);
  r->next=NULL;
  if(!substrhash[h]) substrhash[h]=r;
  else {
    SubStrRec *s=substrhash[h];
    while(s->next) s=s->next;
    s->next=r;
  }
  ins_substr_time+=clock()-tmptime;
}

static wchar_t*
breakup(wchar_t *string, long strlen)
{
  //  elog(NOTICE, "breakup() start");
  // elog(NOTICE, wc2mb(string, NULL));
  clock_t tmptime=clock();
  static wchar_t *cur_pos;
  static wchar_t *substring;
  
  if (new_tuple) {
    cur_pos = string; 
    new_tuple = false;  /* don't initialize this next time */
  }
  
  while (cur_pos < string+strlen-1 && !ischar(*cur_pos)) 
    cur_pos++;
  if(cur_pos==string+strlen-1) return NULL;
  substring=cur_pos;
  while(cur_pos < string+strlen-1 && ischar(*cur_pos)) 
    cur_pos++;
  *cur_pos=L'\0';

  breakup_time+=clock()-tmptime;
  //  elog(NOTICE, wc2mb(substring, NULL));
  //  elog(NOTICE, wc2mb(cur_pos, NULL));
  return substring;
}


int stopword_cmp(const void *a, const void *b) {
  wchar_t *x=(wchar_t*)a;
  wchar_t **y=(wchar_t**)b;
  return wcscmp(x,*y);
}

/* copied from src/backend/parser/keywords.c and adjusted for our situation*/
static bool is_stopword(wchar_t *text) 
{
  clock_t tmptime=clock();
  bool ret= !(bsearch(text, stopwords, stopwordcount, sizeof(wchar_t*), stopword_cmp)==0L);
  lookup_stopword_time+=clock()-tmptime;
  return ret;
}

bool ischar(const wint_t c) {
  return iswalpha(c);
}


wchar_t *mb2wc(char *s, long *len) {
  long l;
  if(s) return (wchar_t*) codec(s, strlen(s), "UTF8", "WCHAR_T", (len?len:&l));
  else return 0L;
}

char *wc2mb(wchar_t *s, long *len) {
  long l;
  return codec((char*)s, mywcslen(s)*sizeof(wchar_t), "WCHAR_T", "UTF8", (len?len:&l));
}

char *codec(char *s, int len, char *from, char *to, long *wclen) {
  clock_t tmptime=clock();
  if(!s || len==0 || !*s) return NULL;
  iconv_t cd = iconv_open(to, from);
  size_t avail=0;
  if(cd==(iconv_t)-1) elog(INFO, strerror(errno));
  int buflen=(len+1)*sizeof(wchar_t); // +len/4;
  avail=buflen;
  char *buffer = (char *) palloc(buflen);
  char *bufpointer=buffer;
  char *spointer=s;
  size_t nconv=0;
  size_t l=len;
  
  nconv=iconv(cd, &spointer, &l, &bufpointer, &avail);
  if(nconv==(size_t)-1) {
    if(errno==EILSEQ) elog(INFO, "EILSEQ");
    if(errno==EINVAL) elog(INFO, "EINVAL");
    if(errno==EBADF) elog(INFO, "EBADF");
    elog(ERROR, strerror(errno));
  }
  
  nconv=iconv(cd, NULL, NULL, &bufpointer, &avail);

  buffer=(char*) repalloc(buffer, buflen-avail+sizeof(wchar_t));
  *(wchar_t*)bufpointer=L'\0';
  iconv_close(cd);
  codec_time+=clock()-tmptime;

  *wclen=(buflen-avail)/sizeof(wchar_t);

  return buffer;
}

static long mywcslen(wchar_t *s) {
  long i=0;
  while(s[i]!=L'\0')  i++;
  return i;
}

---