Use Kill Query for Non-Transaction Query Execution and Update Query Timeout / Cancelled Error Message

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)
@@ -193,6 +194,11 @@ It should be enabled in VTGate with the `--enable-udfs` flag.
 
 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