4.3BSD/usr/ingres/source/dbu/update.c
# include <stdio.h>
# include <ingres.h>
# include <resp.h>
# include <aux.h>
# include <symbol.h>
# include <access.h>
# include <batch.h>
# include <opsys.h>
# include <catalog.h>
# include <btree.h>
# include <version.h>
# include <sccs.h>
# include <errors.h>
SCCSID(@(#)update.c 8.4 2/8/85)
/*
** Update reads a batch file written by the
** access method routines (openbatch, addbatch, closebatch)
** and performs the updates stored in the file.
**
** It assumes that it is running in the database. It
** is driven by the data in the Batchhd struct (see ../batch.h).
** If the relation has a secondary index then update calls
** secupdate. As a last step the batch file is removed.
**
** The global flag Batch_recovery is tested in case
** of an error. It should be FALSE if update is being
** run as the dbu deferred update processor. It should
** be TRUE if it is being used as part of the recovery
** procedure.
*/
update()
{
register int i, mode;
DESC rel, d;
long oldtid, tupcnt;
char oldtup[MAXTUP], newtup[MAXTUP], tuple[MAXTUP];
char *batchname(), *trim_relname();
char *tp;
long bad_lid[MAXLID], lid, old_lid[MAXLID], new_lid;
char bad[MAXLID][10], *locv();
char delbtree[MAXNAME + 4], replbtree[MAXNAME + 4];
char tup_buf[3 * LIDSIZE], lid_buf[LIDSIZE];
char btree[MAXNAME + 4], out[MAXNAME + 4];
char filname[MAXNAME+4];
TID tidpos;
struct stat sbuf;
long num;
int first, end, batchcnt, j, k;
FILE *fp;
long temp;
long del_cnt;
extern int Btree_fd;
extern DESC Btreesec;
# ifdef xZTR1
if (tTf(48, -1))
printf("Update on %s\n", batchname());
# endif
/* set up to read batchhd */
Batch_cnt = BATCHSIZE; /* force a read on next getbatch */
Batch_dirty = FALSE;
if ((Batch_fp = open(batchname(), O_RDWR)) < 0)
syserr("prim:can't open %s", batchname());
getbatch(&Batchhd, sizeof Batchhd);
tupcnt = Batchhd.num_updts;
# ifdef xZTR1
if (tTf(48, 0))
printf("rel=%s tups=%ld\n", Batchhd.rel_name, tupcnt);
# endif
Resp.resp_tups = 0;
if (!tupcnt)
{
rmbatch();
return (1);
}
/* update the primary relation */
if (i = openr(&rel, OR_WRITE, Batchhd.rel_name))
syserr("prim:can't openr %s %d", Batchhd.rel_name, i);
if (rel.reldum.reldim > 0)
{
bmove(rel.relbtree, &Btreesec, sizeof(Btreesec));
Btree_fd = rel.btree_fd;
}
mode = Batchhd.mode_up;
if (rel.reldum.reldim > 0)
/* create files necessary for updating btrees in specified order */
{
concat(REPL_IN, Fileset, replbtree);
if ((Repl_infp = fopen(replbtree, "w")) == NULL)
syserr("can't open %s", replbtree);
concat(DEL_IN, Fileset, delbtree);
if ((Del_infp = fopen(delbtree, "w")) == NULL)
syserr("can't open %s", delbtree);
}
Del_cnt = 0;
for (i = 0; i < MAXLID; ++i)
{
Prev_lid[i] = 0;
Repl_cnt[i] = 0;
}
if (rel.reldum.reldim > 0)
{
if (tupcnt <= 1 || (mode != mdREPL && mode != mdAPP))
fclose(Repl_infp);
else
/* do replace's in ascending lid-value order */
{
d.reloff[1] = 0;
d.reloff[2] = Batchhd.tido_size + Batchhd.tupo_size + Batchhd.tupn_size - LIDSIZE;
for (i = 1; i <= rel.reldum.reldim; ++i)
{
d.reloff[i+2] = d.reloff[i+1] + LIDSIZE;
d.relfrmt[i+1] = INT;
d.relfrml[i+1] = LIDSIZE;
d.relgiven[i+1] = i;
}
d.relfrmt[i+1] = CHAR;
d.relfrml[i+1] = Batchhd.tidn_size;
d.relgiven[0] = 0;
d.relgiven[1] = i;
d.relgiven[i+1] = i + 1;
d.relfrmt[1] = CHAR;
d.relfrml[1] = d.reloff[2];
d.reldum.relspec = M_ORDER;
d.reldum.relatts = 2 + rel.reldum.reldim;
d.reldum.relwid = d.reloff[2] + LIDSIZE + Batchhd.tidn_size;
/* extract information about tuples from batch file */
if (stat(batchname(), &sbuf) < 0)
syserr("bad file for stat %s", batchname());
num = sbuf.st_size / (BATCHSIZE + IDSIZE);
if (num >= 1)
first = BATCHSIZE - Batch_cnt;
else
first = sbuf.st_size - sizeof Batchhd - IDSIZE;
if ((i = fwrite(&Batchbuf.bbuf[Batch_cnt], 1, first, Repl_infp)) != first)
syserr("can't write replace file");
for (i = 2; i <= num; ++i)
{
Batch_cnt = BATCHSIZE;
readbatch();
if (fwrite(Batchbuf.bbuf, 1, BATCHSIZE, Repl_infp) != BATCHSIZE)
syserr("can't write to replace file");
}
Batch_cnt = BATCHSIZE;
readbatch();
end = ((sbuf.st_size - BATCHSIZE - IDSIZE) % (BATCHSIZE + IDSIZE)) - IDSIZE;
if (end > 0)
if (fwrite(Batchbuf.bbuf, 1, end, Repl_infp) != end)
syserr("can't write to replace file 2");
fclose(Repl_infp);
sortfile(replbtree, &d, FALSE);
if ((Repl_outfp = fopen(ztack(REPL_OUT, Fileset), "r")) == NULL)
syserr("can't open replace file in update for reading\n");
concat("_SYStemp", Fileset, filname);
/* rewrite in batch file in sorted order */
if ((fp = fopen(filname, "w")) == NULL)
syserr("can't open %s", filname);
if ((k = fread(Batchbuf.bbuf, 1, first, Repl_outfp)) != first)
syserr("read error0 from replace file %d", k);
if (fwrite(Batchbuf.file_id, 1, IDSIZE, fp) != IDSIZE)
syserr("write error in batch file");
if (fwrite(&Batchhd, 1, sizeof Batchhd, fp) != sizeof Batchhd)
syserr("write error in batch file");
if (fwrite(Batchbuf.bbuf, 1, first, fp) != first)
syserr("write error in batch file");
for (i = 2; i <= num; ++i)
{
if ((k = fread(Batchbuf.bbuf, 1, BATCHSIZE, Repl_outfp)) != BATCHSIZE)
syserr("read error1 in replace file %d", k);
if (fwrite(&Batchbuf, 1, BATCHSIZE + IDSIZE, fp) != BATCHSIZE + IDSIZE)
syserr("write error into temp repl file");
}
if (end > 0)
{
if ((k = fread(Batchbuf.bbuf, 1, end, Repl_outfp)) != end)
syserr("read error2 from replace file %d", k);
if (fwrite(&Batchbuf, 1, end + IDSIZE, fp) != end + IDSIZE)
syserr("write error into temp repl file");
}
fclose(fp);
fclose(Repl_outfp);
unlink(ztack(REPL_OUT, Fileset));
rmbatch();
if (link(filname, batchname()) == -1)
syserr("can't link %s", batchname());
unlink(filname);
Batch_cnt = BATCHSIZE;
Batch_dirty = FALSE;
if ((Batch_fp = open(batchname(), O_RDWR)) < 0)
syserr("can't open new batch file");
getbatch(&Batchhd, sizeof Batchhd);
}
unlink(replbtree);
}
while (tupcnt--)
{
getbatch(&oldtid, Batchhd.tido_size); /* read old tid */
getbatch(oldtup, Batchhd.tupo_size); /* and portions of old tuple */
if (!rel.reldum.reldim)
getbatch(newtup, Batchhd.tupn_size); /* and the newtup */
else
{
if (Batchhd.tupn_size > 0)
{
getbatch(newtup, Batchhd.tupn_size - rel.reldum.reldim * LIDSIZE);
batchcnt = Batch_cnt;
tp = newtup + Batchhd.tupn_size - rel.reldum.reldim * LIDSIZE;
getbatch(tp, rel.reldum.reldim * LIDSIZE);
}
}
switch (mode)
{
case mdDEL:
if ((i = delete(&rel, &oldtid)) < 0)
syserr("prim:bad del %d %s", i, Batchhd.rel_name);
break;
case mdREPL:
if (i = replace(&rel, &oldtid, newtup, TRUE))
{
/* if newtuple is a duplicate, then ok */
if (i == 1)
{
if (rel.reldum.reldim)
++Resp.resp_tups;
break;
}
if (i == 3)
{
bmove(newtup + rel.reldum.relwid - LIDSIZE, &new_lid, LIDSIZE);
bmove(tp, bad_lid, LIDSIZE * rel.reldum.reldim);
for(j = 0; j < rel.reldum.reldim; ++j)
strcpy(bad[j], locv(bad_lid[j]));
switch (rel.reldum.reldim)
{
case 1:
nferror(BADLID1, trim_relname(rel.reldum.relid), bad[0], 0);
break;
case 2:
nferror(BADLID2, trim_relname(rel.reldum.relid), bad[0], bad[1], 0);
break;
case 3:
nferror(BADLID3, trim_relname(rel.reldum.relid), bad[0], bad[1], bad[2], 0);
break;
}
Batch_cnt = batchcnt + LIDSIZE * (rel.reldum.reldim - 1);
lid = -1;
putbatch(&lid, LIDSIZE);
break;
}
/* if this is recovery and oldtup not there, try to insert newtup */
if (Batch_recovery && i == 2)
goto upinsert;
syserr("prim:Non-functional replace on %s (%d)", i, Batchhd.rel_name);
}
Resp.resp_tups++;
break;
case mdAPP:
upinsert:
if ((i = insert(&rel, &oldtid, newtup, TRUE)) < 0)
syserr("prim:bad insert %d %s", i, Batchhd.rel_name);
if (i == 2)
{
tp = newtup + rel.reldum.relwid - rel.reldum.reldim * LIDSIZE;
bmove(tp, bad_lid, LIDSIZE * rel.reldum.reldim);
for (j = 0; j < rel.reldum.reldim; ++j)
strcpy(bad[j], locv(bad_lid[j]));
switch (rel.reldum.reldim)
{
case 1:
nferror(BADLID1, trim_relname(rel.reldum.relid), bad[0], 0);
break;
case 2:
nferror(BADLID2, trim_relname(rel.reldum.relid), bad[0], bad[1], 0);
break;
case 3:
nferror(BADLID3, trim_relname(rel.reldum.relid), bad[0], bad[1], bad[2], 0);
break;
}
oldtid = -1;
}
else if (rel.reldum.reldim > 0)
{
if (batchcnt + rel.reldum.reldim * LIDSIZE > BATCHSIZE)
{
if ((j = lseek(Batch_fp, (long) -(Batch_cnt + BATCHSIZE + 2 * IDSIZE), 1)) < 0)
syserr("Lseek error in update");
readbatch();
}
Batch_cnt = batchcnt;
tp = newtup + rel.reldum.relwid - LIDSIZE * rel.reldum.reldim;
putbatch(tp, rel.reldum.reldim * LIDSIZE);
}
break;
default:
syserr("prim:impossible mode %d", mode);
}
putbatch(&oldtid, Batchhd.tidn_size); /* write new tid if necessary */
}
if (rel.reldum.reldim > 0)
/* do deletions in decending lid-value order */
{
fclose(Del_infp);
if (Del_cnt != 0)
{
if (Del_cnt > 1)
{
d.reloff[0] = -LIDSIZE;
d.relgiven[0] = 0;
for (i = 1; i <= rel.reldum.reldim; ++i)
{
d.reloff[i] = d.reloff[i-1] + LIDSIZE;
d.relfrmt[i] = INT;
d.relfrml[i] = LIDSIZE;
d.relgiven[i] = -i;
}
d.reldum.relspec = -M_ORDER;
d.reldum.relatts = rel.reldum.reldim;
d.reldum.relwid = LIDSIZE * rel.reldum.reldim;
sortfile(delbtree, &d, TRUE);
}
btreename(rel.reldum.relid, btree);
del_cnt = Del_cnt;
if (del_cnt == 1)
concat(DEL_IN, Fileset, out);
else
concat(DEL_OUT, Fileset, out);
if ((Del_outfp = fopen(out, "r")) == NULL)
syserr("can't open delete file in update for reading\n");
while (del_cnt--)
{
if (fread(old_lid, 1, LIDSIZE * rel.reldum.reldim, Del_outfp) != LIDSIZE * rel.reldum.reldim)
syserr("tup_buf read error");
if (delete_btree(old_lid, rel.reldum.reldim) < 0)
{
printf("DELETE ERROR: %s: bad lid(s)\n", trim_relname(rel.reldum.relid));
for (i = 0; i < rel.reldum.reldim; ++i)
printf("\tlid%d=%ld\n", i + 1, old_lid[i]);
syserr("DELETE ERROR");
}
}
fclose(Del_outfp);
unlink(out);
}
unlink(delbtree);
}
/* fix the tupchanged count if delete or append */
if (mode != mdREPL)
Resp.resp_tups = rel.reladds >= 0 ? rel.reladds : -rel.reladds;
/* close the relation but secupdate will still use the decriptor */
temp = rel.reladds;
if (i = closer(&rel))
syserr("prim:close err %d %s", i, Batchhd.rel_name);
rel.reladds = temp;
batchflush();
/* if this relation is indexed, update the indexes */
if (rel.reldum.relindxd > 0)
secupdate(&rel);
if (rel.reldum.reldim > 0)
btreeupdate(&rel);
rmbatch();
# ifdef xZTR1
if (tTf(48, 2))
printf("%ld tups changed\n", Resp.resp_tups);
# endif
return (0);
}