Method Run

exec/join.go:54–98

()

}

func (m *JoinKey) Run() error {
	defer m.Ctx.Recover()
	defer close(m.msgOutCh)

	outCh := m.MessageOut()
	inCh := m.MessageIn()
	joinNodes := m.p.Source.Stmt.JoinNodes()

	for {

		select {
		case <-m.SigChan():
			//u.Debugf("got signal quit")
			return nil
		case msg, ok := <-inCh:
			if !ok {
				//u.Debugf("NICE, got msg shutdown")
				return nil
			}

			//u.Infof("In joinkey msg %#v", msg)
		msgTypeSwitch:
			switch mt := msg.(type) {
			case *datasource.SqlDriverMessageMap:
				vals := make([]string, len(joinNodes))
				for i, node := range joinNodes {
					joinVal, ok := vm.Eval(mt, node)
					//u.Debugf("evaluating: ok?%v T:%T result=%v node '%v'", ok, joinVal, joinVal.ToString(), node.String())
					if !ok {
						u.Errorf("could not evaluate: %T %#v   %v", joinVal, joinVal, msg)
						break msgTypeSwitch
					}
					vals[i] = joinVal.ToString()
				}
				//u.Infof("joinkey: %v row:%v", vals, mt)
				key := strings.Join(vals, string(byte(0)))
				mt.SetKeyHashed(key)
				outCh <- mt
			default:
				return fmt.Errorf("To use JoinKey must use SqlDriverMessageMap but got %T", msg)
			}

		}
	}
}

// Scans 2 source tasks for rows, evaluate keys, use for join
//