ddl: rely on error-path cleanup

Author: wjhuang2016

pkg/ddl/backfilling_read_index.go

@@ -94,16 +94,6 @@ type readIndexEngineRegistrar interface {
 	Register(indexIDs []int64, uniques []bool, tbl table.Table) ([]ingest.Engine, error)
 }
 
-const pebbleLockHeldByCurrentProcessMsg = "lock held by current process"
-
-func isPebbleLockHeldByCurrentProcess(err error) bool {
-	// Pebble's vfs.Lock returns a plain error containing `pebbleLockHeldByCurrentProcessMsg`
-	// when the lock is already held by the current process (see pebble/vfs/file_lock_unix.go).
-	// Pebble does not export a sentinel error/type for this case, so we must match on the
-	// message.
-	return err != nil && strings.Contains(err.Error(), pebbleLockHeldByCurrentProcessMsg)
-}
-
 func newReadIndexExecutor(
 	store kv.Storage,
 	sessPool *sess.Pool,
@@ -415,7 +405,7 @@ func (r *readIndexStepExecutor) buildLocalStorePipeline(
 		}
 		idxNames.WriteString(index.Name.O)
 	}
-	engines, err := registerReadIndexEngines(wctx, r.job.ID, backendCtx, indexIDs, uniques, r.ptbl)
+	engines, err := registerReadIndexEngines(backendCtx, indexIDs, uniques, r.ptbl)
 	if err != nil {
 		tidblogutil.Logger(wctx).Error("cannot register new engine",
 			zap.Error(err),
@@ -513,24 +503,8 @@ func cleanupReadIndexLocalEngines(jobID, subtaskID int64, backendCtx readIndexLo
 	}
 }
 
-func registerReadIndexEngines(
-	wctx *workerpool.Context,
-	jobID int64,
-	backendCtx readIndexEngineRegistrar,
-	indexIDs []int64,
-	uniques []bool,
-	tbl table.Table,
-) ([]ingest.Engine, error) {
-	engines, err := backendCtx.Register(indexIDs, uniques, tbl)
-	if err != nil && isPebbleLockHeldByCurrentProcess(err) {
-		// Cleanup happens in the caller's error-path defer; dist-task retry will re-run.
-		tidblogutil.Logger(wctx).Warn("register ingest engine got lock held",
-			zap.Error(err),
-			zap.Int64("job ID", jobID),
-			zap.Int64s("index IDs", indexIDs),
-		)
-	}
-	return engines, err
+func registerReadIndexEngines(backendCtx readIndexEngineRegistrar, indexIDs []int64, uniques []bool, tbl table.Table) ([]ingest.Engine, error) {
+	return backendCtx.Register(indexIDs, uniques, tbl)
 }
 
 type distTaskRowCntCollector struct {

pkg/ddl/backfilling_test.go

@@ -18,7 +18,6 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"fmt"
 	"math"
 	"sync/atomic"
 	"testing"
@@ -110,21 +109,8 @@ func (c *testReadIndexBackendCtx) Register(indexIDs []int64, uniques []bool, tbl
 
 func (c *testReadIndexBackendCtx) GetLocalBackend() *local.Backend { return c.localBackend }
 
-func TestIsPebbleLockHeldByCurrentProcess(t *testing.T) {
-	// Matches Pebble's lock error message (no exported sentinel).
-	require.False(t, isPebbleLockHeldByCurrentProcess(nil))
-	require.False(t, isPebbleLockHeldByCurrentProcess(errors.New("other error")))
-
-	baseErr := errors.New("lock held by current process")
-	require.True(t, isPebbleLockHeldByCurrentProcess(baseErr))
-	require.True(t, isPebbleLockHeldByCurrentProcess(fmt.Errorf("wrapped: %w", baseErr)))
-}
-
 func TestRegisterReadIndexEnginesNoRetryOnLockHeld(t *testing.T) {
 	// No in-function retry; cleanup is centralized in caller error-path.
-	wctx := workerpool.NewContext(context.Background())
-	defer wctx.Cancel()
-
 	var registerCalls int32
 	backendCtx := &testReadIndexBackendCtx{
 		localBackend: &local.Backend{},
@@ -134,7 +120,7 @@ func TestRegisterReadIndexEnginesNoRetryOnLockHeld(t *testing.T) {
 		},
 	}
 
-	engines, err := registerReadIndexEngines(wctx, 1, backendCtx, []int64{1}, []bool{false}, nil)
+	engines, err := registerReadIndexEngines(backendCtx, []int64{1}, []bool{false}, nil)
 	require.ErrorContains(t, err, "lock held by current process")
 	require.Nil(t, engines)
 	require.Equal(t, int32(1), registerCalls)