dxf: use global entry_size_limit when executing SQLs (#59516)

close #59506

Author: D3Hunter

pkg/disttask/framework/storage/BUILD.bazel

@@ -42,7 +42,7 @@ go_test(
     embed = [":storage"],
     flaky = True,
     race = "on",
-    shard_count = 23,
+    shard_count = 24,
     deps = [
         "//pkg/config",
         "//pkg/disttask/framework/proto",

pkg/disttask/framework/storage/table_test.go

@@ -1185,3 +1185,30 @@ func TestGetActiveTaskExecInfo(t *testing.T) {
 	require.NoError(t, err)
 	require.Empty(t, taskExecInfos)
 }
+
+func TestTaskManagerEntrySize(t *testing.T) {
+	store, tm, ctx := testutil.InitTableTest(t)
+	getMeta := func(l int) []byte {
+		meta := make([]byte, l)
+		for i := 0; i < l; i++ {
+			meta[i] = 'a'
+		}
+		return meta
+	}
+	insertSubtask := func(meta []byte) error {
+		return tm.WithNewSession(func(se sessionctx.Context) error {
+			_, err := sqlexec.ExecSQL(ctx, se.GetSQLExecutor(), `
+			insert into mysql.tidb_background_subtask(`+storage.InsertSubtaskColumns+`) values`+
+				`(%?, %?, %?, %?, %?, %?, %?, NULL, CURRENT_TIMESTAMP(), '{}', '{}')`,
+				1, "1", "execID", meta, proto.SubtaskStatePending, proto.Type2Int(proto.TaskTypeExample), 1)
+			return err
+		})
+	}
+	meta6m := getMeta(6 << 20)
+	require.ErrorContains(t, insertSubtask(meta6m), "entry too large")
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec(fmt.Sprintf("set global tidb_txn_entry_size_limit = %d", 7<<20))
+	require.NoError(t, insertSubtask(meta6m))
+	// TiKV also have a limit raftstore.raft-entry-max-size which is 8M by default,
+	// we won't test that param here
+}

pkg/disttask/framework/storage/task_table.go

@@ -141,12 +141,20 @@ func SetTaskManager(is *TaskManager) {
 
 // WithNewSession executes the function with a new session.
 func (mgr *TaskManager) WithNewSession(fn func(se sessionctx.Context) error) error {
-	se, err := mgr.sePool.Get()
+	v, err := mgr.sePool.Get()
 	if err != nil {
 		return err
 	}
-	defer mgr.sePool.Put(se)
-	return fn(se.(sessionctx.Context))
+	// when using global sort, the subtask meta might quite large as it include
+	// filenames of all the generated kv/stat files.
+	se := v.(sessionctx.Context)
+	limitBak := se.GetSessionVars().TxnEntrySizeLimit
+	defer func() {
+		se.GetSessionVars().TxnEntrySizeLimit = limitBak
+		mgr.sePool.Put(v)
+	}()
+	se.GetSessionVars().TxnEntrySizeLimit = vardef.TxnEntrySizeLimit.Load()
+	return fn(se)
 }
 
 // WithNewTxn executes the fn in a new transaction.