loaddata: revert concurrent writing to TiKV and WITH detached (#44056)

ref #44078
pingcap · May 23, 2023 · 0ac7473 · 0ac7473
1 parent 8c1fae8
commit 0ac7473
Show file tree

Hide file tree

Showing 10 changed files with 95 additions and 277 deletions.
diff --git a/executor/BUILD.bazel b/executor/BUILD.bazel
@@ -312,7 +312,6 @@ go_test(
         "join_pkg_test.go",
         "join_test.go",
         "joiner_test.go",
-        "load_data_test.go",
         "main_test.go",
         "memtable_reader_test.go",
         "merge_join_test.go",

diff --git a/executor/importer/import.go b/executor/importer/import.go
@@ -83,14 +83,13 @@ const (
 	maxWriteSpeedOption = "max_write_speed"
 	splitFileOption     = "split_file"
 	recordErrorsOption  = "record_errors"
+	detachedOption      = "detached"
 
 	// test option, not for user
 	distributedOption = "__distributed"
 )
 
 var (
-	detachedOption = plannercore.DetachedOption
-
 	// name -> whether the option has value
 	supportedOptions = map[string]bool{
 		importModeOption:    true,

diff --git a/executor/insert_common.go b/executor/insert_common.go
@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"sync"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -93,6 +94,11 @@ type InsertValues struct {
 
 	stats *InsertRuntimeStat
 
+	// in LOAD DATA, one InsertValues is used by two goroutine, we need to lock
+	// when using the txn
+	isLoadData bool
+	txnInUse   sync.Mutex
+
 	// fkChecks contains the foreign key checkers.
 	fkChecks   []*FKCheckExec
 	fkCascades []*FKCascadeExec
@@ -1050,6 +1056,10 @@ func (e *InsertValues) allocAutoRandomID(ctx context.Context, fieldType *types.F
 	if shardFmt.IncrementalMask()&autoRandomID != autoRandomID {
 		return 0, autoid.ErrAutoRandReadFailed
 	}
+	if e.isLoadData {
+		e.txnInUse.Lock()
+		defer e.txnInUse.Unlock()
+	}
 	_, err = e.ctx.Txn(true)
 	if err != nil {
 		return 0, err