2.11BSD#431 - /usr/ingres/source/decomp/call_ovqp.c

   1: # include   "../pipes.h"
   2: # include   "../ingres.h"
   3: # include   "../aux.h"
   4: # include   "../unix.h"
   5: # include   "../tree.h"
   6: # include   "../symbol.h"
   7: # include   "decomp.h"
   8: 
   9: 
  10: /*
  11: ** CALL_OVQP -- Routines which interface to the One Variable Query Processor.
  12: **
  13: **	This file contains the routines associated with sending queries
  14: **	and receiving results from OVQP. The interface to these routines is
  15: **	still messy. Call_ovqp is given the query, mode, and result relation
  16: **	as parameters and gets the source relation, and two flags
  17: **	(Newq, Newr) as globals. The routines include:
  18: **
  19: **	Call_ovqp -- Sends a One-var query to ovqp, flushes the pipe,
  20: **			and reads result from ovqp.
  21: **
  22: **	Endovqp    -- Informs ovqp that the query is over. Helps to synchronize
  23: **			the batch file (if any).
  24: **
  25: **	Pipesym    -- Writes a token symbol to ovqp.
  26: */
  27: 
  28: struct pipfrmt Outpipe, Inpipe;
  29: struct retcode  Retcode;    /* return code structure */
  30: 
  31: 
  32: 
  33: call_ovqp(qtree, mode, resultnum)
  34: struct querytree    *qtree;
  35: int         mode;
  36: int         resultnum;
  37: 
  38: /*
  39: ** Call_ovqp -- send query down pipe to ovqp and flush pipe.
  40: **	Inputs are:
  41: **		mode		retrieve, append, etc.
  42: **		resultnum	result relation id
  43: **		qtree		the query
  44: **		Sourcevar	(global) if >= 0 then source var
  45: **		Newq		send NEWQ symbol
  46: **		Newr		send NEWR symbol
  47: */
  48: 
  49: {
  50:     register struct querytree   *tree;
  51:     register int            i;
  52:     char                *rangename();
  53:     struct retcode          ret;
  54:     struct symbol           s;
  55:     int             j;
  56: 
  57:     tree = qtree;
  58: #	ifdef xDTR1
  59:     if (tTf(8, -1))
  60:     {
  61:         if (tTf(8, 0))
  62:             printf("CALL_OVQP-\n");
  63:         if (tTf(8, 1))
  64:         {
  65:             if (Newq)
  66:                 printree(tree, "new query to ovqp");
  67:             else
  68:                 printf("query same as previous\n");
  69:         }
  70:         if (tTf(8, 2))
  71:         {
  72:             printf("Sourcevar=%d\t", Sourcevar);
  73:             if (Sourcevar >= 0)
  74:                 printf("relid=%s\t", rangename(Sourcevar));
  75:             if (resultnum >= 0)
  76:                 printf("Resultname=%s", rnum_convert(resultnum));
  77:             if (((struct qt_root *) tree)->rootuser)
  78:                 printf(", userquery");
  79:             printf("\n");
  80:         }
  81:     }
  82: #	endif
  83: 
  84: 
  85: 
  86:     /*  header information needed by OVQP for each query  */
  87:     wrpipe(P_PRIME, &Outpipe, EXEC_OVQP, 0, 0);
  88:     pipesym(QMODE, 1, mode);
  89: 
  90:     if (Newq)
  91:         pipesym(CHANGESTRAT, 0);
  92: 
  93:     if (Newr)
  94:     {
  95:         pipesym(REOPENRES, 0);
  96:         Newr = FALSE;
  97:     }
  98: 
  99:     if (Sourcevar >= 0)
 100:     {
 101:         s.type = SOURCEID;
 102:         s.len = MAXNAME;
 103:         wrpipe(P_NORM, &Outpipe, W_ovqp, &s, 2);
 104:         wrpipe(P_NORM, &Outpipe, W_ovqp, rangename(Sourcevar), MAXNAME);
 105:     }
 106: 
 107:     if (resultnum >= 0)
 108:     {
 109:         s.type = RESULTID;
 110:         s.len = MAXNAME;
 111:         wrpipe(P_NORM, &Outpipe, W_ovqp, &s, 2);
 112:         wrpipe(P_NORM, &Outpipe, W_ovqp, rnum_convert(resultnum), MAXNAME);
 113:     }
 114: 
 115:     /*	this symbol, USERQRY, should be sent last because OVQP
 116: 	 *	requires the query mode and result relation when it
 117: 	 *	encounters USERQRY.
 118: 	 */
 119: 
 120:     if (((struct qt_root *) tree)->rootuser)
 121:     {
 122:         pipesym(USERQRY, 0);
 123:     }
 124: 
 125:     /*  now write the query list itself  */
 126: 
 127:     pipesym(TREE, 0);
 128: 
 129:     if (tree->sym.type == AGHEAD)
 130:     {
 131:         pipesym(AGHEAD, 0);
 132:         if (tree->left->sym.type == BYHEAD)
 133:         {
 134:             mklist(tree->left->right);
 135:             pipesym(BYHEAD, 0);
 136:             mklist(tree->left->left);
 137:         }
 138:         else
 139:             mklist(tree->left);
 140:     }
 141:     else
 142:         mklist(tree->left);
 143: 
 144:     pipesym(ROOT, 0);
 145:     mklist(tree->right);
 146:     pipesym(QLEND, 0);
 147: 
 148:     /* now flush the query to ovqp */
 149:     wrpipe(P_FLUSH, &Outpipe, W_ovqp);
 150: 
 151:     /*
 152: 	** Read the result of the query from ovqp. A Retcode
 153: 	** struct is expected.
 154: 	** The possble values of Retcode.rt_status are:
 155: 	**	EMPTY     -- no tuples satisfied
 156: 	**	NONEMPTY  -- at least one tuple satisfied
 157: 	**	error num -- ovqp user error
 158: 	*/
 159: 
 160: 
 161:     rdpipe(P_PRIME, &Inpipe);
 162:     if (rdpipe(P_NORM, &Inpipe, R_ovqp, &ret, sizeof (ret)) != sizeof (ret))
 163:         syserr("readresult: read error on R_ovqp");
 164:     j = ret.rc_status;
 165:     switch (j)
 166:     {
 167:       case EMPTY:
 168:         i = FALSE;
 169:         break;
 170: 
 171:       case NONEMPTY:
 172:         i = TRUE;
 173:         if (((struct qt_root *) tree)->rootuser)
 174:             Retcode.rc_tupcount = ret.rc_tupcount;
 175:         break;
 176: 
 177:       default:
 178:         derror(j);
 179: 
 180:     }
 181:     return (i);
 182: }
 183: 
 184: 
 185: 
 186: readagg_result(result)
 187: struct querytree    *result[];
 188: 
 189: /*
 190: ** Readagg_result -- read results from ovqp from a simple aggregate.
 191: */
 192: 
 193: {
 194:     register struct querytree   **r, *aop;
 195:     register int            vallen;
 196:     int             i;
 197: 
 198:     r = result;
 199:     while (aop = *r++)
 200:     {
 201:         vallen = aop->sym.len & I1MASK;
 202:         if (rdpipe(P_NORM, &Inpipe, R_ovqp, &aop->sym, 2) != 2)
 203:             syserr("ageval:rd err (1)");
 204:         if (vallen != (aop->sym.len & I1MASK))
 205:             syserr("ageval:len %d,%d", vallen, aop->sym.len);
 206:         if ((i = rdpipe(P_NORM, &Inpipe, R_ovqp, aop->sym.value, vallen)) != vallen)
 207:             syserr("ageval:rd err %d,%d", i, vallen);
 208: #		ifdef xDTR1
 209:         if (tTf(8, 3))
 210:             writenod(aop);
 211: #		endif
 212:     }
 213: }
 214: 
 215: 
 216: 
 217: 
 218: endovqp(ack)
 219: int ack;
 220: 
 221: /*
 222: ** Endovqp -- Inform ovqp that processing is complete. "Ack" indicates
 223: **	whether to wait for an acknowledgement from ovqp. The overall
 224: **	mode of the query is sent followed by an EXIT command.
 225: **
 226: **	Ovqp decides whether to use batch update or not. If ack == ACK
 227: **	then endovqp will read a RETVAL symbol from ovqp and return
 228: **	a token which specifies whether to call the update processor or not.
 229: */
 230: 
 231: {
 232:     register int        done, j;
 233:     struct retcode      ret;
 234:     extern int      Qry_mode;
 235:     extern struct descriptor Inddes;
 236: 
 237:     if (ack != RUBACK)
 238:     {
 239:         wrpipe(P_PRIME, &Outpipe, EXEC_OVQP, 0, 0);
 240: 
 241:         /* send mode of the query */
 242:         pipesym(QMODE, 1, Qry_mode);
 243: 
 244:         /* send exit code */
 245:         pipesym(EXIT, 1, ack);
 246: 
 247:         wrpipe(P_END, &Outpipe, W_ovqp);
 248:     }
 249: 
 250:     j = NOUPDATE;
 251:     if (ack == ACK)
 252:     {
 253:         /* read acknowledgement from ovqp */
 254:         rdpipe(P_PRIME, &Inpipe);
 255:         if (rdpipe(P_NORM, &Inpipe, R_ovqp, &ret, sizeof (ret)) != sizeof (ret))
 256:             syserr("endovqp no retcode");
 257: 
 258:         j = ret.rc_status;
 259:         if (j != UPDATE && j != NOUPDATE)
 260:             syserr("endovqp:%d", j);
 261: 
 262:         rdpipe(P_SYNC, &Inpipe, R_ovqp);
 263:     }
 264: 
 265:     cleanrel(&Inddes);  /* could be closecatalog(FALSE) except for
 266: 				** text space limitations */
 267:     return (j);
 268: }
 269: 
 270: 
 271: 
 272: 
 273: 
 274: 
 275: pipesym(token, length, val)
 276: int token;
 277: int length;
 278: int val;
 279: {
 280:     register struct symbol  *s;
 281:     char            temp[4];
 282:     register int        i;
 283: 
 284:     s = (struct symbol *) temp;
 285:     i = length;
 286: 
 287:     s->type = token;
 288:     if (s->len = i)
 289:         s->value[0] = val;
 290: 
 291:     wrpipe(P_NORM, &Outpipe, W_ovqp, s, i + 2);
 292: }
 293: 
 294: ovqpnod(node)
 295: struct querytree    *node;
 296: {
 297: 
 298:     register int            typ;
 299:     register struct querytree   *q;
 300:     register struct symbol      *s;
 301:     int             i;
 302:     char                temp[4];
 303:     struct querytree        *ckvar();
 304: 
 305:     q = node;
 306:     s = (struct symbol *) temp;
 307: 
 308:     typ = q->sym.type;
 309:     s->type = typ;
 310: 
 311:     switch (typ)
 312:     {
 313:       case VAR:
 314:         q = ckvar(q);
 315: 
 316:         /*  check to see if this variable has been substituted for.
 317: 		**  if so, pass the constant value to OVQP		      */
 318: 
 319:         if (((struct qt_var *)q)->valptr)
 320:         {
 321:             s->type = ((struct qt_var *)q)->frmt;
 322:             s->len = ((struct qt_var *)q)->frml;
 323:             wrpipe(P_NORM, &Outpipe, W_ovqp, s, 2);
 324:             wrpipe(P_NORM, &Outpipe, W_ovqp, ((struct qt_var *)q)->valptr, s->len & 0377);
 325:             return;
 326:         }
 327:         else
 328:         {
 329:             /* double check that variable is sourcevar */
 330:             if (((struct qt_var *)q)->varno != Sourcevar)
 331:                 syserr("mklist:src=%d,var=%d", Sourcevar, ((struct qt_var *)q)->varno);
 332:             s->len = 1;
 333:             s->value[0] = ((struct qt_var *)q)->attno;
 334:             break;
 335:         }
 336: 
 337:       case TREE:
 338:         return;
 339: 
 340:       case AND:
 341:       case OR:
 342:         s->len = 0;
 343:         break;
 344: 
 345:       case BOP:
 346:       case RESDOM:
 347:       case UOP:
 348:         s->len = 1;
 349:         s->value[0] = *(q->sym.value);
 350:         break;
 351: 
 352:       default:
 353:         i = (q->sym.len & 0377) + 2;
 354:         wrpipe(P_NORM, &Outpipe, W_ovqp, &q->sym, i);
 355:         return;
 356:     }
 357: 
 358:     wrpipe(P_NORM, &Outpipe, W_ovqp, s, s->len + 2);
 359: }
 360: 
 361: /*
 362: **	Use "inpcloser()" for closing relations. See
 363: **	desc_close in openrs.c for details.
 364: */
 365: extern int  inpcloser();
 366: int     (*Des_closefunc)()  = inpcloser;
 367: 
 368: 
 369: init_decomp()
 370: {
 371:     /* this is a null routine called from main() */
 372: }
 373: 
 374: 
 375: startdecomp()
 376: {
 377:     /* called at the start of each user query */
 378:     Retcode.rc_tupcount = 0;
 379:     initrange();
 380:     rnum_init();
 381: }
 382: 
 383: 
 384: /*
 385: ** Files_avail -- returns how many file descriptors are available
 386: **	for decomposition. For decomp running by itself the fixed
 387: **	overhead for files is:
 388: **
 389: **		4 -- pipes
 390: **		1 -- relation relation
 391: **		1 -- attribute relation
 392: **		1 -- indexes relation
 393: **		1 -- standard output
 394: **		1 -- concurrency
 395: */
 396: 
 397: files_avail(mode)
 398: int mode;
 399: {
 400:     return (MAXFILES - 9);
 401: }

Defined functions

Defined variables