MODULE HashJoin;
IMPORT
  ADT:Dictionary,
  ADT:LinkedList,
  NPCT:Tools,
  Object,
  Object:Boxed,
  Out;
TYPE
  (* Some Aliases *)
  Age= Boxed.LongInt;
  Name= STRING;
  Nemesis= STRING;

  (* Generic Tuple *)
  Tuple(E1: Object.Object; E2: Object.Object) = POINTER TO TupleDesc(E1,E2);
  TupleDesc(E1: Object.Object; E2: Object.Object) = RECORD
    (Object.ObjectDesc)
    _1: E1;
    _2: E2;
  END;

  (* Relations *)
  RelationA = ARRAY 5 OF Tuple(Age,Name);
  RelationB = ARRAY 5 OF Tuple(Name,Nemesis);

VAR
  tableA: RelationA;
  tableB: RelationB;
  dict: Dictionary.Dictionary(Name,LinkedList.LinkedList(Age));
  ll: LinkedList.LinkedList(Age);

  PROCEDURE (t: Tuple(E1, E2)) INIT*(e1: E1; e2: E2);
  BEGIN
    t._1 := e1;
    t._2 := e2;
  END INIT;

  PROCEDURE DoHashPhase(t: RelationA;VAR dict: Dictionary.Dictionary(Name,LinkedList.LinkedList(Age)));
  VAR
    i: INTEGER;
    ll: LinkedList.LinkedList(Age);
  BEGIN
    i := 0;
    WHILE (i < LEN(t)) & (t[i] # NIL) DO
      IF (dict.HasKey(t[i]._2)) THEN
        ll := dict.Get(t[i]._2);
      ELSE
        ll := NEW(LinkedList.LinkedList(Age));
        dict.Set(t[i]._2,ll)
      END;
      ll.Append(t[i]._1);
      INC(i)
    END
  END DoHashPhase;

  PROCEDURE DoJoinPhase(t: RelationB; dict: Dictionary.Dictionary(Name,LinkedList.LinkedList(Age)));
  VAR
    i: INTEGER;
    age: Age;
    iterll: LinkedList.Iterator(Age);
  BEGIN
    FOR i := 0 TO LEN(t) - 1 DO
    ll := dict.Get(t[i]._1);
    iterll := ll.GetIterator(NIL);
    WHILE iterll.HasNext() DO
      age := iterll.Next();
      Out.LongInt(age.value,4);
      Out.Object(Tools.AdjustRight(t[i]._1,10));
      Out.Object(Tools.AdjustRight(t[i]._2,10));Out.Ln
    END
  END
  END DoJoinPhase;

BEGIN
  (* tableA initialization *)
  tableA[0] := NEW(Tuple(Age,Name),NEW(Age,27),"Jonah");
  tableA[1] := NEW(Tuple(Age,Name),NEW(Age,18),"Alan");
  tableA[2] := NEW(Tuple(Age,Name),NEW(Age,28),"Glory");
  tableA[3] := NEW(Tuple(Age,Name),NEW(Age,18),"Popeye");
  tableA[4] := NEW(Tuple(Age,Name),NEW(Age,28),"Alan");

  (* tableB initialization *)
  tableB[0] := NEW(Tuple(Name,Nemesis),"Jonah","Whales");
  tableB[1] := NEW(Tuple(Name,Nemesis),"Jonah","Spiders");
  tableB[2] := NEW(Tuple(Name,Nemesis),"Alan","Ghost");
  tableB[3] := NEW(Tuple(Name,Nemesis),"Alan","Zombies");
  tableB[4] := NEW(Tuple(Name,Nemesis),"Glory","Buffy");

  dict := NEW(Dictionary.Dictionary(Name,LinkedList.LinkedList(Age)));

  DoHashPhase(tableA,dict);
  DoJoinPhase(tableB,dict);
END HashJoin.