RE: notes

[Ashish Thusoo <at...@facebook.com>]

Cool write up, it does capture a lot of our discussions...

Moving the discussion to hive-dev@hadoop.apache.org...

Some comments on the size of the chain that is needed...

We do not have to do a closest match. If we specify the notion of an anchor in the rule e.g. TS,RS,$ means that the last two symbols encountered in the chain are TS and RS then we can still do
this as an exact match (we can implicitly have $ be a part of the rule i.e. instead of a generic regular expression we say that the expression is always anchored at the top of the stack).

And we can choose to just pass the matching subpath though I think passing the entire subpath is probably more generic.

Also instead of multiple different processors we could also share the processors by introducing the notion of or in the rule and then letting the rule call the processor with which branch matched. So we could still have a rule specific switch statement rather than a big switch statement in the dispatcher.

I would also add that lets make the rule list to be inside the dispatcher instead of being inside the walker. Clearly that can be shared by many different types of walkers.

One thing about the context though, how would you pass that around from one processor to another. Maybe it makes sense to register a context with the dispatcher that gets passed to all the processors.

Something on the following lines...

Define a class called ProcessContext that can be extended by GenMapRedTaskProcContext 

i.e

class ProcessContext {
}

class GenMapRedTaskProcContext extends ProcessContext {

... all the context stuff that you had ...
}

class TriggerDispatcher {
 
  private List<Trigger> triggers;

  void add(Trigger);
}

class Trigger {
  
   private Rule r;

   private Action a;
}

interface Action {

   process(ProcessContext ctx, RuleContext rctx, PathStack);
}


while RuleContext passes some rule specific information like which rule or which component of the rule fired

Then you could encode the whole things as

tw = new TreeWalker();
td = new TriggerDispatcher();
td.add(new Trigger(new Rule()..., new RedMapTaskAction..)
tw.setDispatcher(td);
tw.walk();

Something on those lines..

Ashish

________
________________________________
From: Namit Jain
Sent: Wednesday, November 19, 2008 7:11 PM
To: Ashish Thusoo; Raghu Murthy
Subject: notes

Was writing whatever was coming to mind: not coherent at all.

Basically, glance over it - the last part was what I think will work - not most general, but should suffice.


Let us create 1 more operator type: MapSink.
All parents of MapJoin are of type MapSink (similar to ReduceSink).

Currently, the client specifies whether the join is Mapside Join (based on a hint), but it can be later on
determined by cost.

For a give Map-Join, one parent is special i.e although it is a map-sink, it is invoked by the map-reduce
framework. In this note, it is referred to as DummyMapSink

For eg: consider the following join:

select /*+ MAPJOIN(src1) */ ... from src1 JOIN src2 on (...) where ...;

In the above example, src1 is read in memory by all mappers and src2 is invoked by the map-reduce framework.
So, the MapSink operator corresponding to src2 (child of Table Scan of src2) is a dummy MapSink operator.
Note that only one of the parents of the map-join operator can be a dummy MapSink, all others are MapSink

The behavior at the following transitions is as follows:

TS:  TableScan
MS:  MapSink
MD:  DummyMapSink
RS:  ReduceSink


1.  TS -> RS: No Change, mark reducer for the plan if not assigned already
2.  TS -> RS: if reducer assigned already, split if reducer is different
3.  RS -> RS: split plan
4.  RS -> MS: split plan
5.  RS -> MD: no change
6.  MS -> RS: split plan
7.  MD -> RS: split plan
9.  MS -> MS: no change
10. MS -> MD: split plan
11. MD -> MS: split plan
12. MD -> MD: no change
13. TS -> MS: no change
14. TS -> MD: no change

As is obvious from above, it is a messy state transition, which can be coded easily but will become difficult to
maintain and enhance. The state transitions will be coded in the code which can make them painful to change.

Therefore, it might be a good idea to have a different processor per rule.


The pseudo code can be as follows:


public class GenMapRedWalker extends DefaultOpGraphWalker {
  private List<RuleDispatcher>   dispList;
  private Stack                  opNameStack;
}

Walker maintains a list of rules and dispatchers.


The current processor has the following state:


public class GenMapRedTaskProcessor extends OperatorProcessor {

  private HashMap<Operator<? extends Serializable>, Task<? extends Serializable>> opTaskMap =
    new HashMap<Operator<? extends Serializable>, Task<? extends Serializable>>();
  private List<Operator<? extends Serializable>> seenOps = new ArrayList<Operator<? extends Serializable>>();

  private ParseContext                          parseCtx;
  private Task<? extends Serializable>          mvTask;
  private List<Task<? extends Serializable>>    rootTasks;
  private String scratchDir;
  private int randomid;
  private int pathid;

  private Map<Operator<? extends Serializable>, GenMapRedCtx> mapCurrCtx;
  private Task<? extends Serializable>         currTask;
  private Operator<? extends Serializable>     currTopOp;
  private String                               currAliasId;

  /**
   * GenMapRedCtx is used to keep track of the current state.
   */
  private static class GenMapRedCtx {
    Task<? extends Serializable>         currTask;
    Operator<? extends Serializable>     currTopOp;
    String                               currAliasId;
  }
}



Even if we have 12/14 different processors, they need to share the above state - that can be implemented easily.
All the processors will be subclasses of GenMapRedTaskProcessor(). The new structure might be like:


public class GenMapRedTaskProcContext {
  private HashMap<Operator<? extends Serializable>, Task<? extends Serializable>> opTaskMap; =
  private List<Operator<? extends Serializable>> seenOps;

  private ParseContext                          parseCtx;
  private Task<? extends Serializable>          mvTask;
  private List<Task<? extends Serializable>>    rootTasks;
  private String scratchDir;
  private int randomid;
  private int pathid;

  private Map<Operator<? extends Serializable>, GenMapRedCtx> mapCurrCtx;
  private Task<? extends Serializable>         currTask;
  private Operator<? extends Serializable>     currTopOp;
  private String                               currAliasId;

  /**
   * GenMapRedCtx is used to keep track of the current state.
   */
  private static class GenMapRedCtx {
    Task<? extends Serializable>         currTask;
    Operator<? extends Serializable>     currTopOp;
    String                               currAliasId;
  }
}



public class GenMapRedTaskProcessor extends OperatorProcessor {
  private GenMapRedTaskProcContext ctx;

  process()...
}



One of the problems is that we need to find the closest match. For eg, the stack might be:

TS ..  RS .. MS .. RS

When the last RS is encountered, we need to fire for the MS -> RS rule, not for TS -> RS or RS -> RS.
So, we can't go over the rules one by one and fire the first match. We need to find the closest match.

Is that true ? Does the destination determine everything ? Can we have some minimal state and then only
depend on the last operator encountered.
I guess what I am trying to get is: is the entire chain important ? There are some special operators :
RS, MS, MSD, FS etc. Can we simplify the model by assuming that we keep track of the last important
operator in the path (maybe as state in plan or as plan->state map in the processor) and then take
appropriate action based on last operator and last important operator in the plan.

The problem is that of extensiblity. I am not sure though, if we code it nicely it might be still OK.


Or we can make the simplifying assumption that only 2 operators are important ? The process takes 2 arguments, the
two operators: last important and the current operator.

Processor() will have a API which will be invoked by the dispatcher: isOperatorNeededForStateTransition().
I think this should work - I will try to code ip up soon.