removing transaction committing node (#2708)

jycor · web-flow · commit baa759c6d4ed · 2024-10-21T14:53:15.000-07:00
diff --git a/engine.go b/engine.go
@@ -446,7 +446,8 @@ func (e *Engine) QueryWithBindings(ctx *sql.Context, query string, parsed sqlpar
 
 		return nil, nil, nil, err
 	}
-	iter = rowexec.AddExpressionCloser(analyzed, iter)
+
+	iter = finalizeIters(analyzed, qFlags, iter)
 
 	return analyzed.Schema(), iter, qFlags, nil
 }
@@ -480,7 +481,8 @@ func (e *Engine) PrepQueryPlanForExecution(ctx *sql.Context, _ string, plan sql.
 
 		return nil, nil, nil, err
 	}
-	iter = rowexec.AddExpressionCloser(plan, iter)
+
+	iter = finalizeIters(plan, nil, iter)
 
 	return plan.Schema(), iter, nil, nil
 }
@@ -722,31 +724,6 @@ func (e *Engine) CloseSession(connID uint32) {
 	e.PreparedDataCache.DeleteSessionData(connID)
 }
 
-// Count number of BindVars in given tree
-func countBindVars(node sql.Node) int {
-	var bindVars map[string]bool
-	bindCntFunc := func(e sql.Expression) bool {
-		if bv, ok := e.(*expression.BindVar); ok {
-			if bindVars == nil {
-				bindVars = make(map[string]bool)
-			}
-			bindVars[bv.Name] = true
-		}
-		return true
-	}
-	transform.InspectExpressions(node, bindCntFunc)
-
-	// InsertInto.Source not a child of InsertInto, so also need to traverse those
-	transform.Inspect(node, func(n sql.Node) bool {
-		if in, ok := n.(*plan.InsertInto); ok {
-			transform.InspectExpressions(in.Source, bindCntFunc)
-			return false
-		}
-		return true
-	})
-	return len(bindVars)
-}
-
 func (e *Engine) beginTransaction(ctx *sql.Context) error {
 	beginNewTransaction := ctx.GetTransaction() == nil || plan.ReadCommitted(ctx)
 	if beginNewTransaction {
@@ -852,7 +829,8 @@ func (e *Engine) executeEvent(ctx *sql.Context, dbName, createEventStatement, us
 		}
 		return err
 	}
-	iter = rowexec.AddExpressionCloser(definitionNode, iter)
+
+	iter = finalizeIters(definitionNode, nil, iter)
 
 	// Drain the iterate to execute the event body/definition
 	// NOTE: No row data is returned for an event; we just need to execute the statements
@@ -885,3 +863,10 @@ func findCreateEventNode(planTree sql.Node) (*plan.CreateEvent, error) {
 
 	return createEventNode, nil
 }
+
+// finalizeIters applies the final transformations on sql.RowIter before execution.
+func finalizeIters(analyzed sql.Node, qFlags *sql.QueryFlags, iter sql.RowIter) sql.RowIter {
+	iter = rowexec.AddTransactionCommittingIter(iter, qFlags)
+	iter = rowexec.AddExpressionCloser(analyzed, iter)
+	return iter
+}
diff --git a/enginetest/memory_engine_test.go b/enginetest/memory_engine_test.go
@@ -30,7 +30,6 @@ import (
 	"github.com/dolthub/go-mysql-server/memory"
 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/dolthub/go-mysql-server/sql/expression"
-	"github.com/dolthub/go-mysql-server/sql/plan"
 	"github.com/dolthub/go-mysql-server/sql/types"
 	_ "github.com/dolthub/go-mysql-server/sql/variables"
 )
@@ -196,13 +195,6 @@ func TestSingleQueryPrepared(t *testing.T) {
 	enginetest.TestScriptWithEnginePrepared(t, engine, harness, test)
 }
 
-func newUpdateResult(matched, updated int) types.OkResult {
-	return types.OkResult{
-		RowsAffected: uint64(updated),
-		Info:         plan.UpdateInfo{Matched: matched, Updated: updated},
-	}
-}
-
 // Convenience test for debugging a single query. Unskip and set to the desired query.
 func TestSingleScript(t *testing.T) {
 	t.Skip()
@@ -1065,14 +1057,6 @@ func findTable(dbs []sql.Database, tableName string) (sql.Database, sql.Table) {
 	return nil, nil
 }
 
-func mergeSetupScripts(scripts ...setup.SetupScript) []string {
-	var all []string
-	for _, s := range scripts {
-		all = append(all, s...)
-	}
-	return all
-}
-
 func TestSQLLogicTests(t *testing.T) {
 	enginetest.TestSQLLogicTests(t, enginetest.NewMemoryHarness("default", 1, testNumPartitions, true, mergableIndexDriver))
 }
diff --git a/server/handler.go b/server/handler.go
@@ -519,28 +519,26 @@ func resultForEmptyIter(ctx *sql.Context, iter sql.RowIter, resultFields []*quer
 func GetDeferredProjections(iter sql.RowIter) (sql.RowIter, []sql.Expression) {
 	switch i := iter.(type) {
 	case *rowexec.ExprCloserIter:
-		_, projs := GetDeferredProjections(i.GetIter())
-		return i, projs
+		if newChild, projs := GetDeferredProjections(i.GetIter()); projs != nil {
+			return i.WithChildIter(newChild), projs
+		}
 	case *plan.TrackedRowIter:
-		_, projs := GetDeferredProjections(i.GetIter())
-		return i, projs
+		if newChild, projs := GetDeferredProjections(i.GetIter()); projs != nil {
+			return i.WithChildIter(newChild), projs
+		}
 	case *rowexec.TransactionCommittingIter:
-		newChild, projs := GetDeferredProjections(i.GetIter())
-		if projs != nil {
-			i.WithChildIter(newChild)
+		if newChild, projs := GetDeferredProjections(i.GetIter()); projs != nil {
+			return i.WithChildIter(newChild), projs
 		}
-		return i, projs
 	case *iters.LimitIter:
-		newChild, projs := GetDeferredProjections(i.ChildIter)
-		if projs != nil {
+		if newChild, projs := GetDeferredProjections(i.ChildIter); projs != nil {
 			i.ChildIter = newChild
+			return i, projs
 		}
-		return i, projs
 	case *rowexec.ProjectIter:
 		if i.CanDefer() {
 			return i.GetChildIter(), i.GetProjections()
 		}
-		return i, nil
 	}
 	return iter, nil
 }
diff --git a/sql/analyzer/analyzer.go b/sql/analyzer/analyzer.go
@@ -404,7 +404,6 @@ func NewProcRuleSelector(sel RuleSelector) RuleSelector {
 			unnestInSubqueriesId,
 
 			// once after default rules should only be run once
-			AutocommitId,
 			TrackProcessId,
 			parallelizeId:
 			return false
diff --git a/sql/analyzer/autocommit.go b/sql/analyzer/autocommit.go
diff --git a/sql/analyzer/parallelize.go b/sql/analyzer/parallelize.go
@@ -54,10 +54,6 @@ func shouldParallelize(node sql.Node, scope *plan.Scope) bool {
 		return false
 	}
 
-	if tc, ok := node.(*plan.TransactionCommittingNode); ok {
-		return shouldParallelize(tc.Child(), scope)
-	}
-
 	// Do not try to parallelize DDL or descriptive operations
 	return !plan.IsNoRowNode(node)
 }
diff --git a/sql/analyzer/resolve_subqueries.go b/sql/analyzer/resolve_subqueries.go
@@ -296,8 +296,6 @@ func StripPassthroughNodes(n sql.Node) sql.Node {
 		switch tn := n.(type) {
 		case *plan.QueryProcess:
 			n = tn.Child()
-		case *plan.TransactionCommittingNode:
-			n = tn.Child()
 		default:
 			nodeIsPassthrough = false
 		}
diff --git a/sql/analyzer/rule_ids.go b/sql/analyzer/rule_ids.go
@@ -86,8 +86,7 @@ const (
 
 	// after all
 	cacheSubqueryAliasesInJoinsId    // cacheSubqueryAliasesInJoins
-	backtickDefaulColumnValueNamesId // backtickDefaultColumnValueNames
-	AutocommitId                     // addAutocommit
+	BacktickDefaulColumnValueNamesId // backtickDefaultColumnValueNames
 	TrackProcessId                   // trackProcess
 	parallelizeId                    // parallelize
 )
diff --git a/sql/analyzer/ruleid_string.go b/sql/analyzer/ruleid_string.go
diff --git a/sql/analyzer/rules.go b/sql/analyzer/rules.go
@@ -25,8 +25,7 @@ func init() {
 		{wrapWithRollbackId, wrapWithRollback},
 		{inlineSubqueryAliasRefsId, inlineSubqueryAliasRefs},
 		{cacheSubqueryAliasesInJoinsId, cacheSubqueryAliasesInJoins},
-		{backtickDefaulColumnValueNamesId, backtickDefaultColumnValueNames},
-		{AutocommitId, addAutocommit},
+		{BacktickDefaulColumnValueNamesId, backtickDefaultColumnValueNames},
 		{TrackProcessId, trackProcess},
 		{parallelizeId, parallelize},
 	}
diff --git a/sql/plan/process.go b/sql/plan/process.go
@@ -402,6 +402,12 @@ func (i *TrackedRowIter) updateSessionVars(ctx *sql.Context) {
 	}
 }
 
+func (i *TrackedRowIter) WithChildIter(childIter sql.RowIter) sql.RowIter {
+	ni := *i
+	ni.iter = childIter
+	return &ni
+}
+
 type trackedPartitionIndexKeyValueIter struct {
 	sql.PartitionIndexKeyValueIter
 	OnPartitionDone  NamedNotifyFunc
diff --git a/sql/plan/transaction_committing_iter.go b/sql/plan/transaction_committing_iter.go
@@ -15,84 +15,9 @@
 package plan
 
 import (
-	"fmt"
-	"os"
-
 	"github.com/dolthub/go-mysql-server/sql"
 )
 
-const (
-	fakeReadCommittedEnvVar = "READ_COMMITTED_HACK"
-)
-
-var fakeReadCommitted bool
-
-func init() {
-	_, ok := os.LookupEnv(fakeReadCommittedEnvVar)
-	if ok {
-		fakeReadCommitted = true
-	}
-}
-
-// TransactionCommittingNode implements autocommit logic. It wraps relevant queries and ensures the database commits
-// the transaction.
-type TransactionCommittingNode struct {
-	UnaryNode
-}
-
-var _ sql.Node = (*TransactionCommittingNode)(nil)
-var _ sql.CollationCoercible = (*TransactionCommittingNode)(nil)
-
-// NewTransactionCommittingNode returns a TransactionCommittingNode.
-func NewTransactionCommittingNode(child sql.Node) *TransactionCommittingNode {
-	return &TransactionCommittingNode{UnaryNode: UnaryNode{Child: child}}
-}
-
-// String implements the sql.Node interface.
-func (t *TransactionCommittingNode) String() string {
-	return t.Child().String()
-}
-
-// DebugString implements the sql.DebugStringer interface.
-func (t *TransactionCommittingNode) DebugString() string {
-	return sql.DebugString(t.Child())
-}
-
-// Describe implements the sql.Describable interface.
-func (t *TransactionCommittingNode) Describe(options sql.DescribeOptions) string {
-	return sql.Describe(t.Child(), options)
-}
-
-func (t *TransactionCommittingNode) IsReadOnly() bool {
-	return t.Child().IsReadOnly()
-}
-
-// WithChildren implements the sql.Node interface.
-func (t *TransactionCommittingNode) WithChildren(children ...sql.Node) (sql.Node, error) {
-	if len(children) != 1 {
-		return nil, fmt.Errorf("ds")
-	}
-
-	t2 := *t
-	t2.UnaryNode = UnaryNode{Child: children[0]}
-	return &t2, nil
-}
-
-// CheckPrivileges implements the sql.Node interface.
-func (t *TransactionCommittingNode) CheckPrivileges(ctx *sql.Context, opChecker sql.PrivilegedOperationChecker) bool {
-	return t.Child().CheckPrivileges(ctx, opChecker)
-}
-
-// CollationCoercibility implements the interface sql.CollationCoercible.
-func (*TransactionCommittingNode) CollationCoercibility(ctx *sql.Context) (collation sql.CollationID, coercibility byte) {
-	return sql.Collation_binary, 7
-}
-
-// Child implements the sql.UnaryNode interface.
-func (t *TransactionCommittingNode) Child() sql.Node {
-	return t.UnaryNode.Child
-}
-
 // IsSessionAutocommit returns true if the current session is using implicit transaction management
 // through autocommit.
 func IsSessionAutocommit(ctx *sql.Context) (bool, error) {
@@ -108,10 +33,6 @@ func IsSessionAutocommit(ctx *sql.Context) (bool, error) {
 }
 
 func ReadCommitted(ctx *sql.Context) bool {
-	if !fakeReadCommitted {
-		return false
-	}
-
 	val, err := ctx.GetSessionVariable(ctx, "transaction_isolation")
 	if err != nil {
 		return false
diff --git a/sql/rowexec/expr_closer.go b/sql/rowexec/expr_closer.go
@@ -68,3 +68,9 @@ func (eci *ExprCloserIter) Close(ctx *sql.Context) error {
 func (eci *ExprCloserIter) GetIter() sql.RowIter {
 	return eci.iter
 }
+
+func (eci *ExprCloserIter) WithChildIter(childIter sql.RowIter) sql.RowIter {
+	neci := *eci
+	neci.iter = childIter
+	return &neci
+}
diff --git a/sql/rowexec/node_builder.gen.go b/sql/rowexec/node_builder.gen.go
@@ -54,8 +54,6 @@ func (b *BaseBuilder) buildNodeExecNoAnalyze(ctx *sql.Context, n sql.Node, row s
 		return b.buildCreateRole(ctx, n, row)
 	case *plan.Loop:
 		return b.buildLoop(ctx, n, row)
-	case *plan.TransactionCommittingNode:
-		return b.buildTransactionCommittingNode(ctx, n, row)
 	case *plan.DropColumn:
 		return b.buildDropColumn(ctx, n, row)
 	case *plan.AnalyzeTable:
diff --git a/sql/rowexec/transaction.go b/sql/rowexec/transaction.go
@@ -300,11 +300,3 @@ func (b *BaseBuilder) buildExecuteQuery(ctx *sql.Context, n *plan.ExecuteQuery,
 func (b *BaseBuilder) buildUse(ctx *sql.Context, n *plan.Use, row sql.Row) (sql.RowIter, error) {
 	return n.RowIter(ctx, row)
 }
-
-func (b *BaseBuilder) buildTransactionCommittingNode(ctx *sql.Context, n *plan.TransactionCommittingNode, row sql.Row) (sql.RowIter, error) {
-	iter, err := b.Build(ctx, n.Child(), row)
-	if err != nil {
-		return nil, err
-	}
-	return &TransactionCommittingIter{childIter: iter}, nil
-}
diff --git a/sql/rowexec/transaction_iters.go b/sql/rowexec/transaction_iters.go

Original file line number	Diff line number	Diff line change
`@@ -54,10 +54,6 @@ func shouldParallelize(node sql.Node, scope *plan.Scope) bool {`
`54`	`54`	`return false`
`55`	`55`	`}`
`56`	`56`
`57`		`- if tc, ok := node.(*plan.TransactionCommittingNode); ok {`
`58`		`- return shouldParallelize(tc.Child(), scope)`
`59`		`- }`
`60`		`-`
`61`	`57`	`// Do not try to parallelize DDL or descriptive operations`
`62`	`58`	`return !plan.IsNoRowNode(node)`
`63`	`59`	`}`
Original file line number	Diff line number	Diff line change
`@@ -296,8 +296,6 @@ func StripPassthroughNodes(n sql.Node) sql.Node {`
`296`	`296`	`switch tn := n.(type) {`
`297`	`297`	`case *plan.QueryProcess:`
`298`	`298`	`n = tn.Child()`
`299`		`- case *plan.TransactionCommittingNode:`
`300`		`- n = tn.Child()`
`301`	`299`	`default:`
`302`	`300`	`nodeIsPassthrough = false`
`303`	`301`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,8 +86,7 @@ const (`
`86`	`86`
`87`	`87`	`// after all`
`88`	`88`	`cacheSubqueryAliasesInJoinsId // cacheSubqueryAliasesInJoins`
`89`		`- backtickDefaulColumnValueNamesId // backtickDefaultColumnValueNames`
`90`		`- AutocommitId // addAutocommit`
	`89`	`+ BacktickDefaulColumnValueNamesId // backtickDefaultColumnValueNames`
`91`	`90`	`TrackProcessId // trackProcess`
`92`	`91`	`parallelizeId // parallelize`
`93`	`92`	`)`