(p *plan.Source)
| 198 | return NewSource(m.Ctx, p) |
| 199 | } |
| 200 | func (m *JobExecutor) WalkSourceExec(p *plan.Source) (Task, error) { |
| 201 | |
| 202 | if p.Conn == nil { |
| 203 | if p.DataSource == nil { |
| 204 | u.Warnf("no datasource") |
| 205 | return nil, fmt.Errorf("missing data source") |
| 206 | } |
| 207 | source, err := p.DataSource.Open(p.Stmt.SourceName()) |
| 208 | if err != nil { |
| 209 | return nil, err |
| 210 | } |
| 211 | p.Conn = source |
| 212 | } |
| 213 | |
| 214 | e, hasSourceExec := p.Conn.(ExecutorSource) |
| 215 | if hasSourceExec { |
| 216 | return e.WalkExecSource(p) |
| 217 | } |
| 218 | u.Warnf("source %T does not implement datasource.Scanner", p.Conn) |
| 219 | return nil, fmt.Errorf("%T Must Implement Scanner for %q", p.Conn, p.Stmt.String()) |
| 220 | } |
| 221 | func (m *JobExecutor) WalkWhere(p *plan.Where) (Task, error) { |
| 222 | return NewWhere(m.Ctx, p), nil |
| 223 | } |
nothing calls this directly
no test coverage detected