spark/src/DataStreamSuite/DataStream/Primitives/LeftJoin/LeftInternalJoinStream.cs

using System;
using System.Collections;
using BCG.Utilities;

namespace BCG
{
   /// <summary>
   /// Stream to internally handle Left Joins. The only difference between this and the left Join is the waty the
   /// DataPacket thats fired is initialized.
   /// </summary>
   [DataStreamUserRespresentation (false)]
   public class LeftInternalJoinStream : InternalJoinStream
   {
      protected DataPacket myLeftDataPacket, myRightDataPacket;
      //Container to contain all the left rows that are orphaned during a deletion operation
      private IDictionary myLeftOrphanedRows;
      public LeftInternalJoinStream (ViewStream vsLeft, string jcLeft, ViewStream vsRight, string jcRight) :
         base (vsLeft, jcLeft, vsRight, jcRight)   {}

      protected override void myOnFirstSubscriber() 
      {
         myLeftOperand.master = this;
         myRightOperand.master = this;
         //the events are not hooked onto the operands delegate, but rather a level of indirection
         //added in the form on myLeftChangeHandler
         myLeftStream.onChange += myLeftChangeHandler;
         myRightStream.onChange += myRightChangeHandler;
      }
       
      protected override void myOnLastSubscriber () 
      {
         myInit ();
         myLeftStream.onChange -= myLeftChangeHandler;
         myRightStream.onChange -= myRightChangeHandler;
      }

      protected override void myInit()
      {
         base.myInit ();
         myLeftOrphanedRows = null;
         myLeftDataPacket = myRightDataPacket = null;
      }


      private void myLeftChangeHandler (object o)
      {
         myLeftDataPacket = (DataPacket)o;
         myTestAndFire (true);       
      }

      private void myRightChangeHandler (object o)
      {
         myRightDataPacket = (DataPacket)o;
         myTestAndFire (false);
      }

      /// <summary>
      /// enforces the order of firing the events, this is to enforce that first the right operand
      /// has data (possibly 0 rows) before we the join is actually calculated => Performance boost
      /// </summary>
      /// <param name="isLeft"></param>
      private void myTestAndFire (bool isLeft)
      {
         JoinOperand jo = isLeft ? myLeftOperand : myRightOperand;
         DataPacket dp = isLeft ? myLeftDataPacket : myRightDataPacket;
         //to test whether both have received call backs
         if (myLeftOperand.hasReceivedData)
            jo.changeHandler (dp);
         else if (myLeftDataPacket != null && myRightDataPacket != null)
         {
            myRightOperand.changeHandler (myRightDataPacket);
            myLeftOperand.changeHandler (myLeftDataPacket);
         }
      }

      /// <summary>
      /// Called by JoinOperand
      /// </summary>
      /// <param name="rows"></param>
      /// <param name="isLeft"></param>
      public override void addRows (IBlueRowCollection rows, bool isLeft)
      {
         foreach (IBlueRow row in rows)
            myAddRow (row, isLeft);
         if (!isLeft)
            foreach (RowNode node in myLeftOrphanedRows.Values)
               ((JoinTable)myLastFiredPacket.newRows).addRow (node.row, null, myLastFiredPacket);
      }

      /// <summary>
      /// called by JoinOperand
      /// </summary>
      /// <param name="rows"></param>
      /// <param name="isLeft"></param>
      public override void deleteRows (IBlueRowCollection rows, bool isLeft)
      {
         myLeftOrphanedRows = new Hashtable ();
         foreach (IBlueRow row in rows)
            myDeleteRow (row, isLeft);
      }

      private void myAddRow (IBlueRow row, bool isLeft)
      {
         if (isLeft)
         {
            myLeftOperand.addRow (row);
            RowList matchedRows = myRightOperand.getMatchingRows (row[myLeftJoinColOrdinal]);
            if (matchedRows.Count == 0)
               ((JoinTable)myLastFiredPacket.newRows).addRow (row, null, myLastFiredPacket);
            else 
               foreach (RowNode matchedNode in matchedRows)
                    ((JoinTable)myLastFiredPacket.newRows).addRow (row, matchedNode.row, myLastFiredPacket);
         }
         else
         {
            RowList matchedRows = myLeftOperand.getMatchingRows (row[myRightJoinColOrdinal]);
            int rightMatchingRowCount = myRightOperand.getMatchingRows (row[myRightJoinColOrdinal]).Count;
            myRightOperand.addRow (row);
            int newRightMatchingRowCount = myRightOperand.getMatchingRows (row[myRightJoinColOrdinal]).Count;
            if (matchedRows.Count != 0){
               foreach (RowNode matchedNode in matchedRows)
               {
                  //if the left rows were orphaned in this transaction (ie in this callback), they should
                  //just be removed from the orphanedlist and we dont need to worry about the consistency of
                  //the jointable as a new row will replace it
                  if (rightMatchingRowCount == 0 && newRightMatchingRowCount > 0)
                  {
                     if (myLeftOrphanedRows[matchedNode.row["id"]] == null)
                        ((JoinTable)myLastFiredPacket.newRows).deleteRow (true, matchedNode.row, myLastFiredPacket);
                     else
                        myLeftOrphanedRows.Remove (matchedNode.row["id"]);
                  }
                  //they were orphaned by a previous transaction and hence the orphans need to be
                  //removed from the Jointable and the complete row added
                  ((JoinTable)myLastFiredPacket.newRows).addRow (matchedNode.row, row, myLastFiredPacket);

               }
            }
         }
      }

      private void myDeleteRow (IBlueRow row, bool isLeft)
      {
         if (isLeft)
         {
            myLeftOperand.deleteRow (row);
            ((JoinTable)myLastFiredPacket.newRows).deleteRow (true, row, myLastFiredPacket);
         }
         else
         {
            RowList matchedRows = myLeftOperand.getMatchingRows (row[myRightJoinColOrdinal]);
            int rightMatchingRowCount = myRightOperand.getMatchingRows (row[myRightJoinColOrdinal]).Count;
            myRightOperand.deleteRow (row);
            int newRightMatchingRowCount = myRightOperand.getMatchingRows (row[myRightJoinColOrdinal]).Count;
            if (matchedRows.Count != 0)
            {
               if (rightMatchingRowCount > 0 && newRightMatchingRowCount == 0)
               //ie all the matchedRows were joined with 'this' right row (method arg), and hence they need to
               //be handled specially, ie joined with nulls, if there are no right rows with this
               //joincolumn value being inserted. 
                  foreach (RowNode matchedNode in matchedRows)
                  {
                     // BlueEventLog.logAssert (myLeftOrphanedRows[matchedNode.row["id"]] == null);
                     myLeftOrphanedRows[matchedNode.row["id"]] = matchedNode;
                  }
               ((JoinTable)myLastFiredPacket.newRows).deleteRow (false, row, myLastFiredPacket);
            }
         }           
      }
      
      public override void onTableClear (bool isLeft)
      {
         myLastFiredPacket.deletedRows = myLastFiredPacket.newRows;
         myLastFiredPacket.newRows = new JoinTable (mySchema);
         myLastFiredPacket.addedRows = new BlueTable (mySchema);
         if (!isLeft)
            foreach (IBlueRow row in myLeftDataPacket.newRows)
               ((JoinTable)myLastFiredPacket.newRows).addRow (row, null, myLastFiredPacket);
      }
      
   }
}