ddl to commit open transaction before sending ddl to vttablet on shar…

…d targeting

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

go/test/endtoend/vtgate/misc_test.go

@@ -798,6 +798,18 @@ 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)`)
+ utils.Exec(t, conn, `commit`)
+}
+
 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) {