refactor: consolidate unused code and bookkeeping cleanup

pkg/catalog/secondary_index_utils.go

@@ -90,21 +90,6 @@ func IsFullText2IndexAlgo(algo string) bool {
 	return _algo == MoIndexFullText2Algo.ToString()
 }
 
-func IsHnswIndexAlgo(algo string) bool {
-	_algo := ToLower(algo)
-	return _algo == MoIndexHnswAlgo.ToString()
-}
-
-func IsCagraIndexAlgo(algo string) bool {
-	_algo := ToLower(algo)
-	return _algo == MoIndexCagraAlgo.ToString()
-}
-
-func IsIvfpqIndexAlgo(algo string) bool {
-	_algo := ToLower(algo)
-	return _algo == MoIndexIvfpqAlgo.ToString()
-}
-
 // ------------------------[START] IndexAlgoParams------------------------
 const (
 	IndexAlgoParamLists     = "lists"
@@ -621,13 +606,6 @@ func indexParamsToMap(def interface{}) (map[string]string, error) {
 	return res, moerr.NewInternalErrorNoCtx("indexParamsToMap: invalid index type")
 }
 
-func DefaultIvfIndexAlgoOptions() map[string]string {
-	res := make(map[string]string)
-	res[IndexAlgoParamLists] = "1"                       // set lists = 1 as default
-	res[IndexAlgoParamOpType] = metric.OpType_L2Distance // set l2 as default
-	return res
-}
-
 // IndexAlgoParamVersionOf returns the engine version recorded in an index's
 // algo_params (currently only fulltext uses it: v2 = WAND engine). Absent or
 // unparseable ⇒ 1 (the classic engine), so pre-version indexes read as v1.

pkg/cdc/observability.go

@@ -75,7 +75,6 @@ type ProgressTracker struct {
 	totalSQLExecuted      atomic.Uint64 // Total SQL statements executed
 
 	// Timing metrics
-	lastRoundStartTime  time.Time
 	lastRoundDuration   time.Duration
 	avgRoundDuration    time.Duration
 	maxRoundDuration    time.Duration
@@ -241,7 +240,6 @@ func (pt *ProgressTracker) EndRound(success bool, err error) {
 
 	duration := time.Since(pt.currentRoundStartTime)
 	pt.lastRoundDuration = duration
-	pt.lastRoundStartTime = pt.currentRoundStartTime
 	pt.totalProcessingTime += duration
 
 	// Update average and max duration

pkg/cnservice/cnclient/client.go

@@ -56,9 +56,7 @@ func GetPipelineClient(
 }
 
 type pipelineClient struct {
-	localServiceAddress string
-	config              *PipelineConfig
-	client              morpc.RPCClient
+	client morpc.RPCClient
 }
 
 func pipelineBackendCreateOptions(cfg *PipelineConfig) []morpc.ClientOption {
@@ -83,10 +81,7 @@ func NewPipelineClient(
 
 	cfg.fill()
 
-	c := &pipelineClient{
-		localServiceAddress: localServiceAddress,
-		config:              cfg,
-	}
+	c := &pipelineClient{}
 
 	codec := morpc.NewMessageCodec(
 		sid,

pkg/cnservice/cnclient/client_test.go

@@ -49,8 +49,7 @@ func (c *testRPCClient) CloseBackend() error                            { return
 func TestPipelineClient_NewStreamAllowsLocalBackend(t *testing.T) {
 	rpcClient := &testRPCClient{}
 	client := &pipelineClient{
-		localServiceAddress: "127.0.0.1:1234",
-		client:              rpcClient,
+		client: rpcClient,
 	}
 
 	_, err := client.NewStream(context.Background(), "127.0.0.1:1234")

pkg/common/morpc/backend.go

@@ -201,7 +201,6 @@ type remoteBackend struct {
 	cancel          context.CancelFunc
 	cancelOnce      sync.Once
 	pingTimer       *time.Timer
-	lastPingTime    time.Time
 	livenessEpoch   time.Time
 
 	options struct {
@@ -904,7 +903,6 @@ func (rb *remoteBackend) fetch(messages []*Future, maxFetchCount int) ([]*Future
 	messages = messages[:0]
 
 	doHeartbeat := func() {
-		rb.lastPingTime = time.Now()
 		f := rb.getFuture(context.TODO(), &flagOnlyMessage{flag: flagPing}, true)
 		// no need wait response, close immediately
 		f.Close()

pkg/common/morpc/client_gc.go

@@ -336,21 +336,6 @@ func init() {
 	globalClientGC.mu.Unlock()
 }
 
-// ResetGlobalClientGCManagerForTest resets the global GC manager for testing purposes.
-// This stops the current manager and creates a new one.
-// WARNING: Only use this in tests!
-func ResetGlobalClientGCManagerForTest() {
-	globalClientGCMu.Lock()
-	defer globalClientGCMu.Unlock()
-	mgr := globalClientGC
-	mgr.stop()
-	replacement := newClientGCManager()
-	replacement.mu.Lock()
-	replacement.ensureStartedLocked()
-	replacement.mu.Unlock()
-	globalClientGC = replacement
-}
-
 // triggerGCInactive triggers GC inactive task for a specific client and backend.
 // If channel is full, the request is dropped to avoid blocking.
 func (m *clientGCManager) triggerGCInactive(c *client, remote string) {

pkg/container/batch/batch.go

@@ -122,18 +122,6 @@ func NewWithSchema(offHeap bool, attrs []string, attTypes []types.Type) *Batch {
 	return bat
 }
 
-func EmptyBatchWithAttrs(attrs []string) Batch {
-	bat := Batch{
-		Attrs: attrs,
-		Vecs:  make([]*vector.Vector, len(attrs)),
-	}
-	for i := range attrs {
-		bat.Vecs[i] = vector.NewVec(types.T_any.ToType())
-	}
-
-	return bat
-}
-
 func SetLength(bat *Batch, n int) {
 	for _, vec := range bat.Vecs {
 		vec.SetLength(n)

pkg/container/vector/vector.go

@@ -3976,15 +3976,6 @@ func NewOffHeapVecWithType(typ types.Type) *Vector {
 	return vec
 }
 
-func NewOffHeapVecWithTypeAndData(typ types.Type, data []byte, length, cap int) *Vector {
-	vec := NewVec(typ)
-	vec.offHeap = true
-	vec.data = data
-	vec.length = length
-	vec.areaDisjoint = !typ.IsVarlen() || length == 0
-	return vec
-}
-
 func NewOffHeapVec() *Vector {
 	vec := NewVecFromReuse()
 	vec.offHeap = true

pkg/frontend/databranchutils/branch_hashmap.go

@@ -2522,7 +2522,6 @@ type spillStats struct {
 	spilledEntries      uint64
 	spilledPayloadBytes uint64
 	spilledBytes        uint64
-	segmentsCreated     uint64
 
 	bucketScans       uint64
 	bucketScanBytes   uint64
@@ -2857,7 +2856,6 @@ func (ss *spillStore) createSegment(bucket *spillBucket, bucketID uint32) (*spil
 	}
 	seg := &spillSegment{path: path}
 	bucket.segments = append(bucket.segments, seg)
-	ss.stats.segmentsCreated++
 	return seg, file, bufio.NewWriter(file), nil
 }
 

pkg/fulltext/plugin/search/scan.go

@@ -59,8 +59,6 @@ type Scan struct {
 	n_result         uint64
 	sacc             *fulltext.SearchAccum
 	limit            uint64
-	nrows            int
-	idx2word         map[int]string
 	agghtab          map[any]uint64
 	aggcnt           []int64
 	mpool            *fulltext.FixedBytePool
@@ -155,8 +153,6 @@ func (u *Scan) ResetRowState(proc *process.Process) {
 	u.streamingStarted = false
 	u.n_result = 0
 	u.sacc = nil
-	u.nrows = 0
-	u.idx2word = make(map[int]string)
 	u.agghtab = nil
 	u.aggcnt = nil
 	u.docLenMap = make(map[any]int32)
@@ -763,8 +759,6 @@ func groupby(u *Scan, proc *process.Process, s *fulltext.SearchAccum) (stream_cl
 	}
 	needSetDocLen := len(bat.Vecs) == 3
 
-	u.nrows += bat.RowCount()
-
 	for i := 0; i < bat.RowCount(); i++ {
 		// doc_id any
 		doc_id := u.normalizeDocID(vector.GetAny(bat.Vecs[0], i, false))

pkg/fulltext2/tailbuild.go

@@ -72,7 +72,6 @@ type TailBuilder struct {
 	tokenize   func(string) []WordPos
 	dir        string
 	seq        int
-	delSeq     int
 	cur        *Builder       // open INSERT segment (nil between insert runs)
 	delBatch   []DeleteRecord // open DELETE run (sealed on an op switch to insert, at capacity, or at Finish)
 	frames     []TailSegment  // sealed frames (insert segs + delete frames) in chronological (recency) order
@@ -246,7 +245,6 @@ func (t *TailBuilder) sealDeletes() error {
 		return err
 	}
 	t.delBatch = nil // free the run; the next delete re-grows a fresh slice
-	t.delSeq++
 	path, off, err := t.writeFrame(framed)
 	if err != nil {
 		return err

pkg/hakeeper/checkers/logservice/parse_test.go

@@ -386,7 +386,7 @@ func TestCollectStats(t *testing.T) {
 						Replicas: []pb.LogReplicaInfo{}},
 				},
 			},
-			expectedNormal: &stats{toStart: []replica{{"c", 1, 0, 3}},
+			expectedNormal: &stats{toStart: []replica{{uuid: "c", shardID: 1, replicaID: 3}},
 				toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}},
 			expectedNonVoting: &stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}, nonVoting: true},
 		},
@@ -440,7 +440,7 @@ func TestCollectStats(t *testing.T) {
 						}}},
 				},
 			},
-			expectedNormal: &stats{zombies: []replica{{"d", 1, 0, 0}},
+			expectedNormal: &stats{zombies: []replica{{uuid: "d", shardID: 1, replicaID: 0}},
 				toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}},
 			expectedNonVoting: &stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}, nonVoting: true},
 		},
@@ -541,7 +541,7 @@ func TestCollectStats(t *testing.T) {
 								Term:     0}}}},
 				},
 			},
-			expectedNormal: &stats{toRemove: map[uint64][]replica{1: {{"a", 1, 0, 1}}},
+			expectedNormal: &stats{toRemove: map[uint64][]replica{1: {{uuid: "a", shardID: 1, replicaID: 1}}},
 				toAdd: map[uint64]uint32{},
 			},
 			expectedNonVoting: &stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}, nonVoting: true},
@@ -899,7 +899,7 @@ func TestCollectStats(t *testing.T) {
 				},
 			},
 			expectedNormal: &stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}},
-			expectedNonVoting: &stats{toStart: []replica{{"f", 1, 0, 6}},
+			expectedNonVoting: &stats{toStart: []replica{{uuid: "f", shardID: 1, replicaID: 6}},
 				toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}, nonVoting: true},
 		},
 		{
@@ -993,7 +993,7 @@ func TestCollectStats(t *testing.T) {
 							}}}},
 				},
 			},
-			expectedNormal: &stats{zombies: []replica{{"g", 1, 0, 0}},
+			expectedNormal: &stats{zombies: []replica{{uuid: "g", shardID: 1, replicaID: 0}},
 				toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}},
 			expectedNonVoting: &stats{toRemove: map[uint64][]replica{}, toAdd: map[uint64]uint32{}, nonVoting: true},
 		},

pkg/hakeeper/checkers/logservice/stats.go

@@ -18,7 +18,6 @@ type replica struct {
 	// uuid is the store UUID
 	uuid    string
 	shardID uint64
-	epoch   uint64
 
 	replicaID uint64
 }

pkg/hakeeper/operator/controller.go

@@ -124,29 +124,6 @@ func (c *Controller) GetNonVotingExecutingReplicas() ExecutingReplicas {
 	return executing
 }
 
-func (c *Controller) GetExecutingNonVotingReplicas() ExecutingReplicas {
-	executing := ExecutingReplicas{
-		Adding:   make(map[uint64][]uint64),
-		Removing: make(map[uint64][]uint64),
-		Starting: make(map[uint64][]uint64),
-	}
-	for shardID, operators := range c.operators {
-		for _, op := range operators {
-			for _, step := range op.steps {
-				switch step := step.(type) {
-				case RemoveLogService:
-					executing.Removing[shardID] = append(executing.Removing[shardID], step.ReplicaID)
-				case AddLogService:
-					executing.Adding[shardID] = append(executing.Adding[shardID], step.ReplicaID)
-				case StartLogService:
-					executing.Starting[shardID] = append(executing.Starting[shardID], step.ReplicaID)
-				}
-			}
-		}
-	}
-	return executing
-}
-
 func (c *Controller) RemoveFinishedOperator(
 	logState pb.LogState, tnState pb.TNState, cnState pb.CNState, proxyState pb.ProxyState,
 ) {

pkg/logutil/dragonboat.go

@@ -23,9 +23,8 @@ import (
 var _ logger.ILogger = (*DragonboatAdaptLogger)(nil)
 
 type DragonboatAdaptLogger struct {
-	logger  *zap.SugaredLogger
-	atom    *zap.AtomicLevel
-	pkgName string
+	logger *zap.SugaredLogger
+	atom   *zap.AtomicLevel
 }
 
 func (d *DragonboatAdaptLogger) SetLevel(level logger.LogLevel) {
@@ -78,9 +77,8 @@ func DragonboatFactory(name string) logger.ILogger {
 	options := cfg.getOptions()
 	options = append(options, zap.AddCallerSkip(2))
 	return &DragonboatAdaptLogger{
-		logger:  zap.New(zapcore.NewTee(cores...), options...).Named(name).Sugar(),
-		atom:    &atom,
-		pkgName: name,
+		logger: zap.New(zapcore.NewTee(cores...), options...).Named(name).Sugar(),
+		atom:   &atom,
 	}
 }
 

pkg/sql/colexec/mongoscan/mongoscan.go

@@ -185,7 +185,6 @@ func (scan *MongoScan) Prepare(proc *process.Process) (err error) {
 		return moerr.NewInternalErrorf(proc.Ctx, "MongoDB %s failed", operation)
 	}
 	metric.MongoDBCursorEventCounter.WithLabelValues("open").Inc()
-	scan.ctr.generation++
 	scan.ctr.done = false
 	scan.ctr.rows = 0
 	scan.ctr.rawBytes = 0

pkg/sql/colexec/mongoscan/types.go

@@ -43,7 +43,6 @@ type container struct {
 	done          bool
 	rows          int64
 	rawBytes      int64
-	generation    uint64
 	pendingRaw    []byte
 	pendingBudget *process.ExecutionResourceGeneration
 	pendingCharge uint64

pkg/tests/testutils/test_utils.go

@@ -392,14 +392,6 @@ func GetSQLExecutor(
 	return cn.RawService().(cnservice.Service).GetSQLExecutor()
 }
 
-func DBExists(
-	t *testing.T,
-	name string,
-	cn embed.ServiceOperator,
-) bool {
-	return DBExistsWithAccount(t, 0, name, cn)
-}
-
 func DBExistsWithAccount(
 	t *testing.T,
 	account int32,

pkg/tools/interactive/generic_page.go

@@ -117,10 +117,6 @@ type GenericPage struct {
 	inputMode   string // "", "search", "filter"
 	inputBuffer string
 	inputCursor int // Cursor position in input buffer
-
-	// Screen size
-	width  int
-	height int
 }
 
 // NewGenericPage creates a page with config
@@ -140,8 +136,6 @@ func NewGenericPage(config PageConfig, provider DataProvider, handler ActionHand
 		provider: provider,
 		handler:  handler,
 		list:     NewGenericList(listOpts),
-		width:    120,
-		height:   30,
 	}
 
 	// Set row number label if provided
@@ -169,8 +163,6 @@ func (p *GenericPage) Refresh() {
 
 // SetSize sets terminal size
 func (p *GenericPage) SetSize(w, h int) {
-	p.width = w
-	p.height = h
 	// Reserve: title(1) + overview(1) + separator(1) + hints(2) + status(1) = 6
 	p.list.SetScreenHeight(h - 6)
 	p.list.SetScreenWidth(w)

pkg/util/batchpipe/batch_pipe.go

@@ -185,7 +185,6 @@ type BaseBatchPipe[T HasName, B any] struct {
 	impl              PipeImpl[T, B]
 	isRunning         int32
 	opts              baseBatchPipeOpts
-	dropped           int64
 	itemCh            chan T
 	batchCh           chan B
 	sendingLock       sync.RWMutex
@@ -220,7 +219,6 @@ func (bc *BaseBatchPipe[T, B]) SendItem(ctx context.Context, items ...T) error {
 		select {
 		case bc.itemCh <- item:
 		default:
-			atomic.AddInt64(&bc.dropped, 1)
 		}
 	}
 	return nil