public void initialize(InputInitializerContext context, MapWork work, JobConf jobConf) throws SerDeException {
    this.clear();
    this.context = context;
    this.work = work;
    this.jobConf = jobConf;

    Map<String, SourceInfo> columnMap = new HashMap<String, SourceInfo>();
    // sources represent vertex names
    Set<String> sources = work.getEventSourceTableDescMap().keySet();

    sourcesWaitingForEvents.addAll(sources);

    for (String s : sources) {
      // Set to 0 to start with. This will be decremented for all columns for which events
      // are generated by this source - which is eventually used to determine number of expected
      // events for the source. #colums X #tasks
      numExpectedEventsPerSource.put(s, new MutableInt(0));
      numEventsSeenPerSource.put(s, new MutableInt(0));
      // Virtual relation generated by the reduce sync
      List<TableDesc> tables = work.getEventSourceTableDescMap().get(s);
      // Real column name - on which the operation is being performed
      List<String> columnNames = work.getEventSourceColumnNameMap().get(s);
      // Column type
      List<String> columnTypes = work.getEventSourceColumnTypeMap().get(s);
      // Expression for the operation. e.g. N^2 > 10
      List<ExprNodeDesc> partKeyExprs = work.getEventSourcePartKeyExprMap().get(s);
      // Expression for pruning
      List<ExprNodeDesc> predicates = work.getEventSourcePredicateExprMap().get(s);

      // eventSourceTableDesc, eventSourceColumnName, evenSourcePartKeyExpr move in lock-step.
      // One entry is added to each at the same time

      Iterator<String> cit = columnNames.iterator();
      Iterator<String> typit = columnTypes.iterator();
      Iterator<ExprNodeDesc> pit = partKeyExprs.iterator();
      Iterator<ExprNodeDesc> predit = predicates.iterator();
      // A single source can process multiple columns, and will send an event for each of them.
      for (TableDesc t : tables) {
        numExpectedEventsPerSource.get(s).decrement();
        ++sourceInfoCount;
        String columnName = cit.next();
        String columnType = typit.next();
        ExprNodeDesc partKeyExpr = pit.next();
        ExprNodeDesc predicate = predit.next();
        SourceInfo si = createSourceInfo(t, partKeyExpr, predicate, columnName, columnType, jobConf);
        if (!sourceInfoMap.containsKey(s)) {
          sourceInfoMap.put(s, new ArrayList<SourceInfo>());
        }
        List<SourceInfo> sis = sourceInfoMap.get(s);
        sis.add(si);

        // We could have multiple sources restrict the same column, need to take
        // the union of the values in that case.
        if (columnMap.containsKey(columnName)) {
          // All Sources are initialized up front. Events from different sources will end up getting added to the same list.
          // Pruning is disabled if either source sends in an event which causes pruning to be skipped
          si.values = columnMap.get(columnName).values;
          si.skipPruning = columnMap.get(columnName).skipPruning;
        }
        columnMap.put(columnName, si);
      }
    }
  }
DynamicPartitionPruner takes a list of assigned partitions at runtime (split
generation) and prunes them using events generated during execution of the
dag.
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DynamicPartitionPruner.java