ttl: enable 1pc for ttl delete

ttl/ttlworker/session.go

@@ -19,6 +19,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/pingcap/tidb/parser/terror"
+
 	"github.com/ngaut/pools"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/sessionctx"
@@ -61,12 +63,24 @@ func getSession(pool sessionPool) (session.Session, error) {
 	}
 
 	originalRetryLimit := sctx.GetSessionVars().RetryLimit
+	originalEnable1PC := sctx.GetSessionVars().Enable1PC
+	originalEnableAsyncCommit := sctx.GetSessionVars().EnableAsyncCommit
 	se := session.NewSession(sctx, exec, func(se session.Session) {
 		_, err = se.ExecuteSQL(context.Background(), fmt.Sprintf("set tidb_retry_limit=%d", originalRetryLimit))
 		if err != nil {
 			logutil.BgLogger().Error("fail to reset tidb_retry_limit", zap.Int64("originalRetryLimit", originalRetryLimit), zap.Error(err))
 		}
 
+		if !originalEnable1PC {
+			_, err = se.ExecuteSQL(context.Background(), "set tidb_enable_1pc=OFF")
+			terror.Log(err)
+		}
+
+		if !originalEnableAsyncCommit {
+			_, err = se.ExecuteSQL(context.Background(), "set tidb_enable_async_commit=OFF")
+			terror.Log(err)
+		}
+
 		pool.Put(resource)
 	})
 
@@ -77,6 +91,20 @@ func getSession(pool sessionPool) (session.Session, error) {
 		return nil, err
 	}
 
+	// set enable 1pc to ON
+	_, err = se.ExecuteSQL(context.Background(), "set tidb_enable_1pc=ON")
+	if err != nil {
+		se.Close()
+		return nil, err
+	}
+
+	// set enable async commit to ON
+	_, err = se.ExecuteSQL(context.Background(), "set tidb_enable_async_commit=ON")
+	if err != nil {
+		se.Close()
+		return nil, err
+	}
+
 	// Force rollback the session to guarantee the session is not in any explicit transaction
 	if _, err = se.ExecuteSQL(context.Background(), "ROLLBACK"); err != nil {
 		se.Close()