(ctx context.Context, c *mysql.Conn, query string, parsed sqlparser.Statement, analyzedPlan sql.Node, queryExec QueryExecutor, callback func(*sql.Context, *Result) error, formatCodes []int16)
| 338 | var queryLoggingRegex = regexp.MustCompile(`[\r\n\t ]+`) |
| 339 | |
| 340 | func (h *DoltgresHandler) doQuery(ctx context.Context, c *mysql.Conn, query string, parsed sqlparser.Statement, analyzedPlan sql.Node, queryExec QueryExecutor, callback func(*sql.Context, *Result) error, formatCodes []int16) error { |
| 341 | sqlCtx, err := h.sm.NewContextWithQuery(ctx, c, query) |
| 342 | if err != nil { |
| 343 | return err |
| 344 | } |
| 345 | sqlCtx.SetPrivilegeSet(auth.NewPrivilegeSetLayer(sqlCtx), 1) |
| 346 | |
| 347 | start := time.Now() |
| 348 | var queryStrToLog string |
| 349 | if h.encodeLoggedQuery { |
| 350 | queryStrToLog = base64.StdEncoding.EncodeToString([]byte(query)) |
| 351 | } else if logrus.IsLevelEnabled(logrus.DebugLevel) { |
| 352 | // this is expensive, so skip this unless we're logging at DEBUG level |
| 353 | queryStrToLog = string(queryLoggingRegex.ReplaceAll([]byte(query), []byte(" "))) |
| 354 | } |
| 355 | |
| 356 | if queryStrToLog != "" { |
| 357 | sqlCtx.SetLogger(sqlCtx.GetLogger().WithField("query", queryStrToLog)) |
| 358 | } |
| 359 | sqlCtx.GetLogger().Debugf("Starting query") |
| 360 | sqlCtx.GetLogger().Tracef("beginning execution") |
| 361 | |
| 362 | // TODO: it would be nice to put this logic in the engine, not the handler, but we don't want the process to be |
| 363 | // marked done until we're done spooling rows over the wire |
| 364 | lgr := sqlCtx.GetLogger() |
| 365 | sqlCtx, err = sqlCtx.ProcessList.BeginQuery(sqlCtx, query) |
| 366 | if err != nil { |
| 367 | lgr.WithError(err).Warn("error running query; could not open process list context") |
| 368 | return err |
| 369 | } |
| 370 | defer sqlCtx.ProcessList.EndQuery(sqlCtx) |
| 371 | |
| 372 | schema, rowIter, qFlags, err := queryExec(sqlCtx, query, parsed, analyzedPlan) |
| 373 | if err != nil { |
| 374 | if printErrorStackTraces { |
| 375 | fmt.Printf("error running query: %+v\n", err) |
| 376 | } |
| 377 | sqlCtx.GetLogger().WithError(err).Warn("error running query") |
| 378 | return err |
| 379 | } |
| 380 | |
| 381 | // create result before goroutines to avoid |ctx| racing |
| 382 | var r *Result |
| 383 | var processedAtLeastOneBatch bool |
| 384 | |
| 385 | // zero/single return schema use spooling shortcut |
| 386 | if types.IsOkResultSchema(schema) { |
| 387 | r, err = resultForOkIter(sqlCtx, rowIter) |
| 388 | if err != nil { |
| 389 | return err |
| 390 | } |
| 391 | } else if schema == nil { |
| 392 | r, err = resultForEmptyIter(sqlCtx, rowIter) |
| 393 | if err != nil { |
| 394 | return err |
| 395 | } |
| 396 | } else if analyzer.FlagIsSet(qFlags, sql.QFlagMax1Row) { |
| 397 | resultFields, err := schemaToFieldDescriptions(sqlCtx, schema, formatCodes) |
no test coverage detected