DDL allowed outside of transaction in vttablet (#16661)

Signed-off-by: Harshit Gangal <harshit@planetscale.com>

go/test/endtoend/vtgate/misc_test.go

@@ -798,6 +798,24 @@ func TestRowCountExceed(t *testing.T) {
  utils.AssertContainsError(t, conn, "select id1 from t1 where id1 < 1000", `Row count exceeded 100`)
 }
 
+func TestDDLTargeted(t *testing.T) {
+ ctx := context.Background()
+ conn, err := mysql.Connect(ctx, &vtParams)
+ require.NoError(t, err)
+ defer conn.Close()
+
+ utils.Exec(t, conn, "use `ks/-80`")
+ utils.Exec(t, conn, `begin`)
+ utils.Exec(t, conn, `create table ddl_targeted (id bigint primary key)`)
+ // implicit commit on ddl would have closed the open transaction
+ // so this execution should happen as autocommit.
+ utils.Exec(t, conn, `insert into ddl_targeted (id) values (1)`)
+ // this will have not impact and the row would have inserted.
+ utils.Exec(t, conn, `rollback`)
+ // validating the row
+ utils.AssertMatches(t, conn, `select id from ddl_targeted`, `[[INT64(1)]]`)
+}
+
 func TestLookupErrorMetric(t *testing.T) {
  conn, closer := start(t)
  defer closer()

go/vt/vtgate/engine/send.go

@@ -47,6 +47,8 @@ type Send struct {
  // IsDML specifies how to deal with autocommit behaviour
  IsDML bool
 
+ IsDDL bool
+
  // SingleShardOnly specifies that the query must be send to only single shard
  SingleShardOnly bool
 
@@ -94,6 +96,10 @@ func (s *Send) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[str
  ctx, cancelFunc := addQueryTimeout(ctx, vcursor, s.QueryTimeout)
  defer cancelFunc()
 
+ if err := s.commitIfDDL(ctx, vcursor); err != nil {
+ return nil, err
+ }
+
  rss, err := s.checkAndReturnShards(ctx, vcursor)
  if err != nil {
  return nil, err
@@ -158,6 +164,10 @@ func copyBindVars(in map[string]*querypb.BindVariable) map[string]*querypb.BindV
 
 // TryStreamExecute implements Primitive interface
 func (s *Send) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+ if err := s.commitIfDDL(ctx, vcursor); err != nil {
+ return err
+ }
+
  rss, err := s.checkAndReturnShards(ctx, vcursor)
  if err != nil {
  return err
@@ -204,3 +214,11 @@ func (s *Send) description() PrimitiveDescription {
  Other: other,
  }
 }
+
+// commitIfDDL commits any open transaction before executing the ddl query.
+func (s *Send) commitIfDDL(ctx context.Context, vcursor VCursor) error {
+ if s.IsDDL {
+ return vcursor.Session().Commit(ctx)
+ }
+ return nil
+}

go/vt/vtgate/planbuilder/ddl.go

@@ -45,7 +45,7 @@ func (fk *fkContraint) FkWalk(node sqlparser.SQLNode) (kontinue bool, err error)
 // and which chooses which of the two to invoke at runtime.
 func buildGeneralDDLPlan(ctx context.Context, sql string, ddlStatement sqlparser.DDLStatement, reservedVars *sqlparser.ReservedVars, vschema plancontext.VSchema, enableOnlineDDL, enableDirectDDL bool) (*planResult, error) {
  if vschema.Destination() != nil {
- return buildByPassPlan(sql, vschema)
+ return buildByPassPlan(sql, vschema, true)
  }
  normalDDLPlan, onlineDDLPlan, err := buildDDLPlans(ctx, sql, ddlStatement, reservedVars, vschema, enableOnlineDDL, enableDirectDDL)
  if err != nil {
@@ -80,7 +80,7 @@ func buildGeneralDDLPlan(ctx context.Context, sql string, ddlStatement sqlparser
  return newPlanResult(eddl, tc.getTables()...), nil
 }
 
-func buildByPassPlan(sql string, vschema plancontext.VSchema) (*planResult, error) {
+func buildByPassPlan(sql string, vschema plancontext.VSchema, isDDL bool) (*planResult, error) {
  keyspace, err := vschema.DefaultKeyspace()
  if err != nil {
  return nil, err
@@ -89,6 +89,7 @@ func buildByPassPlan(sql string, vschema plancontext.VSchema) (*planResult, erro
  Keyspace: keyspace,
  TargetDestination: vschema.Destination(),
  Query: sql,
+ IsDDL: isDDL,
  }
  return newPlanResult(send), nil
 }

go/vt/vtgate/planbuilder/show.go

@@ -45,7 +45,7 @@ const (
 
 func buildShowPlan(sql string, stmt *sqlparser.Show, _ *sqlparser.ReservedVars, vschema plancontext.VSchema) (*planResult, error) {
  if vschema.Destination() != nil {
- return buildByPassPlan(sql, vschema)
+ return buildByPassPlan(sql, vschema, false)
  }
 
  var prim engine.Primitive

go/vt/vttablet/endtoend/settings_test.go

@@ -138,16 +138,9 @@ func TestDDLNoConnectionReservationOnSettings(t *testing.T) {
  query := "create table temp(c_date datetime default '0000-00-00')"
  setting := "set sql_mode='TRADITIONAL'"
 
- for _, withTx := range []bool{false, true} {
- if withTx {
- err := client.Begin(false)
- require.NoError(t, err)
- }
- _, err := client.ReserveExecute(query, []string{setting}, nil)
- require.Error(t, err, "create table should have failed with TRADITIONAL mode")
- require.Contains(t, err.Error(), "Invalid default value")
- assert.Zero(t, client.ReservedID())
- }
+ _, err := client.ReserveExecute(query, []string{setting}, nil)
+ assert.ErrorContains(t, err, "Invalid default value for 'c_date'", "create table should have failed with TRADITIONAL mode")
+ assert.Zero(t, client.ReservedID())
 }
 
 func TestDMLNoConnectionReservationOnSettings(t *testing.T) {
@@ -156,7 +149,10 @@ func TestDMLNoConnectionReservationOnSettings(t *testing.T) {
 
  _, err := client.Execute("create table temp(c_date datetime)", nil)
  require.NoError(t, err)
- defer client.Execute("drop table temp", nil)
+ defer func() {
+ client.Rollback()
+ client.Execute("drop table temp", nil)
+ }()
 
  _, err = client.Execute("insert into temp values ('2022-08-25')", nil)
  require.NoError(t, err)
@@ -211,9 +207,8 @@ func TestDDLNoConnectionReservationOnSettingsWithTx(t *testing.T) {
  query := "create table temp(c_date datetime default '0000-00-00')"
  setting := "set sql_mode='TRADITIONAL'"
 
- _, err := client.ReserveBeginExecute(query, []string{setting}, nil, nil)
- require.Error(t, err, "create table should have failed with TRADITIONAL mode")
- require.Contains(t, err.Error(), "Invalid default value")
+ _, err := client.ReserveExecute(query, []string{setting}, nil)
+ require.ErrorContains(t, err, "Invalid default value for 'c_date'", "create table should have failed with TRADITIONAL mode")
  assert.Zero(t, client.ReservedID())
 }
 
@@ -297,12 +292,6 @@ func TestTempTableOnReserveExecute(t *testing.T) {
  require.NoError(t,
  client.Release())
 
- _, err = client.ReserveBeginExecute(tempTblQuery, nil, nil, nil)
- require.NoError(t, err)
- assert.NotZero(t, client.ReservedID())
- require.NoError(t,
- client.Release())
-
  // drop the table
  _, err = client.Execute("drop table if exists temp", nil)
  require.NoError(t, err)
@@ -318,13 +307,6 @@ func TestTempTableOnReserveExecute(t *testing.T) {
  assert.NotZero(t, client.ReservedID(), "as this goes through fallback path of reserving a connection due to temporary tables")
  require.NoError(t,
  client.Release())
-
- _, err = client.ReserveBeginExecute(tempTblQuery, []string{setting}, nil, nil)
- require.Error(t, err, "create table should have failed with TRADITIONAL mode")
- require.Contains(t, err.Error(), "Invalid default value")
- assert.NotZero(t, client.ReservedID(), "as this goes through fallback path of reserving a connection due to temporary tables")
- require.NoError(t,
- client.Release())
 }
 
 func TestInfiniteSessions(t *testing.T) {

go/vt/vttablet/tabletserver/query_executor.go

@@ -29,12 +29,14 @@ import (
  "vitess.io/vitess/go/mysql/replication"
  "vitess.io/vitess/go/mysql/sqlerror"
  "vitess.io/vitess/go/pools/smartconnpool"
-
  "vitess.io/vitess/go/sqltypes"
  "vitess.io/vitess/go/trace"
  "vitess.io/vitess/go/vt/callerid"
  "vitess.io/vitess/go/vt/callinfo"
  "vitess.io/vitess/go/vt/log"
+ querypb "vitess.io/vitess/go/vt/proto/query"
+ topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+ vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
  "vitess.io/vitess/go/vt/schema"
  "vitess.io/vitess/go/vt/sqlparser"
  "vitess.io/vitess/go/vt/tableacl"
@@ -45,10 +47,7 @@ import (
  "vitess.io/vitess/go/vt/vttablet/tabletserver/rules"
  eschema "vitess.io/vitess/go/vt/vttablet/tabletserver/schema"
  "vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
-
- querypb "vitess.io/vitess/go/vt/proto/query"
- topodatapb "vitess.io/vitess/go/vt/proto/topodata"
- vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+ "vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
 )
 
 // QueryExecutor is used for executing a query request.
@@ -192,8 +191,10 @@ func (qre *QueryExecutor) Execute() (reply *sqltypes.Result, err error) {
  return qr, nil
  case p.PlanOtherRead, p.PlanOtherAdmin, p.PlanFlush, p.PlanSavepoint, p.PlanRelease, p.PlanSRollback:
  return qre.execOther()
- case p.PlanInsert, p.PlanUpdate, p.PlanDelete, p.PlanInsertMessage, p.PlanDDL, p.PlanLoad:
+ case p.PlanInsert, p.PlanUpdate, p.PlanDelete, p.PlanInsertMessage, p.PlanLoad:
  return qre.execAutocommit(qre.txConnExec)
+ case p.PlanDDL:
+ return qre.execDDL(nil)
  case p.PlanUpdateLimit, p.PlanDeleteLimit:
  return qre.execAsTransaction(qre.txConnExec)
  case p.PlanCallProc:
@@ -538,7 +539,7 @@ func (qre *QueryExecutor) checkAccess(authorized *tableacl.ACLResult, tableName
  return nil
 }
 
-func (qre *QueryExecutor) execDDL(conn *StatefulConnection) (*sqltypes.Result, error) {
+func (qre *QueryExecutor) execDDL(conn *StatefulConnection) (result *sqltypes.Result, err error) {
  // Let's see if this is a normal DDL statement or an Online DDL statement.
  // An Online DDL statement is identified by /*vt+ .. */ comment with expected directives, like uuid etc.
  if onlineDDL, err := schema.OnlineDDLFromCommentedStatement(qre.plan.FullStmt); err == nil {
@@ -549,6 +550,21 @@ func (qre *QueryExecutor) execDDL(conn *StatefulConnection) (*sqltypes.Result, e
  }
  }
 
+ if conn == nil {
+ conn, err = qre.tsv.te.txPool.createConn(qre.ctx, qre.options, qre.setting)
+ if err != nil {
+ return nil, err
+ }
+ defer conn.Release(tx.ConnRelease)
+ }
+
+ // A DDL statement should commit the current transaction in the VTGate.
+ // The change was made in PR: https://github.com/vitessio/vitess/pull/14110 in v18.
+ // DDL statement received by vttablet will be outside of a transaction.
+ if conn.txProps != nil {
+ return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "DDL statement executed inside a transaction")
+ }
+
  isTemporaryTable := false
  if ddlStmt, ok := qre.plan.FullStmt.(sqlparser.DDLStatement); ok {
  isTemporaryTable = ddlStmt.IsTemporary()
@@ -580,19 +596,7 @@ func (qre *QueryExecutor) execDDL(conn *StatefulConnection) (*sqltypes.Result, e
  return nil, err
  }
  }
- result, err := qre.execStatefulConn(conn, sql, true)
- if err != nil {
- return nil, err
- }
- // Only perform this operation when the connection has transaction open.
- // TODO: This actually does not retain the old transaction. We should see how to provide correct behaviour to client.
- if conn.txProps != nil {
- err = qre.BeginAgain(qre.ctx, conn)
- if err != nil {
- return nil, err
- }
- }
- return result, nil
+ return qre.execStatefulConn(conn, sql, true)
 }
 
 func (qre *QueryExecutor) execLoad(conn *StatefulConnection) (*sqltypes.Result, error) {
@@ -603,20 +607,6 @@ func (qre *QueryExecutor) execLoad(conn *StatefulConnection) (*sqltypes.Result,
  return result, nil
 }
 
-// BeginAgain commits the existing transaction and begins a new one
-func (*QueryExecutor) BeginAgain(ctx context.Context, dc *StatefulConnection) error {
- if dc.IsClosed() || dc.TxProperties().Autocommit {
- return nil
- }
- if _, err := dc.Exec(ctx, "commit", 1, false); err != nil {
- return err
- }
- if _, err := dc.Exec(ctx, "begin", 1, false); err != nil {
- return err
- }
- return nil
-}
-
 func (qre *QueryExecutor) execNextval() (*sqltypes.Result, error) {
  env := evalengine.NewExpressionEnv(qre.ctx, qre.bindVars, evalengine.NewEmptyVCursor(qre.tsv.Environment(), time.Local))
  result, err := env.Evaluate(qre.plan.NextCount)