move applyEventScheduler logic and eventscheduler to builder

Author: James Cor

engine.go

@@ -209,7 +209,7 @@ func (e *Engine) AnalyzeQuery(
 	ctx *sql.Context,
 	query string,
 ) (sql.Node, error) {
-	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser)
+	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser, e.EventScheduler)
 	parsed, _, _, qFlags, err := binder.Parse(query, nil, false)
 	if err != nil {
 		return nil, err
@@ -237,7 +237,7 @@ func (e *Engine) PrepareParsedQuery(
 	statementKey, query string,
 	stmt sqlparser.Statement,
 ) (sql.Node, error) {
-	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser)
+	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser, e.EventScheduler)
 	node, _, err := binder.BindOnly(stmt, query, nil)
 
 	if err != nil {
@@ -495,7 +495,7 @@ func (e *Engine) BoundQueryPlan(ctx *sql.Context, query string, parsed sqlparser
 
 	query = sql.RemoveSpaceAndDelimiter(query, ';')
 
-	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser)
+	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser, e.EventScheduler)
 	binder.SetBindings(bindings)
 
 	// Begin a transaction if necessary (no-op if one is in flight)
@@ -549,7 +549,7 @@ func (e *Engine) preparedStatement(ctx *sql.Context, query string, parsed sqlpar
 		preparedAst, preparedDataFound = e.PreparedDataCache.GetCachedStmt(ctx.Session.ID(), query)
 	}
 
-	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser)
+	binder := planbuilder.New(ctx, e.Analyzer.Catalog, e.Parser, e.EventScheduler)
 	if preparedDataFound {
 		parsed = preparedAst
 		binder.SetBindings(bindings)
@@ -782,6 +782,10 @@ func (e *Engine) EngineAnalyzer() *analyzer.Analyzer {
 	return e.Analyzer
 }
 
+func (e *Engine) EngineEventScheduler() sql.EventScheduler {
+	return e.EventScheduler
+}
+
 // InitializeEventScheduler initializes the EventScheduler for the engine with the given sql.Context
 // getter function, |ctxGetterFunc, the EventScheduler |status|, and the |period| for the event scheduler
 // to check for events to execute. If |period| is less than 1, then it is ignored and the default period
@@ -792,8 +796,6 @@ func (e *Engine) InitializeEventScheduler(ctxGetterFunc func() (*sql.Context, fu
 	if err != nil {
 		return err
 	}
-
-	e.Analyzer.EventScheduler = e.EventScheduler
 	return nil
 }
 

enginetest/engine_only_test.go

@@ -442,7 +442,7 @@ func TestAnalyzer_Exp(t *testing.T) {
 			require.NoError(t, err)
 
 			ctx := enginetest.NewContext(harness)
-			b := planbuilder.New(ctx, e.EngineAnalyzer().Catalog, sql.NewMysqlParser())
+			b := planbuilder.New(ctx, e.EngineAnalyzer().Catalog, sql.NewMysqlParser(), e.EngineEventScheduler())
 			parsed, _, _, _, err := b.Parse(tt.query, nil, false)
 			require.NoError(t, err)
 

enginetest/enginetests.go

@@ -584,7 +584,7 @@ func TestQueryPlanWithName(t *testing.T, name string, harness Harness, e QueryEn
 	t.Run(name, func(t *testing.T) {
 		ctx := NewContext(harness)
 
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, query)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), query)
 		require.NoError(t, err)
 
 		node, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
@@ -611,7 +611,7 @@ func TestQueryPlanWithName(t *testing.T, name string, harness Harness, e QueryEn
 func TestQueryPlanWithEngine(t *testing.T, harness Harness, e QueryEngine, tt queries.QueryPlanTest, verbose bool) {
 	t.Run(tt.Query, func(t *testing.T) {
 		ctx := NewContext(harness)
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, tt.Query)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), tt.Query)
 		require.NoError(t, err)
 
 		node, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
@@ -1362,7 +1362,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t5 ORDER BY 1", []sql.Row{{int64(1), int64(1)}, {int64(2), int64(2)}}, nil, nil, nil)
 
 		deleteStr := "DELETE FROM t5"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1391,7 +1391,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		RunQueryWithContext(t, e, harness, ctx, "INSERT INTO t6parent VALUES (1,1), (2,2)")
 		RunQueryWithContext(t, e, harness, ctx, "INSERT INTO t6child VALUES (1,1), (2,2)")
 
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, "DELETE FROM t6parent")
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), "DELETE FROM t6parent")
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1419,7 +1419,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t7i ORDER BY 1", []sql.Row{{int64(3), int64(3)}}, nil, nil, nil)
 
 		deleteStr := "DELETE FROM t7"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1447,7 +1447,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t8 ORDER BY 1", []sql.Row{{int64(1), int64(4)}, {int64(2), int64(5)}}, nil, nil, nil)
 
 		deleteStr := "DELETE FROM t8"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1476,7 +1476,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t9 ORDER BY 1", []sql.Row{{int64(7), int64(7)}, {int64(8), int64(8)}}, nil, nil, nil)
 
 		deleteStr := "DELETE FROM t9 WHERE pk > 0"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1503,7 +1503,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t10 ORDER BY 1", []sql.Row{{int64(8), int64(8)}, {int64(9), int64(9)}}, nil, nil, nil)
 
 		deleteStr := "DELETE FROM t10 LIMIT 1000"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1530,7 +1530,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t11 ORDER BY 1", []sql.Row{{int64(1), int64(1)}, {int64(9), int64(9)}}, nil, nil, nil)
 
 		deleteStr := "DELETE FROM t11 ORDER BY 1"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)
@@ -1561,7 +1561,7 @@ func TestTruncate(t *testing.T, harness Harness) {
 		TestQueryWithContext(t, ctx, e, harness, "SELECT * FROM t12b ORDER BY 1", []sql.Row{{int64(1), int64(1)}, {int64(2), int64(2)}}, nil, nil, nil)
 
 		deleteStr := "DELETE t12a, t12b FROM t12a INNER JOIN t12b WHERE t12a.pk=t12b.pk"
-		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, deleteStr)
+		parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), deleteStr)
 		require.NoError(t, err)
 		analyzed, err := e.EngineAnalyzer().Analyze(ctx, parsed, nil, qFlags)
 		require.NoError(t, err)

enginetest/evaluation.go

@@ -526,7 +526,7 @@ func injectBindVarsAndPrepare(
 		}
 	}
 
-	b := planbuilder.New(ctx, e.EngineAnalyzer().Catalog, sql.NewMysqlParser())
+	b := planbuilder.New(ctx, e.EngineAnalyzer().Catalog, sql.NewMysqlParser(), e.EngineEventScheduler())
 	b.SetParserOptions(sql.LoadSqlMode(ctx).ParserOptions())
 	resPlan, _, err := b.BindOnly(parsed, q, nil)
 	if err != nil {

enginetest/join_planning_tests.go

@@ -1777,7 +1777,7 @@ func evalJoinTypeTest(t *testing.T, harness Harness, e QueryEngine, query string
 }
 
 func analyzeQuery(ctx *sql.Context, e QueryEngine, query string) (sql.Node, error) {
-	parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, query)
+	parsed, qFlags, err := planbuilder.Parse(ctx, e.EngineAnalyzer().Catalog, e.EngineEventScheduler(), query)
 	if err != nil {
 		return nil, err
 	}

enginetest/mysqlshim/table.go

@@ -381,7 +381,7 @@ func (t Table) getCreateTable() (*plan.CreateTable, error) {
 		return nil, sql.ErrTableNotFound.New(t.name)
 	}
 	// TODO add catalog
-	createTableNode, _, err := planbuilder.Parse(sql.NewEmptyContext(), sql.MapCatalog{Tables: map[string]sql.Table{t.name: t}}, rows[0][1].(string))
+	createTableNode, _, err := planbuilder.Parse(sql.NewEmptyContext(), sql.MapCatalog{Tables: map[string]sql.Table{t.name: t}}, nil, rows[0][1].(string))
 	if err != nil {
 		return nil, err
 	}

enginetest/query_engine.go

@@ -28,6 +28,7 @@ type QueryEngine interface {
 	Query(ctx *sql.Context, query string) (sql.Schema, sql.RowIter, *sql.QueryFlags, error)
 	// TODO: get rid of this, should not be exposed to engine tests
 	EngineAnalyzer() *analyzer.Analyzer
+	EngineEventScheduler() sql.EventScheduler
 	// TODO: get rid of this, should not be exposed to engine tests
 	EnginePreparedDataCache() *sqle.PreparedDataCache
 	QueryWithBindings(ctx *sql.Context, query string, parsed sqlparser.Statement, bindings map[string]sqlparser.Expr, qFlags *sql.QueryFlags) (sql.Schema, sql.RowIter, *sql.QueryFlags, error)

enginetest/server_engine.go

@@ -150,6 +150,10 @@ func (s *ServerQueryEngine) EngineAnalyzer() *analyzer.Analyzer {
 	return s.engine.Analyzer
 }
 
+func (s *ServerQueryEngine) EngineEventScheduler() sql.EventScheduler {
+	return s.engine.EventScheduler
+}
+
 func (s *ServerQueryEngine) EnginePreparedDataCache() *sqle.PreparedDataCache {
 	return s.engine.PreparedDataCache
 }
@@ -632,7 +636,7 @@ func convertGoSqlType(columnType *gosql.ColumnType) (sql.Type, error) {
 // It cannot sort user-defined binding variables (e.g. :var, :foo)
 func prepareBindingArgs(ctx *sql.Context, bindings map[string]sqlparser.Expr) ([]any, error) {
 	// NOTE: using binder with nil catalog and parser since we're only using it to convert SQLVal.
-	binder := planbuilder.New(ctx, nil, nil)
+	binder := planbuilder.New(ctx, nil, nil, nil)
 	numBindVars := len(bindings)
 	args := make([]any, numBindVars)
 	for i := 0; i < numBindVars; i++ {

sql/analyzer/analyzer.go

@@ -286,9 +286,6 @@ type Analyzer struct {
 	Coster memo.Coster
 	// ExecBuilder converts a sql.Node tree into an executable iterator.
 	ExecBuilder sql.NodeExecBuilder
-	// EventScheduler is used to communiate with the event scheduler
-	// for any EVENT related statements. It can be nil if EventScheduler is not defined.
-	EventScheduler sql.EventScheduler
 }
 
 // NewDefault creates a default Analyzer instance with all default Rules and configuration.