Use Kill Query for Non-Transaction Query Execution and Update Query T…

…imeout / Cancelled Error Message (#15694)

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

changelog/20.0/20.0.0/summary.md

@@ -22,6 +22,7 @@
  - [Delete with Subquery Support](#delete-subquery)
  - [Delete with Multi Target Support](#delete-multi-target)
  - [User Defined Functions Support](#udf-support)
+ - **[Query Timeout](#query-timeout)**
  - **[Flag changes](#flag-changes)**
  - [`pprof-http` default change](#pprof-http-default)
  - [New `healthcheck-dial-concurrency` flag](#healthcheck-dial-concurrency-flag)
@@ -196,6 +197,11 @@ Without this flag, VTGate will not be aware that there might be aggregating user
 
 More details about how to load UDFs is available in [MySQL Docs](https://dev.mysql.com/doc/extending-mysql/8.0/en/adding-loadable-function.html)
 
+### <a id="query-timeout"/>Query Timeout
+On a query timeout, Vitess closed the connection using the `kill connection` statement. This leads to connection churn 
+which is not desirable in some cases. To avoid this, Vitess now uses the `kill query` statement to cancel the query. 
+This will only cancel the query and does not terminate the connection.
+
 ### <a id="flag-changes"/>Flag Changes
 
 #### <a id="pprof-http-default"/> `pprof-http` Default Change

go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go

@@ -583,7 +583,7 @@ func testScheduler(t *testing.T) {
  onlineddl.CheckCompleteMigration(t, &vtParams, shards, t1uuid, true)
  })
  t.Run("cut-over fail due to timeout", func(t *testing.T) {
- waitForMessage(t, t1uuid, "due to context deadline exceeded")
+ waitForMessage(t, t1uuid, "(errno 3024) (sqlstate HY000): Query execution was interrupted, maximum statement execution time exceeded")
  status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t1uuid, normalWaitTime, schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusRunning)
  fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
  onlineddl.CheckMigrationStatus(t, &vtParams, shards, t1uuid, schema.OnlineDDLStatusRunning)
@@ -606,7 +606,7 @@ func testScheduler(t *testing.T) {
  })
  t.Run("expect transaction failure", func(t *testing.T) {
  select {
- case commitTransactionChan <- true: //good
+ case commitTransactionChan <- true: // good
  case <-ctx.Done():
  assert.Fail(t, ctx.Err().Error())
  }
@@ -1445,7 +1445,7 @@ DROP TABLE IF EXISTS stress_test
  }
  })
 
- //DROP
+ // DROP
 
  t.Run("online DROP TABLE", func(t *testing.T) {
  uuid := testOnlineDDLStatement(t, createParams(dropStatement, onlineSingletonDDLStrategy, "vtgate", "", "", "", false))

go/vt/vttablet/endtoend/config_test.go

@@ -195,7 +195,7 @@ func TestConsolidatorReplicasOnly(t *testing.T) {
 func TestQueryPlanCache(t *testing.T) {
  var cachedPlanSize = int((&tabletserver.TabletPlan{}).CachedSize(true))
 
- //sleep to avoid race between SchemaChanged event clearing out the plans cache which breaks this test
+ // sleep to avoid race between SchemaChanged event clearing out the plans cache which breaks this test
  framework.Server.WaitForSchemaReset(2 * time.Second)
 
  bindVars := map[string]*querypb.BindVariable{
@@ -276,7 +276,7 @@ func TestQueryTimeout(t *testing.T) {
  assert.Equal(t, vtrpcpb.Code_ABORTED, vterrors.Code(err))
  vend := framework.DebugVars()
  verifyIntValue(t, vend, "QueryTimeout", int(100*time.Millisecond))
- compareIntDiff(t, vend, "Kills/Queries", vstart, 1)
+ compareIntDiff(t, vend, "Kills/Connections", vstart, 1)
 }
 
 func changeVar(t *testing.T, name, value string) (revert func()) {

go/vt/vttablet/endtoend/transaction_test.go

@@ -471,7 +471,7 @@ func TestShutdownGracePeriodWithReserveExecute(t *testing.T) {
  client.Rollback()
 }
 
-func TestShortTxTimeout(t *testing.T) {
+func TestShortTxTimeoutOltp(t *testing.T) {
  client := framework.NewClient()
  defer framework.Server.Config().SetTxTimeoutForWorkload(
  framework.Server.Config().TxTimeoutForWorkload(querypb.ExecuteOptions_OLTP),
@@ -488,6 +488,23 @@ func TestShortTxTimeout(t *testing.T) {
  client.Rollback()
 }
 
+func TestShortTxTimeoutOlap(t *testing.T) {
+ client := framework.NewClient()
+ defer framework.Server.Config().SetTxTimeoutForWorkload(
+ framework.Server.Config().TxTimeoutForWorkload(querypb.ExecuteOptions_OLAP),
+ querypb.ExecuteOptions_OLAP,
+ )
+ framework.Server.Config().SetTxTimeoutForWorkload(10*time.Millisecond, querypb.ExecuteOptions_OLAP)
+
+ err := client.Begin(false)
+ require.NoError(t, err)
+ start := time.Now()
+ _, err = client.StreamExecute("select sleep(10) from dual", nil)
+ assert.Error(t, err)
+ assert.True(t, time.Since(start) < 5*time.Second, time.Since(start))
+ client.Rollback()
+}
+
 func TestMMCommitFlow(t *testing.T) {
  client := framework.NewClient()
  defer client.Execute("delete from vitess_test where intval=4", nil)

go/vt/vttablet/onlineddl/executor.go

@@ -974,7 +974,10 @@ func (e *Executor) cutOverVReplMigration(ctx context.Context, s *VReplStream, sh
  defer renameConn.Recycle()
  defer func() {
  if !renameWasSuccessful {
- renameConn.Conn.Kill("premature exit while renaming tables", 0)
+ err := renameConn.Conn.Kill("premature exit while renaming tables", 0)
+ if err != nil {
+ log.Warningf("Failed to kill connection being used to rename tables in OnlineDDL migration %s: %v", onlineDDL.UUID, err)
+ }
  }
  }()
  // See if backend MySQL server supports 'rename_table_preserve_foreign_key' variable