(wr *pipe.Writer, sql *parser.SQLFlowStmt, db *database.DB, modelDir string, session *pb.Session)
| 159 | } |
| 160 | |
| 161 | func runSingleSQLFlowStatement(wr *pipe.Writer, sql *parser.SQLFlowStmt, db *database.DB, modelDir string, session *pb.Session) (e error) { |
| 162 | defer func(startTime int64) { |
| 163 | // NOTE(tony): EndOfExecution indicates a successful run, |
| 164 | // so we only writes it when e != nil |
| 165 | if e != nil { |
| 166 | wr.Write(EndOfExecution{ |
| 167 | StartTime: startTime, |
| 168 | EndTime: time.Now().UnixNano(), |
| 169 | Statement: sql.Original, |
| 170 | }) |
| 171 | } |
| 172 | }(time.Now().UnixNano()) |
| 173 | |
| 174 | // use system default tmp dir |
| 175 | cwd, err := ioutil.TempDir("/tmp", "sqlflow_models") |
| 176 | if err != nil { |
| 177 | return err |
| 178 | } |
| 179 | defer func(cwd string) { |
| 180 | if err := os.RemoveAll(cwd); err != nil { |
| 181 | e = fmt.Errorf("encounter %v when dealwith error: %s", e, err) |
| 182 | } |
| 183 | }(cwd) |
| 184 | var r ir.SQLFlowStmt |
| 185 | |
| 186 | generateTrainStmtFromModel := executor.New(session.Submitter).GetTrainStmtFromModel() |
| 187 | |
| 188 | if sql.IsExtendedSyntax() { |
| 189 | if sql.Train { |
| 190 | loadPreTrainModel := generateTrainStmtFromModel |
| 191 | r, err = ir.GenerateTrainStmtWithInferredColumns(sql.SQLFlowSelectStmt, session.DbConnStr, modelDir, cwd, loadPreTrainModel, true) |
| 192 | } else if sql.ShowTrain { |
| 193 | r, err = ir.GenerateShowTrainStmt(sql.SQLFlowSelectStmt) |
| 194 | } else if sql.Explain { |
| 195 | r, err = ir.GenerateExplainStmt(sql.SQLFlowSelectStmt, session.DbConnStr, modelDir, cwd, generateTrainStmtFromModel) |
| 196 | } else if sql.Predict { |
| 197 | r, err = ir.GeneratePredictStmt(sql.SQLFlowSelectStmt, session.DbConnStr, modelDir, cwd, generateTrainStmtFromModel) |
| 198 | } else if sql.Evaluate { |
| 199 | r, err = ir.GenerateEvaluateStmt(sql.SQLFlowSelectStmt, session.DbConnStr, modelDir, cwd, generateTrainStmtFromModel) |
| 200 | } else if sql.Optimize { |
| 201 | r, err = ir.GenerateOptimizeStmt(sql.SQLFlowSelectStmt) |
| 202 | } else if sql.Run { |
| 203 | r, err = ir.GenerateRunStmt(sql.SQLFlowSelectStmt) |
| 204 | } |
| 205 | |
| 206 | } else { |
| 207 | standardSQL := ir.NormalStmt(sql.Original) |
| 208 | r = &standardSQL |
| 209 | } |
| 210 | if err != nil { |
| 211 | return err |
| 212 | } |
| 213 | if err = initializeAndCheckAttributes(r); err != nil { |
| 214 | return err |
| 215 | } |
| 216 | r.SetOriginalSQL(sql.Original) |
| 217 | // TODO(typhoonzero): can run feature.LogDerivationResult(wr, trainStmt) here to send |
| 218 | // feature derivation logs to client, yet we disable it for now so that it's less annoying. |
no test coverage detected