tctdb.c

Tokyo Cabinet的Tokyo Cabinet 是一个DBM的实现。这里的数据库由一系列key-value对的记录构成。key和value都可以是任意长度的字节序列,既可以是二进制也可以是字符

  if(!tchdbtrancommit(tdb->hdb)) err = true;
  TDBIDX *idxs = tdb->idxs;
  int inum = tdb->inum;
  for(int i = 0; i < inum; i++){
    TDBIDX *idx = idxs + i;
    switch(idx->type){
    case TDBITLEXICAL:
    case TDBITDECIMAL:
      if(!tcbdbtrancommit(idx->db)){
        tctdbsetecode(tdb, tcbdbecode(idx->db), __FILE__, __LINE__, __func__);
        err = true;
      }
      break;
    }
  }
  return !err;
}


/* Abort the transaction of a table database object.
   `tdb' specifies the table database object.
   If successful, the return value is true, else, it is false. */
static bool tctdbtranabortimpl(TCTDB *tdb){
  assert(tdb);
  bool err = false;
  if(!tchdbtranabort(tdb->hdb)) err = true;
  TDBIDX *idxs = tdb->idxs;
  int inum = tdb->inum;
  for(int i = 0; i < inum; i++){
    TDBIDX *idx = idxs + i;
    switch(idx->type){
    case TDBITLEXICAL:
    case TDBITDECIMAL:
      if(!tcbdbtranabort(idx->db)){
        tctdbsetecode(tdb, tcbdbecode(idx->db), __FILE__, __LINE__, __func__);
        err = true;
      }
      break;
    }
  }
  return !err;
}


/* Set a column index to a table database object.
   `tdb' specifies the table database object. connected as a writer.
   `name' specifies the name of a column.
   `type' specifies the index type.
   If successful, the return value is true, else, it is false. */
static bool tctdbsetindeximpl(TCTDB *tdb, const char *name, int type){
  assert(tdb && name);
  bool err = false;
  bool keep = false;
  if(type & TDBITKEEP){
    type &= ~TDBITKEEP;
    keep = true;
  }
  bool done = false;
  TDBIDX *idxs = tdb->idxs;
  int inum = tdb->inum;
  for(int i = 0; i < inum; i++){
    TDBIDX *idx = idxs + i;
    const char *path;
    if(!strcmp(idx->name, name)){
      if(keep){
        tctdbsetecode(tdb, TCEKEEP, __FILE__, __LINE__, __func__);
        return false;
      }
      if(type == TDBITOPT){
        switch(idx->type){
        case TDBITLEXICAL:
        case TDBITDECIMAL:
          if(!tcbdboptimize(idx->db, -1, -1, -1, -1, -1, UINT8_MAX)){
            tctdbsetecode(tdb, tcbdbecode(idx->db), __FILE__, __LINE__, __func__);
            err = true;
          }
          break;
        }
        done = true;
        break;
      }
      switch(idx->type){
      case TDBITLEXICAL:
      case TDBITDECIMAL:
        path = tcbdbpath(idx->db);
        if(path && unlink(path)){
          tctdbsetecode(tdb, TCEUNLINK, __FILE__, __LINE__, __func__);
          err = true;
        }
        tcbdbdel(idx->db);
        break;
      }
      TCFREE(idx->name);
      tdb->inum--;
      inum = tdb->inum;
      memmove(idxs + i, idxs + i + 1, sizeof(*idxs) * (inum - i));
      done = true;
      break;
    }
  }
  if(type == TDBITOPT || type == TDBITVOID){
    if(!done){
      tctdbsetecode(tdb, TCEINVALID, __FILE__, __LINE__, __func__);
      err = true;
    }
    return !err;
  }
  TCXSTR *pbuf = tcxstrnew();
  tcxstrprintf(pbuf, "%s%c%s%c%?", tchdbpath(tdb->hdb), MYEXTCHR, TDBIDXSUFFIX, MYEXTCHR, name);
  TCREALLOC(tdb->idxs, tdb->idxs, sizeof(tdb->idxs[0]) * (inum + 1));
  TDBIDX *idx = tdb->idxs + inum;
  int homode = tchdbomode(tdb->hdb);
  int bomode = BDBOWRITER | BDBOCREAT | BDBOTRUNC;
  if(homode & HDBONOLCK) bomode |= BDBONOLCK;
  if(homode & HDBOLCKNB) bomode |= BDBOLCKNB;
  if(homode & HDBOTSYNC) bomode |= BDBOTSYNC;
  TCCODEC enc, dec;
  void *encop, *decop;
  tchdbcodecfunc(tdb->hdb, &enc, &encop, &dec, &decop);
  int64_t bbnum = (tchdbbnum(tdb->hdb) / TDBIDXLMEMB) * 4 + TDBIDXLMEMB;
  uint8_t opts = tdb->opts;
  uint8_t bopts = 0;
  if(opts & TDBTLARGE) bopts |= BDBTLARGE;
  if(opts & TDBTDEFLATE) bopts |= BDBTDEFLATE;
  if(opts & TDBTBZIP) bopts |= BDBTBZIP;
  if(opts & TDBTTCBS) bopts |= BDBTTCBS;
  if(opts & TDBTEXCODEC) bopts |= BDBTEXCODEC;
  switch(type){
  case TDBITLEXICAL:
    idx->db = tcbdbnew();
    idx->name = tcstrdup(name);
    tcxstrprintf(pbuf, "%clex", MYEXTCHR);
    if(tdb->mmtx) tcbdbsetmutex(idx->db);
    if(enc && dec) tcbdbsetcodecfunc(idx->db, enc, encop, dec, decop);
    tcbdbtune(idx->db, TDBIDXLMEMB, TDBIDXNMEMB, bbnum, -1, -1, bopts);
    tcbdbsetcache(idx->db, tdb->lcnum, tdb->ncnum);
    tcbdbsetlsmax(idx->db, TDBIDXLSMAX);
    if(!tcbdbopen(idx->db, TCXSTRPTR(pbuf), bomode)){
      tctdbsetecode(tdb, tcbdbecode(idx->db), __FILE__, __LINE__, __func__);
      err = true;
    }
    tdb->inum++;
    break;
  case TDBITDECIMAL:
    idx->db = tcbdbnew();
    idx->name = tcstrdup(name);
    tcxstrprintf(pbuf, "%cdec", MYEXTCHR);
    if(tdb->mmtx) tcbdbsetmutex(idx->db);
    tcbdbsetcmpfunc(idx->db, tccmpdecimal, NULL);
    if(enc && dec) tcbdbsetcodecfunc(idx->db, enc, encop, dec, decop);
    tcbdbtune(idx->db, TDBIDXLMEMB, TDBIDXNMEMB, bbnum, -1, -1, bopts);
    tcbdbsetcache(idx->db, tdb->lcnum, tdb->ncnum);
    tcbdbsetlsmax(idx->db, TDBIDXLSMAX);
    if(!tcbdbopen(idx->db, TCXSTRPTR(pbuf), bomode)){
      tctdbsetecode(tdb, tcbdbecode(idx->db), __FILE__, __LINE__, __func__);
      err = true;
    }
    tdb->inum++;
    break;
  default:
    tctdbsetecode(tdb, TCEINVALID, __FILE__, __LINE__, __func__);
    err = true;
    break;
  }
  idx->type = type;
  if(!err){
    TCHDB *hdb = tdb->hdb;
    if(!tchdbiterinit(hdb)) err = true;
    void *db = idx->db;
    TCXSTR *kxstr = tcxstrnew();
    TCXSTR *vxstr = tcxstrnew();
    int nsiz = strlen(name);
    while(tchdbiternext3(hdb, kxstr, vxstr)){
      if(nsiz < 1){
        const char *kbuf = TCXSTRPTR(kxstr);
        int ksiz = TCXSTRSIZE(kxstr);
        switch(type){
        case TDBITLEXICAL:
        case TDBITDECIMAL:
          if(!tcbdbput(db, kbuf, ksiz, kbuf, ksiz)){
            tctdbsetecode(tdb, tcbdbecode(db), __FILE__, __LINE__, __func__);
            err = true;
          }
          break;
        }
      } else {
        int vsiz;
        char *vbuf = tcmaploadone(TCXSTRPTR(vxstr), TCXSTRSIZE(vxstr), name, nsiz, &vsiz);
        if(vbuf){
          switch(type){
          case TDBITLEXICAL:
          case TDBITDECIMAL:
            if(!tcbdbputdup(db, vbuf, vsiz, TCXSTRPTR(kxstr), TCXSTRSIZE(kxstr))){
              tctdbsetecode(tdb, tcbdbecode(db), __FILE__, __LINE__, __func__);
              err = true;
            }
            break;
          }
          TCFREE(vbuf);
        }
      }
    }
    tcxstrdel(vxstr);
    tcxstrdel(kxstr);
  }
  tcxstrdel(pbuf);
  return !err;
}


/* Generate a unique ID number.
   `tdb' specifies the table database object.
   `inc' specifies the increment of the seed.
   The return value is the new unique ID number or -1 on failure. */
static int64_t tctdbgenuidimpl(TCTDB *tdb, int64_t inc){
  assert(tdb);
  void *opq = tchdbopaque(tdb->hdb);
  uint64_t llnum, uid;
  if(inc < 0){
    uid = -inc - 1;
  } else {
    memcpy(&llnum, opq, sizeof(llnum));
    if(inc == 0) return TCITOHLL(llnum);
    uid = TCITOHLL(llnum) + inc;
  }
  llnum = TCITOHLL(uid);
  memcpy(opq, &llnum, sizeof(llnum));
  return uid;
}


/* Execute the search of a query object.
   `qry' specifies the query object.
   The return value is a list object of the primary keys of the corresponding records. */
static TCLIST *tctdbqrysearchimpl(TDBQRY *qry){
  assert(qry);
  TCTDB *tdb = qry->tdb;
  TCHDB *hdb = tdb->hdb;
  TDBIDX *idxs = tdb->idxs;
  int inum = tdb->inum;
  TDBCOND *conds = qry->conds;
  int cnum = qry->cnum;
  int acnum = cnum;
  int max = qry->max;
  if(max < INT_MAX - qry->skip) max += qry->skip;
  const char *oname = qry->oname;
  int otype = qry->otype;
  TCXSTR *hint = qry->hint;
  TCLIST *res = NULL;
  for(int i = 0; i < cnum; i++){
    TDBCOND *cond = conds + i;
    cond->alive = true;
  }
  tcxstrclear(hint);
  bool isord = oname != NULL;
  TDBCOND *mcond = NULL;
  TDBIDX *midx = NULL;
  TDBCOND *ncond = NULL;
  TDBIDX *nidx = NULL;
  TDBCOND *scond = NULL;
  TDBIDX *sidx = NULL;
  for(int i = 0; i < cnum; i++){
    TDBCOND *cond = conds + i;
    if(!cond->sign || cond->noidx) continue;
    for(int j = 0; j < inum; j++){
      TDBIDX *idx = idxs + j;
      if(!strcmp(cond->name, idx->name)){
        switch(idx->type){
        case TDBITLEXICAL:
          switch(cond->op){
          case TDBQCSTREQ:
          case TDBQCSTRBW:
          case TDBQCSTROREQ:
            if(!mcond){
              mcond = cond;
              midx = idx;
            } else if(!ncond){
              ncond = cond;
              nidx = idx;
            }
            break;
          default:
            if(!scond){
              scond = cond;
              sidx = idx;
            }
            break;
          }
          break;
        case TDBITDECIMAL:
          switch(cond->op){
          case TDBQCNUMEQ:
          case TDBQCNUMGT:
          case TDBQCNUMGE:
          case TDBQCNUMLT:
          case TDBQCNUMLE:
          case TDBQCNUMBT:
          case TDBQCNUMOREQ:
            if(!mcond){
              mcond = cond;
              midx = idx;
            } else if(!ncond){
              ncond = cond;
              nidx = idx;
            }
            break;
          default:
            if(!scond){
              scond = cond;
              sidx = idx;
            }
            break;
          }
          break;
        }
      }
    }
  }
  if(mcond){
    res = tclistnew();
    mcond->alive = false;
    acnum--;
    TCMAP *nmap = NULL;
    if(ncond){
      ncond->alive = false;
      acnum--;
      nmap = tctdbqryidxfetch(qry, ncond, nidx);
      max = tclmin(max, tcmaprnum(nmap));
    }
    const char *expr = mcond->expr;
    int esiz = mcond->esiz;
    TDBCOND *ucond = NULL;
    for(int i = 0; i < cnum; i++){
      TDBCOND *cond = conds + i;
      if(!cond->alive) continue;
      if(ucond){
        ucond = NULL;
        break;
      }
      ucond = cond;
    }
    TCLIST *keys, *tokens;
    BDBCUR *cur;
    const char *kbuf, *pv;
    char numbuf[TCNUMBUFSIZ];
    int knum, ksiz, nsiz, tnum;
    int64_t xnum, lower, upper;
    bool all;
    switch(mcond->op){
    case TDBQCSTREQ:
      tcxstrprintf(hint, "using an index: \"%s\" one (STREQ)\n", mcond->name);
      if(oname && !strcmp(oname, mcond->name) &&
         (otype == TDBQOSTRASC || otype == TDBQOSTRDESC)) oname = NULL;
      keys = tcbdbget4(midx->db, expr, esiz);
      if(keys){
        knum = TCLISTNUM(keys);
        all = oname != NULL;
        if(!all && max < INT_MAX) tcxstrprintf(hint, "limited matching: %d\n", max);
        for(int i = 0; (all || TCLISTNUM(res) < max) && i < knum; i++){
          TCLISTVAL(kbuf, keys, i, ksiz);
          if(!nmap || tcmapget(nmap, kbuf, ksiz, &nsiz)){
            if(acnum < 1){
              TCLISTPUSH(res, kbuf, ksiz);
            } else if(ucond){
              if(tctdbqryonecondmatch(qry, ucond, kbuf, ksiz)) TCLISTPUSH(res, kbuf, ksiz);
            } else if(tctdbqryallcondmatch(qry, kbuf, ksiz)){
              TCLISTPUSH(res, kbuf, ksiz);
            }
          }
        }
        tclistdel(keys);
      }
      break;
    case TDBQCSTRBW:
      tcxstrprintf(hint, "using an index: \"%s\" asc (STRBW)\n", mcond->name);
      cur = tcbdbcurnew(midx->db);
      tcbdbcurjump(cur, expr, esiz);
      all = oname && (strcmp(oname, mcond->name) || otype != TDBQOSTRASC);
      if(!all && max < INT_MAX) tcxstrprintf(hint, "limited matching: %d\n", max);
      while((all || TCLISTNUM(res) < max) && (kbuf = tcbdbcurkey3(cur, &ksiz)) != NULL){
        if(ksiz >= esiz && !memcmp(kbuf, expr, esiz)){
          int vsiz;
          const char *vbuf = tcbdbcurval3(cur, &vsiz);
          if(!nmap || tcmapget(nmap, vbuf, vsiz, &nsiz)){
            if(acnum < 1){
              TCLISTPUSH(res, vbuf, vsiz);
            } else if(ucond){
              if(tctdbqryonecondmatch(qry, ucond, vbuf, vsiz)) TCLISTPUSH(res, vbuf, vsiz);
            } else if(tctdbqryallcondmatch(qry, vbuf, vsiz)){
              TCLISTPUSH(res, vbuf, vsiz);
            }
          }
        } else {
          break;
        }
        tcbdbcurnext(cur);
      }
      tcbdbcurdel(cur);
      if(oname && !strcmp(oname, mcond->name)){
        if(otype == TDBQOSTRASC){
          oname = NULL;
        } else if(otype == TDBQOSTRDESC){
          tclistinvert(res);
          oname = NULL;
        }
      }
      break;
    case TDBQCSTROREQ:
      tcxstrprintf(hint, "using an index: \"%s\" skip (STROREQ)\n", mcond->name);
      tokens = tcstrsplit(expr, " ,");
      tclistsort(tokens);
      if(oname && !strcmp(oname, mcond->name)){
        if(otype == TDBQOSTRASC){
          oname = NULL;
        } else if(otype == TDBQOSTRDESC){
          tclistinvert(tokens);
          oname = NULL;
        }
      }
      tnum = TCLISTNUM(tokens);
      all = oname != NULL;
      if(!all && max < INT_MAX) tcxstrprintf(hint, "limited matching: %d\n", max);
      for(int i = 0; (all || TCLISTNUM(res) < max) && i < tnum; i++){
        const char *token;
        int tsiz;
        TCLISTVAL(token, tokens, i, tsiz);
        keys = tcbdbget4(midx->db, token, tsiz);
        if(keys){
          knum = TCLISTNUM(keys);
          for(int j = 0; (all || TCLISTNUM(res) < max) && j < knum; j++){
            TCLISTVAL(kbuf, keys, j, ksiz);
            if(!nmap || tcmapget(nmap, kbuf, ksiz, &nsiz)){
              if(acnum < 1){
                TCLISTPUSH(res, kbuf, ksiz);
              } else if(ucond){
                if(tctdbqryonecondmatch(qry, ucond, kbuf, ksiz)) TCLISTPUSH(res, kbuf, ksi