Errors returned from Flush are not equally treated as those returned from Commit

[ekexium]

Bug Report

ref #50215

1. Minimal reproduce step (Required)

See the case TestPipelinedDMLDisableRetry.

func TestPipelinedDMLDisableRetry(t *testing.T) {
	// the case tests that
	// 1. auto-retry for pipelined dml is disabled
	// 2. the write conflict error message returned from a Flush (instead of from Commit) is correct
	require.Nil(t, failpoint.Enable("tikvclient/pipelinedMemDBMinFlushKeys", `return(1)`))
	require.Nil(t, failpoint.Enable("tikvclient/pipelinedMemDBMinFlushSize", `return(1)`))
	require.Nil(t, failpoint.Enable("tikvclient/pipelinedMemDBForceFlushSizeThreshold", `return(1)`))
	defer func() {
		require.Nil(t, failpoint.Disable("tikvclient/pipelinedMemDBMinFlushKeys"))
		require.Nil(t, failpoint.Disable("tikvclient/pipelinedMemDBMinFlushSize"))
		require.Nil(t, failpoint.Disable("tikvclient/pipelinedMemDBForceFlushSizeThreshold"))
	}()
	store := realtikvtest.CreateMockStoreAndSetup(t)
	tk1 := testkit.NewTestKit(t, store)
	tk1.Session().GetSessionVars().InitChunkSize = 1
	tk1.Session().GetSessionVars().MaxChunkSize = 1
	tk2 := testkit.NewTestKit(t, store)
	tk1.MustExec("use test")
	tk2.MustExec("use test")
	tk1.MustExec("drop table if exists t")
	tk1.MustExec("create table t(a int primary key, b int)")
	// we need to avoid inserting *literals* into t, so let t2 be the source table.
	tk1.MustExec("create table t2(a int, b int)")
	tk1.MustExec("insert into t2 values (1, 1), (2, 1)")
	require.Nil(t, failpoint.Enable("tikvclient/beforePipelinedFlush", `pause`))
	tk1.MustExec("set session tidb_dml_type = bulk")
	errCh := make(chan error)
	go func() {
		// we expect that this stmt triggers 2 flushes, each containing only 1 row.
		errCh <- tk1.ExecToErr("insert into t select * from t2 order by a")
	}()
	time.Sleep(500 * time.Millisecond)
	tk2.MustExec("insert into t values (1,2)")
	require.Nil(t, failpoint.Disable("tikvclient/beforePipelinedFlush"))
	err := <-errCh
	require.Error(t, err)
	require.True(t, kv.ErrWriteConflict.Equal(err), fmt.Sprintf("error: %s", err))
	require.ErrorContains(t, err, "tableName=test.t")
}

2. What did you expect to see? (Required)

Write conflict and assertion errors should be handled.

3. What did you see instead (Required)

They are directly returned to user.

4. What is your TiDB version? (Required)

Master(6ed307e)