DBStorageGroupScanImpl:
CEDAR
PROGRAM
IMPORTS
DBCommon,
DBSegment,
DBStorageField,
DBStorageGroup,
DBStoragePrivate,
DBStorageTuple,
IO
EXPORTS DBStorage, DBStorageGroupScan
= BEGIN OPEN IO, DBStorageGroup, DBStorageTuple;
TID: TYPE = DBCommon.TID;
NullTID: TID = DBCommon.NullTID;
TupleObject: PUBLIC TYPE = DBCommon.TupleObject;
TupleHandle: TYPE = REF TupleObject;
Type exported to DBStorage
GroupScanObject: PUBLIC TYPE = DBStorageConcrete.GroupScanObject;
GroupScanHandle: TYPE = REF GroupScanObject;
Module global state
activeGroupScanList: GroupScanHandle;
1st item on list is a permanent header node
nullTupleHandle: TupleHandle;
Init:
PUBLIC
PROC = {
activeGroupScanList ← NEW[GroupScanObject ← [tidFieldHandle: NIL, headOfGroup: NIL]];
tempGroupScanHandle ← NEW[GroupScanObject ← [tidFieldHandle: NIL, headOfGroup: NIL]];
nullTupleHandle ← DBStorageTuple.NullTupleHandle[];
};
ReadTID:
PUBLIC
PROC[x: TupleHandle, f: DBStorage.FieldHandle]
RETURNS[TupleHandle] = {
If x belongs to an f-group, ReadTID returns the head of the group; otherwise it returns NIL.
RETURN[ReadGroupField[x, f, headTID]];
};--ReadTID
WriteTID:
PUBLIC
PROC[x: TupleHandle, r: GroupScanHandle] = {
Makes tuple x join the group specified by r at r's current position (i.e. between the tuple just read and the next tuple that would be read by NextInGroup). A NextInGroup following WriteTID returns the tuple following x; a PrevInGroup returns x itself.
Note that: (1) if x's field y had a non-NIL value before the WriteTID call, then WriteTID has the effect of deleting x from some group (using WriteTIDNIL below), (2) WriteTID cannot delete a tuple from one group without inserting it into another, so WriteTIDNil below is necessary for that case, and (3) inserting or deleting a tuple from a group requires the storage level to check all outstanding GroupScanHandles to assure that they stay consistent with the deletion. For now, this works only when inserting at the front...
f: DBStorage.FieldHandle = r.tidFieldHandle;
currentValue: TupleHandle = ReadTID[x, f];
IF currentValue # NIL AND r.headOfGroup.tid = currentValue.tid THEN RETURN; -- No-op
WriteTIDNil[x, f];
WriteGroupField[x, f, headTID, r.headOfGroup];
SELECT
TRUE
FROM
(r.prevInScan=
NIL)
AND (r.nextInScan=
NIL) =>
BEGIN
--x is only tuple in group
CreateHeadEntry[r.headOfGroup, f];
WriteHeadField[r.headOfGroup, f, firstTID, x];
WriteHeadField[r.headOfGroup, f, lastTID, x];
WriteGroupField[x, f, prevTID, nullTupleHandle];
WriteGroupField[x, f, nextTID, nullTupleHandle];
END;
(r.prevInScan=
NIL)
AND (r.nextInScan#
NIL) =>
BEGIN
--x is at front of group
WriteHeadField[r.headOfGroup, f, firstTID, x];
WriteGroupField[x, f, prevTID, nullTupleHandle];
WriteGroupField[x, f, nextTID, r.nextInScan];
WriteGroupField[r.nextInScan, f, prevTID, x];
END;
(r.prevInScan#
NIL)
AND (r.nextInScan=
NIL) =>
BEGIN
--x is at rear of group
WriteHeadField[r.headOfGroup, f, lastTID, x];
WriteGroupField[x, f, nextTID, nullTupleHandle];
WriteGroupField[x, f, prevTID, r.prevInScan];
WriteGroupField[r.prevInScan, f, nextTID, x];
END;
(r.prevInScan#
NIL)
AND (r.nextInScan#
NIL) =>
BEGIN
--x is in middle of group
WriteGroupField[x, f, nextTID, r.nextInScan];
WriteGroupField[r.nextInScan, f, prevTID, x];
WriteGroupField[x, f, prevTID, r.prevInScan];
WriteGroupField[r.prevInScan, f, nextTID, x];
END;
ENDCASE => ERROR;
BEGIN
--fix up active scans that are in identical position to scan (others are unaffected)
FOR p: GroupScanHandle ← activeGroupScanList.link, p.link
UNTIL p =
NIL
DO
IF (DBStorageField.GroupIDOfField[p.tidFieldHandle] #
DBStorageField.GroupIDOfField[r.tidFieldHandle]) OR
(~EqualTuple[p.headOfGroup,r.headOfGroup]) OR
(~EqualTuple[p.prevInScan,r.prevInScan]) THEN LOOP;
p is effectively identical to scan
p.prevInScan ← x; -- no need for explicit free here anymore ...
ENDLOOP;
END;
};--WriteTID
WriteTIDNil:
PUBLIC
PROC[x: TupleHandle, f: DBStorage.FieldHandle] = {
Makes x's TID field point to the null tuple, i.e. removes x from whatever f-group it currently belongs.
head, prev, next: TupleHandle;
head ← ReadGroupField[x, f, headTID];
IF head = NIL THEN RETURN;
prev ← ReadGroupField[x, f, prevTID];
next ← ReadGroupField[x, f, nextTID];
WriteGroupField[x, f, headTID, nullTupleHandle];
WriteGroupField[x, f, prevTID, nullTupleHandle]; --redundant
WriteGroupField[x, f, nextTID, nullTupleHandle]; --redundant
SELECT
TRUE
FROM
(prev=
NIL)
AND (next=
NIL) =>
BEGIN
--x was only tuple in group
DestroyHeadEntry[head, f];
END;
(prev=
NIL)
AND (next#
NIL) =>
BEGIN
--x was first tuple in group
WriteGroupField[next, f, prevTID, nullTupleHandle];
WriteHeadField[head, f, firstTID, next];
END;
(prev#
NIL)
AND (next=
NIL) =>
BEGIN
--x was last tuple in group
WriteGroupField[prev, f, nextTID, nullTupleHandle];
WriteHeadField[head, f, lastTID, prev];
END;
(prev#
NIL)
AND (next#
NIL) =>
BEGIN
--x was in middle of group
WriteGroupField[prev, f, nextTID, next];
WriteGroupField[next, f, prevTID, prev];
END;
ENDCASE => ERROR;
head is free now
BEGIN
--fix up active scans that contain x (others are unaffected)
FOR p: GroupScanHandle ← activeGroupScanList.link, p.link
UNTIL p =
NIL
DO
IF DBStorageField.GroupIDOfField[p.tidFieldHandle] #
DBStorageField.GroupIDOfField[f] THEN LOOP;
IF EqualTuple[p.prevInScan,x] THEN p.prevInScan ← prev;
IF EqualTuple[p.nextInScan,x] THEN p.nextInScan ← next;
ENDLOOP;
END;
prev, next are free now
};--WriteTIDNil
tempGroupScanHandle: GroupScanHandle;
SetTupleField:
PUBLIC
PROC[x: TupleHandle, f: DBStorage.FieldHandle, val: TupleHandle] = {
DBStorageField.CheckFieldHandleType[f, Group];
tempGroupScanHandle.tidFieldHandle ← f; tempGroupScanHandle.headOfGroup ← val;
tempGroupScanHandle.prevInScan ← tempGroupScanHandle.nextInScan ← NIL;
SetGroupHandle[tempGroupScanHandle, Last];
WriteTID[x, tempGroupScanHandle] };
TupleForField:
PUBLIC
PROC[x: TupleHandle, f: DBStorage.FieldHandle]
RETURNS[handle: TupleHandle] = {
DBStorageField.CheckFieldHandleType[f, Group];
tempGroupScanHandle.tidFieldHandle ← f; tempGroupScanHandle.headOfGroup ← x;
tempGroupScanHandle.prevInScan ← tempGroupScanHandle.nextInScan ← NIL;
SetGroupHandle[tempGroupScanHandle, First];
handle ← NextInGroup[tempGroupScanHandle]
};
GetGroupIDs:
PUBLIC
PROC[x: TupleHandle]
RETURNS[
LIST
OF TupleHandle] =
TRUSTED {
Returns a list containing all GroupIDs g such that z is the head of a g-group. The Tuple level is responsible for mapping from GroupIDs to FieldHandles, so that OpenScanGroup can be called when necessary.
groupList: LONG DESCRIPTOR FOR DBStorageGroup.GroupList;
cacheHint: DBCommon.CacheHandle;
result: LIST OF TupleHandle;
[groupList,cacheHint] ← GroupListFromTuple[x];
IF LENGTH[groupList] = 0 THEN result ← NIL
ELSE {
newCons, lastCons: LIST OF TupleHandle;
FOR i:
CARDINAL
IN [0..
LENGTH[groupList])
DO
newCons ←
LIST[
IF groupList[i].groupID <= DBStorage.MaxSystemTupleID
THEN
DBStoragePrivate.TupleFromTID[groupList[i].groupID]
ELSE
DBStorageTuple.ConsTupleObject[tid: groupList[i].groupID, cacheHint: NIL]];
Append the cell just created to the list we're growing.
IF i = 0 THEN result ← newCons ELSE lastCons.rest ← newCons;
lastCons ← newCons;
ENDLOOP;
};--IF
DBSegment.UnlockPage[cacheHint];
RETURN[result];
};--GetGroupIDs
SetGroupHandle:
PROC[handle: GroupScanHandle, start: DBStorage.FirstLast] =
TRUSTED {
head: TupleHandle = handle.headOfGroup;
field: DBStorage.FieldHandle = handle.tidFieldHandle;
groupList: LONG DESCRIPTOR FOR DBStorageGroup.GroupList;
cacheHint: DBCommon.CacheHandle;
[groupList, cacheHint] ← DBStorageGroup.GroupListFromTuple[head];
IF
LENGTH[groupList] # 0
THEN
BEGIN
FOR i:
CARDINAL
IN [0..
LENGTH[groupList])
DO
IF groupList[i].groupID = DBStorageField.GroupIDOfField[field] THEN GOTO FoundIt;
REPEAT
FoundIt =>
SELECT start
FROM
First =>
handle.nextInScan ← DBStorageTuple.ConsTupleObject[groupList[i].firstTID, NIL];
Last =>
handle.prevInScan ← DBStorageTuple.ConsTupleObject[groupList[i].lastTID, NIL];
ENDCASE => ERROR;
FINISHED => NULL; -- this is ok, the scan is just empty
ENDLOOP;
END;--IF
DBSegment.UnlockPage[cacheHint];
};
OpenScanGroup:
PUBLIC
PROC[x: TupleHandle, f: DBStorage.FieldHandle, start: DBStorage.FirstLast]
RETURNS [result: GroupScanHandle] =
TRUSTED {
Returns a GroupScanHandle positioned before the first (or after the last) tuple of the f-group that has tuple x as head (first or last depending upon the value of start). Returns a non-NIL GroupScanHandle even if there is no group (since the scan handle is the only way to CREATE a group).
DBStorageField.CheckFieldHandleType[f, Group];
result ← Alloc[f, x];
result.link ← activeGroupScanList.link;
activeGroupScanList.link ← result;
SetGroupHandle[result, start]
};--OpenScanGroup
Alloc: PROC[f: DBStorage.FieldHandle, x: TupleHandle] RETURNS [GroupScanHandle] =
INLINE {
RETURN[
NEW[GroupScanObject ← [
tidFieldHandle: f,
headOfGroup: x,
prevInScan: NIL, nextInScan: NIL, link: NIL]]] };
NextInGroup:
PUBLIC
PROC[r: GroupScanHandle]
RETURNS[TupleHandle] = {
Returns the tuple x immediately following the position r in a group, and advances r one position. Returns NIL if r was already positioned after the last tuple in the group.
result: TupleHandle ← NIL;
IF r=NIL THEN RETURN[NIL]; -- convenience feature
IF r.tidFieldHandle = NIL THEN ERROR DBCommon.InternalError; -- InvalidGroupScan
IF r.nextInScan#
NIL
THEN {
r.prevInScan ← result ← r.nextInScan;
r.nextInScan ← DBStorageGroup.ReadGroupField[result, r.tidFieldHandle, nextTID];
};--IF
RETURN[result];
};--NextInGroup
PrevInGroup:
PUBLIC
PROC[r: GroupScanHandle]
RETURNS[TupleHandle] = {
Returns the tuple x immediately preceding the position r in a group, and moves r
one position backwards. Returns NIL if r was already positioned before the first
tuple in the group.
result: TupleHandle ← NIL;
IF r=NIL THEN RETURN[NIL]; -- convenience feature
IF r.tidFieldHandle = NIL THEN ERROR DBCommon.InternalError; -- InvalidGroupScan
IF r.prevInScan #
NIL
THEN {
r.nextInScan ← result ← r.prevInScan;
r.prevInScan ← DBStorageGroup.ReadGroupField[result, r.tidFieldHandle, prevTID];
};--IF
RETURN[result];
};--PrevInGroup
CloseScanGroup:
PUBLIC
PROC[r: GroupScanHandle] = {
FOR scanPred: GroupScanHandle ← activeGroupScanList, scanPred.link
UNTIL scanPred=
NIL
DO
IF scanPred.link=r
THEN {
scanPred.link ← r.link; r.link ← NIL;
FreeContents[r];
RETURN;
};
REPEAT
FINISHED => ERROR DBCommon.InternalError; -- [GroupScanNotFound];
ENDLOOP;
};--CloseScanGroup
CallAfterFinishTransaction:
PUBLIC
PROC [] = {
rPrev: GroupScanHandle ← activeGroupScanList;
UNTIL rPrev.link =
NIL
DO
r: GroupScanHandle ← rPrev.link;
IF
NOT DBStorageTuple.IsValidTuple[r.headOfGroup]
THEN {
DBStorageTuple.InvalidateTuple[r.prevInScan];
DBStorageTuple.InvalidateTuple[r.nextInScan];
FreeContents[r];
rPrev.link ← r.link; r.link ← NIL;
}
ELSE rPrev ← r;
ENDLOOP;
};
FreeContents:
PROC[r: GroupScanHandle] = {
Frees all objects that are dependent on r. works even if r is invalid, i.e. FreeContents[r] has been done before.
r.tidFieldHandle ← NIL;
r.headOfGroup ← r.prevInScan ← r.nextInScan ← NIL;
};--FreeContents
CheckState:
PUBLIC
PROC[doPrinting:
BOOLEAN] = {
There isn't really much to check, except that the lists are well-formed. if this runs forever, one isn't.
activeGroupScanListLen: CARDINAL ← 0;
FOR r: GroupScanHandle ← activeGroupScanList.link, r.link
UNTIL r=
NIL
DO
activeGroupScanListLen ← activeGroupScanListLen + 1;
DBStorageField.CheckFieldHandleType[r.tidFieldHandle, Group];
IF r.headOfGroup = NIL THEN ERROR;
ENDLOOP;
IF doPrinting
THEN {
DBCommon.GetDebugStream[].PutF["list of active group scans contains %d elements*n", card[activeGroupScanListLen]];
FOR r: GroupScanHandle ← activeGroupScanList, r.link
UNTIL r=
NIL
DO
PrintGroupScanObject[r];
ENDLOOP;
DBCommon.GetDebugStream[].PutF["*n"];
};--IF
};--CheckState
PrintGroupScanObject:
PROC[scan: GroupScanHandle] =
TRUSTED {
pscan: POINTER TO GroupScanHandle = @scan;
debugStream: IO.STREAM← DBCommon.GetDebugStream[];
debugStream.PutF["groupScanHdl: %12bB, tidFieldHdl: ", card[LOOPHOLE[pscan, CARDINAL]]];
DBStorageField.PrintFieldObject[scan.tidFieldHandle];
debugStream.PutF["*n headOfGroup: "];
DBStorageTuple.PrintTupleObject[scan.headOfGroup];
debugStream.PutF["*n prevInScan: "];
DBStorageTuple.PrintTupleObject[scan.prevInScan];
debugStream.PutF["*n nextInScan: "];
DBStorageTuple.PrintTupleObject[scan.nextInScan];
debugStream.PutF["*n"];
};--PrintGroupScanObject