*: don't handle live updates of column size (#58596)

close #58595

Author: winoros

pkg/disttask/importinto/proto.go

@@ -110,7 +110,6 @@ type SharedVars struct {
 	TableImporter *importer.TableImporter
 	DataEngine    *backend.OpenedEngine
 	IndexEngine   *backend.OpenedEngine
-	Progress      *importer.Progress
 
 	mu       sync.Mutex
 	Checksum *verification.KVGroupChecksum
@@ -183,5 +182,4 @@ type Checksum struct {
 // This portion of the code may be implemented uniformly in the framework in the future.
 type Result struct {
 	LoadedRowCnt uint64
-	ColSizeMap   map[int64]int64
 }

pkg/disttask/importinto/scheduler.go

@@ -577,14 +577,9 @@ func updateResult(handle storage.TaskHandle, task *proto.Task, taskMeta *TaskMet
 		}
 		subtaskMetas = append(subtaskMetas, &subtaskMeta)
 	}
-	columnSizeMap := make(map[int64]int64)
 	for _, subtaskMeta := range subtaskMetas {
 		taskMeta.Result.LoadedRowCnt += subtaskMeta.Result.LoadedRowCnt
-		for key, val := range subtaskMeta.Result.ColSizeMap {
-			columnSizeMap[key] += val
-		}
 	}
-	taskMeta.Result.ColSizeMap = columnSizeMap
 
 	if globalSort {
 		taskMeta.Result.LoadedRowCnt, err = getLoadedRowCountOnGlobalSort(handle, task)
@@ -662,8 +657,7 @@ func (sch *ImportSchedulerExt) finishJob(ctx context.Context, logger *zap.Logger
 		func(ctx context.Context) (bool, error) {
 			return true, taskHandle.WithNewSession(func(se sessionctx.Context) error {
 				if err := importer.FlushTableStats(ctx, se, taskMeta.Plan.TableInfo.ID, &importer.JobImportResult{
-					Affected:   taskMeta.Result.LoadedRowCnt,
-					ColSizeMap: taskMeta.Result.ColSizeMap,
+					Affected: taskMeta.Result.LoadedRowCnt,
 				}); err != nil {
 					logger.Warn("flush table stats failed", zap.Error(err))
 				}

pkg/disttask/importinto/subtask_executor.go

@@ -69,7 +69,6 @@ func (e *importMinimalTaskExecutor) Run(ctx context.Context, dataWriter, indexWr
 			sharedVars.TableImporter,
 			sharedVars.DataEngine,
 			sharedVars.IndexEngine,
-			sharedVars.Progress,
 			logger,
 			checksum,
 		); err != nil {
@@ -82,7 +81,6 @@ func (e *importMinimalTaskExecutor) Run(ctx context.Context, dataWriter, indexWr
 			sharedVars.TableImporter,
 			dataWriter,
 			indexWriter,
-			sharedVars.Progress,
 			logger,
 			checksum,
 		); err != nil {

pkg/disttask/importinto/task_executor.go

@@ -154,7 +154,6 @@ func (s *importStepExecutor) RunSubtask(ctx context.Context, subtask *proto.Subt
 		TableImporter:    s.tableImporter,
 		DataEngine:       dataEngine,
 		IndexEngine:      indexEngine,
-		Progress:         importer.NewProgress(),
 		Checksum:         verification.NewKVGroupChecksumWithKeyspace(s.tableImporter.GetKeySpace()),
 		SortedDataMeta:   &external.SortedKVMeta{},
 		SortedIndexMetas: make(map[int64]*external.SortedKVMeta),
@@ -251,7 +250,6 @@ func (s *importStepExecutor) onFinished(ctx context.Context, subtask *proto.Subt
 	}
 	subtaskMeta.Result = Result{
 		LoadedRowCnt: dataKVCount,
-		ColSizeMap:   sharedVars.Progress.GetColSize(),
 	}
 	allocators := sharedVars.TableImporter.Allocators()
 	subtaskMeta.MaxIDs = map[autoid.AllocatorType]int64{

pkg/executor/delete.go

@@ -274,7 +274,7 @@ func (e *DeleteExec) removeRow(ctx sessionctx.Context, t table.Table, h kv.Handl
 		return err
 	}
 
-	err = t.RemoveRecord(ctx.GetTableCtx(), txn, h, data, posInfo.ExtraPartialRowOption)
+	err = t.RemoveRecord(ctx.GetTableCtx(), txn, h, data, posInfo.IndexesRowLayout)
 	if err != nil {
 		return err
 	}

pkg/executor/importer/BUILD.bazel

@@ -9,7 +9,6 @@ go_library(
         "job.go",
         "kv_encode.go",
         "precheck.go",
-        "progress.go",
         "table_import.go",
     ],
     importpath = "github.com/pingcap/tidb/pkg/executor/importer",

pkg/executor/importer/engine_process.go

@@ -30,7 +30,6 @@ func ProcessChunk(
 	chunk *checkpoints.ChunkCheckpoint,
 	tableImporter *TableImporter,
 	dataEngine, indexEngine *backend.OpenedEngine,
-	progress *Progress,
 	logger *zap.Logger,
 	groupChecksum *verification.KVGroupChecksum,
 ) error {
@@ -65,7 +64,7 @@ func ProcessChunk(
 		}
 	}()
 
-	return ProcessChunkWithWriter(ctx, chunk, tableImporter, dataWriter, indexWriter, progress, logger, groupChecksum)
+	return ProcessChunkWithWriter(ctx, chunk, tableImporter, dataWriter, indexWriter, logger, groupChecksum)
 }
 
 // ProcessChunkWithWriter processes a chunk, and write kv pairs to dataWriter and indexWriter.
@@ -74,7 +73,6 @@ func ProcessChunkWithWriter(
 	chunk *checkpoints.ChunkCheckpoint,
 	tableImporter *TableImporter,
 	dataWriter, indexWriter backend.EngineWriter,
-	progress *Progress,
 	logger *zap.Logger,
 	groupChecksum *verification.KVGroupChecksum,
 ) error {
@@ -116,6 +114,5 @@ func ProcessChunkWithWriter(
 	if err != nil {
 		return err
 	}
-	progress.AddColSize(encoder.GetColumnSize())
 	return nil
 }

pkg/executor/importer/import.go

@@ -1395,9 +1395,8 @@ func getDataSourceType(p *plannercore.ImportInto) DataSourceType {
 
 // JobImportResult is the result of the job import.
 type JobImportResult struct {
-	Affected   uint64
-	Warnings   []contextutil.SQLWarn
-	ColSizeMap variable.DeltaColsMap
+	Affected uint64
+	Warnings []contextutil.SQLWarn
 }
 
 // GetMsgFromBRError get msg from BR error.

pkg/executor/importer/importer_testkit_test.go

@@ -311,11 +311,9 @@ func TestProcessChunkWith(t *testing.T) {
 		defer ti.Backend().CloseEngineMgr()
 		kvWriter := mock.NewMockEngineWriter(ctrl)
 		kvWriter.EXPECT().AppendRows(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-		progress := importer.NewProgress()
 		checksum := verify.NewKVGroupChecksumWithKeyspace(keyspace)
-		err := importer.ProcessChunkWithWriter(ctx, chunkInfo, ti, kvWriter, kvWriter, progress, zap.NewExample(), checksum)
+		err := importer.ProcessChunkWithWriter(ctx, chunkInfo, ti, kvWriter, kvWriter, zap.NewExample(), checksum)
 		require.NoError(t, err)
-		require.Len(t, progress.GetColSize(), 3)
 		checksumMap := checksum.GetInnerChecksums()
 		require.Len(t, checksumMap, 1)
 		require.Equal(t, verify.MakeKVChecksum(74, 2, 15625182175392723123), *checksumMap[verify.DataKVGroupID])
@@ -343,11 +341,9 @@ func TestProcessChunkWith(t *testing.T) {
 		ti.SetSelectedRowCh(rowsCh)
 		kvWriter := mock.NewMockEngineWriter(ctrl)
 		kvWriter.EXPECT().AppendRows(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-		progress := importer.NewProgress()
 		checksum := verify.NewKVGroupChecksumWithKeyspace(keyspace)
-		err := importer.ProcessChunkWithWriter(ctx, chunkInfo, ti, kvWriter, kvWriter, progress, zap.NewExample(), checksum)
+		err := importer.ProcessChunkWithWriter(ctx, chunkInfo, ti, kvWriter, kvWriter, zap.NewExample(), checksum)
 		require.NoError(t, err)
-		require.Len(t, progress.GetColSize(), 3)
 		checksumMap := checksum.GetInnerChecksums()
 		require.Len(t, checksumMap, 1)
 		require.Equal(t, verify.MakeKVChecksum(111, 3, 14585065391351463171), *checksumMap[verify.DataKVGroupID])

pkg/executor/importer/kv_encode.go

@@ -35,8 +35,6 @@ import (
 // KVEncoder encodes a row of data into a KV pair.
 type KVEncoder interface {
 	Encode(row []types.Datum, rowID int64) (*kv.Pairs, error)
-	// GetColumnSize returns the size of each column in the current encoder.
-	GetColumnSize() map[int64]int64
 	io.Closer
 }
 
@@ -91,10 +89,6 @@ func (en *tableKVEncoder) Encode(row []types.Datum, rowID int64) (*kv.Pairs, err
 	return en.Record2KV(record, row, rowID)
 }
 
-func (en *tableKVEncoder) GetColumnSize() map[int64]int64 {
-	return en.SessionCtx.GetColumnSize(en.TableMeta().ID)
-}
-
 // todo merge with code in load_data.go
 func (en *tableKVEncoder) parserData2TableData(parserData []types.Datum, rowID int64) ([]types.Datum, error) {
 	row := make([]types.Datum, 0, len(en.insertColumns))