bdunion.c

免费的Sql数据库系统

/*
 *  bdunion.c  - building of an union, an intersection, a difference
 *               Kernel of GNU SQL-server   
 *
 *  This file is a part of GNU SQL Server
 *
 *  Copyright (c) 1996, 1997, Free Software Foundation, Inc
 *  Developed at the Institute of System Programming
 *  This file is written by  Vera Ponomarenko
 *
 *  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.
 *
 *  Contacts:   [e-mail]
 *
 */

/* $Id: bdunion.c,v 1.247 1997/04/10 06:57:28 vera Exp $ */

#include "destrn.h"
#include "strml.h"
#include "fdcltrn.h"

#define BD_UNION   1
#define INTERSCTN  2
#define DIFFERENCE 3

extern struct des_nseg desnseg;

static struct A outpg;
static struct des_tob *dt;

void
minsfltr (struct A *pg, struct des_tob *dt, struct des_tid *tid)
{
  u2_t offset;
  char *aspfl;

  aspfl = pg->p_shm;
  offset = ((struct p_h_f *) aspfl)->freeoff;
  if (offset + tidsize > BD_PAGESIZE)
    {
      aspfl = getptob (pg, dt);
      offset = phfsize;
    }
  *(struct des_tid *) (aspfl + offset) = *tid;
  ((struct p_h_f *) aspfl)->freeoff = offset + tidsize;
}

static char *
trunn (struct A *ppage, u2_t **aim)
{
  return getcort (ppage, aim);
}

static char *
trunnp (struct A *ppage, u2_t **aim)
{
  u2_t *afi, *ai;
  char *asptr;
  
  asptr = ppage->p_shm;
  afi = (u2_t *) (asptr + phtrsize);
  ai = *aim - 1;
  minstr (&outpg, asptr + *ai, calsc (afi, ai), dt);
  return (trunn (ppage, aim));
}

static char *
flunn (struct A *ppage, struct A *inpage, struct des_tid **tid)
{    /* get a tuple and next tid */
  char *aspfl, *asprel, *cort;
  u2_t offset, pn;
    
  aspfl = ppage->p_shm;
  offset = (char *)tid - aspfl;
  if (offset >= ((struct p_h_f *) aspfl)->freeoff)
    {
      pn = ((struct listtob *) aspfl)->nextpn;
      putwul (ppage, 'n');
      if (pn == (u2_t) ~0)
        return (NULL);
      aspfl = getwl (ppage, NRSNUM, pn);
      *tid = (struct des_tid *)(aspfl + phfsize);
    }
  pn = (*tid)->tpn;
  if (pn != inpage->p_pn)
    {
      putpg (inpage, 'n');
      while ((asprel = getpg (inpage, inpage->p_sn, pn, 's')) == NULL);
    }
  else
    asprel = inpage->p_shm;
  cort = asprel + *((u2_t *) (asprel + phsize) + (*tid)->tindex);
  (*tid)++;
  return (cort);
}

static char *
flunnp (struct A *ppage,  struct A *inpage, struct des_tid **tid)
{
  minsfltr (&outpg, dt, *tid - 1);
  return (flunn (ppage, inpage, tid));
}

static int
tstcmpun (u2_t * afn1, u2_t * afn2, u2_t * afn, u2_t kn)
{
  for (; kn != 0; kn--)
    {
      if (*afn1 != *afn2++)
	return (NCR);
      *afn++ = *afn1++;
    }
  return (OK);
}

static struct ans_ctob
bd_union (struct id_ob *pit1, struct id_ob *pit2, char * (*chng1) (),
          char * (*chng2) (), char * (*flchng1) (), char * (*flchng2) (), int set_type)
{
  struct des_field *df;
  struct des_tob *dt1, *dt2;
  u2_t fn, fdf, kn, pn, fnk, kk, type;
  u2_t *afn1, *afn2, *afn;
  i2_t n;
  int  drctn, d, v;
  char *cort1, *cort2, *asp;  
  struct A inpage1, inpage2;
  struct ans_ctob ans;
  char *arrpnt1[BD_PAGESIZE];
  u2_t arrsz1[BD_PAGESIZE];  
  char *arrpnt2[BD_PAGESIZE];
  u2_t arrsz2[BD_PAGESIZE];
  
  if (pit1->segnum != NRSNUM || pit2->segnum != NRSNUM)
    {
      ans.cpncob = NIOB;
      return (ans);
    }
  if ((u2_t) pit1->obnum > desnseg.mtobnum || (u2_t) pit2->obnum > desnseg.mtobnum)
    {
      ans.cpncob = NIOB;
      return (ans);
    }
  dt1 = (struct des_tob *) * (desnseg.tobtab + pit1->obnum);
  dt2 = (struct des_tob *) * (desnseg.tobtab + pit2->obnum);
  if (dt1 == NULL || dt2 == NULL)
    {
      ans.cpncob = NIOB;
      return (ans);
    }
  if (dt1->prdt.prsort != SORT || dt2->prdt.prsort != SORT)
    {
      ans.cpncob = N_SORT;
      return (ans);
    }
  if (dt1->prdt.prdbl != NODBL || dt2->prdt.prdbl != NODBL)
    {
      ans.cpncob = H_DBL;
      return (ans);
    }
  if (dt1->prdt.prdrctn != dt2->prdt.prdrctn)
    {
      ans.cpncob = D_DRCTN;
      return (ans);
    }
  if ((drctn = dt1->prdt.prob) == TREL && dt2->prdt.prob == TREL)
    {
      struct des_trel *dtr1, *dtr2, *destrel;
      u2_t  *ai1, *ai2, corsize1;
      struct des_field *adf, *adf1, *adf2, *df1, *df2;
      struct A inpage;
      
      dtr1 = (struct des_trel *) dt1;
      dtr2 = (struct des_trel *) dt2;
      fn = dtr1->fieldn;
      fdf = dtr1->fdftr;
      kn = dtr1->keysntr;
      adf1 = df1 = (struct des_field *) (dtr1 + 1);
      adf2 = df2 = (struct des_field *) (dtr2 + 1);
      if (fn != dtr2->fieldn || fdf != dtr2->fdftr || kn != dtr2->keysntr)
	{
	  ans.cpncob = N_EQV;
	  return (ans);
	}
      dt = gettob (&outpg, dtrsize + fn * rfsize + kn * size2b, &n, TREL);
      destrel = (struct des_trel *) dt;      
      adf = df = (struct des_field *) (destrel + 1);
      for (v = 0; v < fn; df1++, df2++, df++, v++)
	{
	  if ((type = df1->field_type) != df2->field_type)
	    {
	      ans.cpncob = N_EQV;
	      return (ans);
	    }
	  if (type == TCH || type == TFL)
	    {
	      if (df1->field_size >= df2->field_size)
		*df = *df1;
	      else
		*df = *df2;
	    }
	  else
	    *df = *df1;
	}
      afn1 = (u2_t *) df1;
      afn2 = (u2_t *) df2;
      afn = (u2_t *) df;
      if (tstcmpun (afn1, afn2, afn, kn) != OK)
	{
	  ans.cpncob = N_EQV;
	  return (ans);
	}
      asp = getwl (&inpage1, NRSNUM, dt1->firstpn);
      ai1 = (u2_t *) (asp + phtrsize);
      asp = getwl (&inpage2, NRSNUM, dt2->firstpn);
      ai2 = (u2_t *) (asp + phtrsize);
      if (drctn == GROW)
        d = 1;
      else
	d = -1;
      for (;;)
	{
	  cort1 = getcort (&inpage1, &ai1);
	  if (cort1 == NULL)
            break;
	  cort2 = getcort (&inpage2, &ai2);
	  if (cort2 == NULL)
            break;
          corsize1 = tuple_break (cort1, arrpnt1, arrsz1, adf1, fdf, fn);
	  tuple_break (cort2, arrpnt2, arrsz2, adf2, fdf, fn);
      m1:
	  for (kk = 0; kk < kn; kk++)
	    {
	      fnk = afn[kk];
	      type = (adf + fnk)->field_type;
	      if ((v = cmpfv (type, d, arrpnt1[fnk], arrpnt2[fnk])) < 0)
		{
		  if ((cort1 = (*chng1) (&inpage1, &ai1)) == NULL)
                    goto m2;
		  corsize1 = tuple_break (cort1, arrpnt1, arrsz1, adf1, fdf, fn);
		  goto m1;
		}
	      else if (v > 0)
		{
	     	  if ((cort2 = (*chng2) (&inpage2, &ai2)) == NULL)
		    goto m2;
		  tuple_break (cort2, arrpnt2, arrsz2, adf2, fdf, fn);
		  goto m1;
		}
	      }
	  if (set_type != DIFFERENCE)
            minstr (&outpg, cort1, corsize1, dt);
	}
    m2:
      if (set_type != INTERSCTN)
        {
          u2_t *afi, *ali;
          if (cort1 == NULL)
            {
              if (set_type == DIFFERENCE)
                goto m5;
              inpage1 = inpage2;
              ai1 = ai2;
            }
          asp = inpage1.p_shm;
          afi = (u2_t *) (asp + phtrsize);
          ali = afi + ((struct p_h_tr *) asp)->linptr;
          for (;;)
            {
              for (; ai1 <= ali; ai1++)
                if (*ai1 != (u2_t) 0)
                  minstr (&outpg, asp + *ai1, calsc (afi, ai1), dt);
              pn = ((struct listtob *) asp)->nextpn;
              putwul (&inpage, 'n');
              if (pn == (u2_t) ~ 0)
                break;
              asp = getwl (&inpage1, NRSNUM, pn);
              ai1 = afi = (u2_t *) (asp + phtrsize);
              ali = ai1 + ((struct p_h_tr *) asp)->linptr;
            }
        }
    m5:
      destrel->fieldn = fn;
      destrel->fdftr = fdf;
      destrel->keysntr = kn;
      dt->prdt = dt1->prdt;
    }
  else if (dt1->prdt.prob == FLTR && dt2->prdt.prob == FLTR)
    {
      struct des_fltr *dfltr1, *dfltr2, *desfltr;
      struct des_tid *tid1, *tid2;
      struct d_r_t *dr;
      struct A inpage;
      
      dfltr1 = (struct des_fltr *) dt1;
      dfltr2 = (struct des_fltr *) dt2;
      dr = dfltr1->pdrtf;
      if (dr != dfltr2->pdrtf)
	{
	  ans.cpncob = N_EQV;
	  return (ans);
	}
      fn = dr->desrbd.fieldnum;
      fdf = dr->desrbd.fdfnum;
      kn = dfltr1->keysnfl;
      if (kn != dfltr2->keysnfl)
	{
	  ans.cpncob = N_EQV;
	  return (ans);
	}
      df = (struct des_field *) (dr + 1);
      dt = gettob (&outpg, dflsize + kn * size2b, &n, FLTR);
      desfltr = (struct des_fltr *) dt;
      afn1 = (u2_t *) ((char *) (dfltr1 + 1) + dfltr1->selszfl);
      afn2 = (u2_t *) ((char *) (dfltr2 + 1) + dfltr2->selszfl);
      afn = (u2_t *) (desfltr + 1);
      if (tstcmpun (afn1, afn2, afn, kn) != OK)
	{
	  ans.cpncob = N_EQV;
	  return (ans);
	}
      if ((drctn = dt1->prdt.prdrctn) == GROW)
	d = 1;
      else
	d = -1;
      if (drctn != dt2->prdt.prdrctn)
	{
	  ans.cpncob = D_DRCTN;
	  return (ans);
	}
      asp = getwl (&inpage1, NRSNUM, dt1->firstpn);
      tid1 = (struct des_tid *) (asp + phfsize);
      asp = getwl (&inpage2, NRSNUM, dt2->firstpn);
      tid2 = (struct des_tid *) (asp + phfsize);
      
      while (getpg (&inpage, dr->segnr, tid1->tpn, 's') == NULL);
      for (;;)
	{
	  cort1 = flunn (&inpage1, &inpage, &tid1);
          if (cort1 == NULL) 
            break;
	  cort2 = flunn (&inpage2, &inpage, &tid2);
          if (cort2 == NULL)
            break;
          tuple_break (cort1, arrpnt1, arrsz1, df, fdf, fn);
	  tuple_break (cort2, arrpnt2, arrsz2, df, fdf, fn);
      m3:
	  for (kk = 0; kk < kn; kk++)
	    {
	      fnk = afn[kk];
	      type = (df + fnk)->field_type;	      
	      if ((v = cmpfv (type, d, arrpnt1[fnk], arrpnt2[fnk])) < 0)
		{
		  if ((cort1 = (*flchng1) (&inpage1, &inpage, &tid1)) == NULL)
		    goto m4;
		  tuple_break (cort1, arrpnt1, arrsz1, df, fdf, fn);
		  goto m3;
		}
	      else if (v > 0)
		{
		  if ((cort2 = (*flchng2) (&inpage2, &inpage, &tid2)) == NULL)
		    goto m4;
		  tuple_break (cort2, arrpnt2, arrsz2, df, fdf, fn);
		  goto m3;
		}
	    }
          if (set_type != DIFFERENCE)
            minsfltr (&outpg, dt1, tid1);
	}
    m4:
      if (set_type != INTERSCTN)
        {
          char *max_byte;
          if (cort1 == NULL)
            {
              if (set_type == DIFFERENCE)
                goto m6;
              inpage1 = inpage2;
              tid1 = tid2;
              dt1 = dt2;
            }
          asp = inpage1.p_shm;
          for (;;)
            {
              max_byte = asp + ((struct p_h_f *) asp)->freeoff;
              for (; (char *)tid1 < max_byte; tid1++)
                minsfltr (&outpg, dt1, tid1);
              pn = ((struct listtob *) asp)->nextpn;
              putwul (&inpage1, 'n');
              if (pn == (u2_t) ~ 0)
                break;
              asp = getwl (&inpage1, NRSNUM, pn);
            }
        }
    m6:
      putpg (&inpage, 'n');
      desfltr->pdrtf = dfltr1->pdrtf;
      desfltr->selszfl = 0;
      dt->prdt = dt1->prdt;
      dt->prdt.prsort = NSORT;
    }
  else
    {
      ans.cpncob = NIOB;
      return (ans);
    }
  putwul (&outpg, 'm');
  dt->lastpn = outpg.p_pn;
  dt->osctob = 0;
  ans.cpncob = OK;
  ans.idob.segnum = NRSNUM;
  ans.idob.obnum = n;
  return (ans);
}

struct ans_ctob
bdunion (struct id_ob *pit1, struct id_ob *pit2)
{
  return (bd_union (pit1, pit2, trunnp, trunnp, flunnp, flunnp, BD_UNION));
}

struct ans_ctob
intersctn (struct id_ob *pit1, struct id_ob *pit2)
{
  return (bd_union (pit1, pit2, trunn, trunn, flunn, flunn, INTERSCTN));
}

struct ans_ctob
differnc (struct id_ob *pit1, struct id_ob *pit2)
{
  return (bd_union (pit1, pit2, trunnp, trunn, flunnp, flunn, DIFFERENCE));
}

char *
getcort (struct A *ppage, u2_t **ai)
{     /* get a tuple and next index address */
  u2_t pn, off, *ali, *afi;
  char *asptr;

  asptr = ppage->p_shm;
  afi = (u2_t *) (asptr + phtrsize);
  ali = afi + ((struct p_h_tr *) asptr)->linptr;
  if (*ai > ali)
    {
      pn = ((struct listtob *) asptr)->nextpn;
      putwul (ppage, 'n');
      if (pn == (u2_t) ~ 0)
	return (NULL);
      asptr = getwl (ppage, NRSNUM, pn);
      *ai = (u2_t *) (asptr + phtrsize);
      ali = *ai + ((struct p_h_tr *) asptr)->linptr;
    }
  for (; *ai <= ali; (*ai)++)
    if ((off = **ai) != (u2_t) 0)
      {
	*ai += 1;
	return (asptr + off);
      }
  error ("TR.getcort: Serious error in the temporary relation page");
  return (NULL);
}

int
cmpfv (u2_t type, i4_t d, char *val1, char *val2)
{
  i4_t v;
  u2_t n1, n2;
  
  if (val1 == NULL)
    {
      if (val2 == NULL)
	return (0);
      else
	return(d);
    }
  else
    if (val2 == NULL)
      return (-d);

  switch (type)
    {
    case T1B:
      if ((v = f1b (val1, val2, size1b, size1b)) != 0)
	return (v * d);
      break;
    case T2B:
      if ((v = f2b (val1, val2, size2b, size2b)) != 0)
	return (v * d);
      break;
    case T4B:
      if ((v = f4b (val1, val2, size4b, size4b)) != 0)
	return (v * d);
      break;
    case TFLOAT:
      if ((v = flcmp (val1, val2, size4b, size4b)) != 0)
	return (v * d);
      break;
    case TFL:
      n1 = t2bunpack (val1);
      val1 += size2b;
      n2 = t2bunpack (val2);
      val2 += size2b;
      if ((v = ffloat (val1, val2, n1, n2)) != 0)
	return (v * d);
      break;
    case TCH:
      n1 = t2bunpack (val1);
      val1 += size2b;
      n2 = t2bunpack (val2);
      val2 += size2b;
      if ((v = chcmp (val1, val2, n1, n2)) != 0)
	return (v * d);
      break;
    default:
      error ("TR.cmpfv: This data type is incorrect");
      break;
    }
  return (0);
}