*: Import Into support full-text index using TiCI

lightning/pkg/importer/table_import_test.go

@@ -881,7 +881,7 @@ func (s *tableRestoreSuite) TestImportKVSuccess() {
 		CloseEngine(ctx, nil, engineUUID).
 		Return(nil)
 	mockBackend.EXPECT().
-		ImportEngine(ctx, engineUUID, gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, engineUUID, gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil)
 	mockBackend.EXPECT().
 		CleanupEngine(ctx, engineUUID).
@@ -916,7 +916,7 @@ func (s *tableRestoreSuite) TestImportKVFailure() {
 		CloseEngine(ctx, nil, engineUUID).
 		Return(nil)
 	mockBackend.EXPECT().
-		ImportEngine(ctx, engineUUID, gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, engineUUID, gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(errors.Annotate(context.Canceled, "fake import error"))
 
 	closedEngine, err := importer.UnsafeCloseEngineWithUUID(ctx, nil, "tag", engineUUID, 0)
@@ -987,7 +987,7 @@ func (s *tableRestoreSuite) TestTableRestoreMetrics() {
 	backendObj := mock.NewMockBackend(controller)
 	backendObj.EXPECT().OpenEngine(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	backendObj.EXPECT().CloseEngine(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-	backendObj.EXPECT().ImportEngine(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	backendObj.EXPECT().ImportEngine(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	backendObj.EXPECT().CleanupEngine(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	backendObj.EXPECT().ShouldPostProcess().Return(false).AnyTimes()
 	backendObj.EXPECT().LocalWriter(gomock.Any(), gomock.Any(), gomock.Any()).Return(mockEngineWriter, nil).AnyTimes()

pkg/ddl/backfilling_import_cloud.go

@@ -104,6 +104,7 @@ func (e *cloudImportExecutor) RunSubtask(ctx context.Context, subtask *proto.Sub
 	}
 
 	_, engineUUID := backend.MakeUUID(e.ptbl.Meta().Name.L, idxID)
+	engineID := int32(common.IndexEngineID)
 
 	all := external.SortedKVMeta{}
 	for _, g := range sm.MetaGroups {
@@ -135,7 +136,7 @@ func (e *cloudImportExecutor) RunSubtask(ctx context.Context, subtask *proto.Sub
 	if err != nil {
 		return err
 	}
-	err = localBackend.ImportEngine(ctx, engineUUID, int64(config.SplitRegionSize), int64(config.SplitRegionKeys))
+	err = localBackend.ImportEngine(ctx, engineUUID, engineID, int64(config.SplitRegionSize), int64(config.SplitRegionKeys))
 	failpoint.Inject("mockCloudImportRunSubtaskError", func(_ failpoint.Value) {
 		err = context.DeadlineExceeded
 	})

pkg/disttask/importinto/task_executor.go

@@ -525,6 +525,16 @@ func (e *writeAndIngestStepExecutor) RunSubtask(ctx context.Context, subtask *pr
 	}()
 
 	_, engineUUID := backend.MakeUUID("", subtask.ID)
+
+	// We specify engineID as a tag so that the callee can determine
+	// whether the call is from a data engine or an index engine.
+	var engineID int32
+	if sm.KVGroup == dataKVGroup {
+		engineID = int32(0)
+	} else {
+		engineID = int32(common.IndexEngineID)
+	}
+
 	localBackend := e.tableImporter.Backend()
 	// compatible with old version task meta
 	jobKeys := sm.RangeJobKeys
@@ -557,7 +567,7 @@ func (e *writeAndIngestStepExecutor) RunSubtask(ctx context.Context, subtask *pr
 	if err != nil {
 		return err
 	}
-	err = localBackend.ImportEngine(ctx, engineUUID, int64(config.SplitRegionSize), int64(config.SplitRegionKeys))
+	err = localBackend.ImportEngine(ctx, engineUUID, engineID, int64(config.SplitRegionSize), int64(config.SplitRegionKeys))
 	if err != nil {
 		return errors.Trace(err)
 	}

pkg/domain/infosync/BUILD.bazel

@@ -11,7 +11,6 @@ go_library(
         "region.go",
         "resource_manager_client.go",
         "schedule_manager.go",
-        "tici_manager_client.go",
         "tiflash_manager.go",
     ],
     importpath = "github.com/pingcap/tidb/pkg/domain/infosync",
@@ -59,8 +58,6 @@ go_library(
         "@com_github_tikv_pd_client//opt",
         "@io_etcd_go_etcd_client_v3//:client",
         "@io_etcd_go_etcd_client_v3//concurrency",
-        "@org_golang_google_grpc//:grpc",
-        "@org_golang_google_grpc//credentials/insecure",
         "@org_uber_go_zap//:zap",
     ],
 )

pkg/domain/infosync/info.go

@@ -47,6 +47,7 @@ import (
 	"github.com/pingcap/tidb/pkg/resourcegroup"
 	"github.com/pingcap/tidb/pkg/session/cursor"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
+	"github.com/pingcap/tidb/pkg/tici"
 	util2 "github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/dbterror"
 	"github.com/pingcap/tidb/pkg/util/engine"
@@ -1345,11 +1346,11 @@ func GetTiFlashStoresStat(ctx context.Context) (*pdhttp.StoresInfo, error) {
 
 // CreateFulltextIndex create fulltext infex on TiCI.
 func CreateFulltextIndex(ctx context.Context, tblInfo *model.TableInfo, indexInfo *model.IndexInfo, schemaName string) error {
-	ticiManager, err := NewTiCIManager("0.0.0.0", "50061")
+	ticiManager, err := tici.NewTiCIManager("0.0.0.0", "50061")
 	if err != nil {
 		return err
 	}
-	defer ticiManager.conn.Close()
+	defer ticiManager.Conn.Close()
 	return ticiManager.CreateFulltextIndex(ctx, tblInfo, indexInfo, schemaName)
 }
 
@@ -1637,3 +1638,33 @@ func (is *InfoSyncer) setDynamicServerInfo(ds *DynamicServerInfo) {
 	}
 	is.info.Store(newInfo)
 }
+
+// GetCloudStoragePath requests the S3 path from TiCI Meta Service for a baseline shard upload.
+func GetCloudStoragePath(
+	ctx context.Context,
+	tblInfo *model.TableInfo,
+	indexInfo *model.IndexInfo,
+	schemaName string,
+	lowerBound, upperBound []byte,
+) (string, error) {
+	ticiManager, err := tici.NewTiCIManager("0.0.0.0", "50061")
+	if err != nil {
+		return "", err
+	}
+	defer ticiManager.Conn.Close()
+	return ticiManager.GetCloudStoragePath(ctx, tblInfo, indexInfo, schemaName, lowerBound, upperBound)
+}
+
+// MarkTableUploadFinished notifies TiCI Meta Service that the whole table/index upload is finished.
+func MarkTableUploadFinished(
+	ctx context.Context,
+	tableID int64,
+	indexID int64,
+) error {
+	ticiManager, err := tici.NewTiCIManager("0.0.0.0", "50061")
+	if err != nil {
+		return err
+	}
+	defer ticiManager.Conn.Close()
+	return ticiManager.MarkTableUploadFinished(ctx, tableID, indexID)
+}

pkg/executor/importer/table_import.go

@@ -201,6 +201,8 @@ func NewTableImporter(
 		return nil, err
 	}
 
+	localBackend.SetTiCIWriterGroup(ctx, e.Table.Meta(), e.DBName)
+
 	return &TableImporter{
 		LoadDataController: e,
 		id:                 id,

pkg/lightning/backend/backend.go

@@ -204,7 +204,7 @@ type Backend interface {
 	// ImportEngine imports engine data to the backend. If it returns ErrDuplicateDetected,
 	// it means there is duplicate detected. For this situation, all data in the engine must be imported.
 	// It's safe to reset or cleanup this engine.
-	ImportEngine(ctx context.Context, engineUUID uuid.UUID, regionSplitSize, regionSplitKeys int64) error
+	ImportEngine(ctx context.Context, engineUUID uuid.UUID, engineID int32, regionSplitSize, regionSplitKeys int64) error
 
 	CleanupEngine(ctx context.Context, engineUUID uuid.UUID) error
 
@@ -394,7 +394,7 @@ func (engine *ClosedEngine) Import(ctx context.Context, regionSplitSize, regionS
 
 	for i := range importMaxRetryTimes {
 		task := engine.logger.With(zap.Int("retryCnt", i)).Begin(zap.InfoLevel, "import")
-		err = engine.backend.ImportEngine(ctx, engine.uuid, regionSplitSize, regionSplitKeys)
+		err = engine.backend.ImportEngine(ctx, engine.uuid, engine.id, regionSplitSize, regionSplitKeys)
 		if !common.IsRetryableError(err) {
 			if common.ErrFoundDuplicateKeys.Equal(err) {
 				task.End(zap.WarnLevel, err)

pkg/lightning/backend/backend_test.go

@@ -72,7 +72,7 @@ func TestOpenCloseImportCleanUpEngine(t *testing.T) {
 		Return(nil).
 		After(openCall)
 	importCall := s.mockBackend.EXPECT().
-		ImportEngine(ctx, engineUUID, gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, engineUUID, gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).
 		After(closeCall)
 	s.mockBackend.EXPECT().
@@ -276,7 +276,7 @@ func TestImportFailedNoRetry(t *testing.T) {
 
 	s.mockBackend.EXPECT().CloseEngine(ctx, nil, gomock.Any()).Return(nil)
 	s.mockBackend.EXPECT().
-		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(errors.Annotate(context.Canceled, "fake unrecoverable import error"))
 
 	closedEngine, err := s.engineMgr.UnsafeCloseEngine(ctx, nil, "`db`.`table`", 1)
@@ -294,7 +294,7 @@ func TestImportFailedWithRetry(t *testing.T) {
 
 	s.mockBackend.EXPECT().CloseEngine(ctx, nil, gomock.Any()).Return(nil)
 	s.mockBackend.EXPECT().
-		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(errors.Annotate(driver.ErrBadConn, "fake recoverable import error")).
 		MinTimes(2)
 	s.mockBackend.EXPECT().RetryImportDelay().Return(time.Duration(0)).AnyTimes()
@@ -314,10 +314,10 @@ func TestImportFailedRecovered(t *testing.T) {
 
 	s.mockBackend.EXPECT().CloseEngine(ctx, nil, gomock.Any()).Return(nil)
 	s.mockBackend.EXPECT().
-		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(gmysql.ErrInvalidConn)
 	s.mockBackend.EXPECT().
-		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any()).
+		ImportEngine(ctx, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil)
 	s.mockBackend.EXPECT().RetryImportDelay().Return(time.Duration(0)).AnyTimes()
 

pkg/lightning/backend/local/BUILD.bazel

@@ -64,6 +64,7 @@ go_library(
         "//pkg/table",
         "//pkg/table/tables",
         "//pkg/tablecodec",
+        "//pkg/tici",
         "//pkg/util",
         "//pkg/util/codec",
         "//pkg/util/compress",

pkg/lightning/backend/local/local.go

@@ -56,6 +56,7 @@ import (
 	"github.com/pingcap/tidb/pkg/lightning/tikv"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/tablecodec"
+	"github.com/pingcap/tidb/pkg/tici"
 	"github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/codec"
 	"github.com/pingcap/tidb/pkg/util/engine"
@@ -531,6 +532,8 @@ type Backend struct {
 	logger        log.Logger
 
 	nextgenHTTPCli *http.Client
+
+	ticiWriteGroup *tici.DataWriterGroup // TiCI writer group
 }
 
 var _ DiskUsage = (*Backend)(nil)
@@ -1145,6 +1148,7 @@ func checkDiskAvail(ctx context.Context, store *pdhttp.StoreInfo) error {
 func (local *Backend) ImportEngine(
 	ctx context.Context,
 	engineUUID uuid.UUID,
+	engineID int32,
 	regionSplitSize, regionSplitKeys int64,
 ) error {
 	kvRegionSplitSize, kvRegionSplitKeys, err := GetRegionSplitSizeKeys(ctx, local.pdCli, local.tls)
@@ -1173,6 +1177,9 @@ func (local *Backend) ImportEngine(
 		localEngine.regionSplitKeyCnt = regionSplitKeys
 		e = localEngine
 	}
+
+	tici.SetTiCIDataWriterGroupWritable(ctx, local.ticiWriteGroup, engineUUID, engineID)
+
 	lfTotalSize, lfLength := e.KVStatistics()
 	if lfTotalSize == 0 {
 		// engine is empty, this is likes because it's a index engine but the table contains no index
@@ -1454,6 +1461,14 @@ func (local *Backend) doImport(
 	if err != nil && !common.IsContextCanceledError(err) {
 		log.FromContext(ctx).Error("do import meets error", zap.Error(err))
 	}
+
+	if err == nil && local.ticiWriteGroup != nil {
+		// If the import is done, we can close the write group.
+		if err2 := local.ticiWriteGroup.MarkTableUploadFinished(ctx); err2 != nil {
+			err = err2 // mark upload itself failed
+		}
+	}
+
 	return err
 }
 
@@ -1750,3 +1765,8 @@ func GetRegionSplitSizeKeys(ctx context.Context, cli pd.Client, tls *common.TLS)
 	}
 	return 0, 0, errors.New("get region split size and keys failed")
 }
+
+// SetTiCIWriterGroup initializes the ticiWriteGroup field for the Backend using the given table info and schema.
+func (local *Backend) SetTiCIWriterGroup(ctx context.Context, tblInfo *model.TableInfo, schema string) {
+	local.ticiWriteGroup = tici.NewTiCIDataWriterGroup(ctx, tblInfo, schema)
+}