Merge branch 'main' into or_expr

Author: gouhongshen

CODEOWNERS

@@ -22,19 +22,19 @@
 
 # Top-level folders
 /cgo @aunjgr
-/docs @fengttt
+/docs @fengttt @XuPeng-SH
 /etc @zhangxu19830126
-/optools @fengttt
-/pkg @fengttt
+/optools @fengttt @XuPeng-SH
+/pkg @fengttt @XuPeng-SH
 /proto @zhangxu19830126 @XuPeng-SH
-/cmd @daviszhen @zhangxu19830126 @XuPeng-SH
+/cmd @zhangxu19830126 @XuPeng-SH
 /test @aressu1985 @heni02
 /clients @XuPeng-SH
 
 # cmd
 /cmd/mo-service @zhangxu19830126
 
-/pkg/backup @daviszhen @LeftHandCold
+/pkg/backup @LeftHandCold
 
 /pkg/datasync @volgariver6 @LeftHandCold
 
@@ -49,21 +49,21 @@
 /pkg/iscp @XuPeng-SH
 
 # pkg/cnservice
-/pkg/cnservice @reusee @daviszhen @XuPeng-SH
+/pkg/cnservice @XuPeng-SH
 
 /pkg/cacheservice @volgariver6 @aptend
 
 /pkg/clusterservice @volgariver6 @aptend
 
 # pkg/common @fengttt
 /pkg/common @zhangxu19830126 @XuPeng-SH
-/pkg/common/mpool @reusee @XuPeng-SH
-/pkg/common/malloc @reusee
-/pkg/common/spool @reusee @ouyuanning @aunjgr
-/pkg/common/async @zhangxu19830126
+/pkg/common/mpool @XuPeng-SH
+/pkg/common/malloc @XuPeng-SH
+/pkg/common/spool @ouyuanning @aunjgr
+/pkg/common/async @zhangxu19830126 @XuPeng-SH
 /pkg/common/bitmap @aunjgr @XuPeng-SH
 /pkg/common/hashmap @aunjgr
-/pkg/common/moerr @xzxiong
+/pkg/common/moerr @xzxiong @XuPeng-SH
 /pkg/common/morpc @zhangxu19830126
 /pkg/common/moprobe @fengttt
 /pkg/common/stopper @zhangxu19830126
@@ -78,15 +78,15 @@
 /pkg/compress @aunjgr
 
 # pkg/config
-/pkg/config @daviszhen @fengttt
+/pkg/config @fengttt @XuPeng-SH
 
 # pkg/container
 /pkg/container @XuPeng-SH
 /pkg/container/vector @aunjgr @XuPeng-SH
-/pkg/container/pSpool @reusee @aunjgr @ouyuanning
+/pkg/container/pSpool @aunjgr @ouyuanning
 
 # pkg/defines
-/pkg/defines @daviszhen @XuPeng-SH
+/pkg/defines @XuPeng-SH
 
 # pkg/embed
 /pkg/embed @zhangxu19830126
@@ -95,22 +95,22 @@
 /pkg/tnservice @zhangxu19830126 @XuPeng-SH
 
 # pkg/fileservice
-/pkg/fileservice @reusee @fengttt
+/pkg/fileservice @fengttt @XuPeng-SH
 
-# pkg/frontend @daviszhen
-/pkg/frontend @daviszhen @XuPeng-SH
+# pkg/frontend
+/pkg/frontend @XuPeng-SH
 
 # pkg/hakeeper
 /pkg/hakeeper @zhangxu19830126 @volgariver6
 
 # pkg/incrservice
-/pkg/incrservice @zhangxu19830126
+/pkg/incrservice @zhangxu19830126 @XuPeng-SH
 
 # pkg/lockservice
-/pkg/lockservice @zhangxu19830126
+/pkg/lockservice @zhangxu19830126 @iamlinjunhong
 
 # pkg/shardservice
-/pkg/shardservice @zhangxu19830126
+/pkg/shardservice @zhangxu19830126 @iamlinjunhong
 
 # pkg/partitionservice
 /pkg/partitionservice @zhangxu19830126 @iamlinjunhong
@@ -119,7 +119,7 @@
 /pkg/logservice @zhangxu19830126 @volgariver6
 
 # pkg/logutil
-/pkg/logutil @daviszhen @XuPeng-SH
+/pkg/logutil @XuPeng-SH
 
 # pkg/objectio
 /pkg/objectio @LeftHandCold @XuPeng-SH
@@ -131,21 +131,21 @@
 # for compatibility issues required for rolling upgrade.
 /pkg/pb @zhangxu19830126 @XuPeng-SH
 /pkg/pb/logservice @zhangxu19830126
-/pkg/pb/metadata @zhangxu19830126
+/pkg/pb/metadata @zhangxu19830126 @XuPeng-SH
 /pkg/pb/metric @aptend
 /pkg/pb/pipeline @ouyuanning @aunjgr
 /pkg/pb/plan @ouyuanning @aunjgr
 /pkg/pb/timestamp @zhangxu19830126
 /pkg/pb/txn @zhangxu19830126
 
 # pkg/perfcounter
-/pkg/perfcounter @reusee
+/pkg/perfcounter @XuPeng-SH
 
 # pkg/sort
 /pkg/sort @aunjgr
 
 # pkg/bootstrap
-/pkg/bootstrap @daviszhen @zhangxu19830126 @LeftHandCold
+/pkg/bootstrap @zhangxu19830126 @LeftHandCold
 
 # pkg/sql
 /pkg/sql @aunjgr
@@ -156,7 +156,7 @@
 /pkg/sql/plan @ouyuanning @aunjgr
 /pkg/sql/plan/function @ouyuanning @aunjgr
 /pkg/sql/plan/explain @ouyuanning @aunjgr
-/pkg/sql/plan/tools @daviszhen @gouhongshen
+/pkg/sql/plan/tools @gouhongshen
 /pkg/sql/colexec/indexjoin @aunjgr
 /pkg/sql/colexec/indexbuild @aunjgr
 /pkg/sql/colexec/intersect @aunjgr
@@ -182,19 +182,19 @@
 /pkg/sql/colexec/multi_update @ouyuanning
 
 # pkg/taskservice
-/pkg/taskservice @zhangxu19830126
+/pkg/taskservice @zhangxu19830126 @XuPeng-SH
 
 # pkg/tests
 /pkg/tests @zhangxu19830126
 /pkg/tests/service @zhangxu19830126
 /pkg/tests/txn @zhangxu19830126
 
 # pkg/testutil
-/pkg/testutil @ouyuanning @daviszhen @XuPeng-SH
+/pkg/testutil @ouyuanning @XuPeng-SH
 
 # pkg/txn
 /pkg/txn @zhangxu19830126
-/pkg/txn/storage/memorystorage @reusee
+/pkg/txn/storage/memorystorage @XuPeng-SH
 
 # pkg/util
 /pkg/util @zhangxu19830126 @XuPeng-SH
@@ -213,12 +213,12 @@
 /pkg/vm @XuPeng-SH
 /pkg/vm/engine @XuPeng-SH
 /pkg/vm/pipeline @ouyuanning @aunjgr
-/pkg/vm/process @reusee @aunjgr @XuPeng-SH
+/pkg/vm/process @aunjgr @XuPeng-SH
 /pkg/vm/message @aunjgr
 /pkg/vm/engine/disttae @XuPeng-SH @gouhongshen
 /pkg/vm/engine/disttae/logtailreplay @XuPeng-SH @gouhongshen
 /pkg/vm/engine/tae @XuPeng-SH
-/pkg/vm/engine/memoryengine @reusee
+/pkg/vm/engine/memoryengine @XuPeng-SH
 /pkg/vm/engine/tae/logtail/service @XuPeng-SH @volgariver6
 /pkg/vm/engine/tae/logstore @XuPeng-SH @jiangxinmeng1
 /pkg/vm/engine/tae/wal @XuPeng-SH @jiangxinmeng1

etc/launch/cn.toml

@@ -8,6 +8,15 @@ level = "info"
 uuid = "dd1dccb4-4d3c-41f8-b482-5251dc7a41bf"
 port-base = 18000
 
+[cn.Engine]
+# only prefetch the matched db.table
+prefetch-on-subscribed = [
+    '^mo_catalog\.mo_tables$',
+    '^mysql\..*$',
+    '^test1\..*$',
+    '^test2\.t1$',
+]
+
 [malloc]
 check-fraction = 65536
 enable-metrics = true

pkg/cnservice/distributed_tae.go

@@ -79,7 +79,11 @@ func (s *service) initDistributedTAE(
 		s.cfg.LogtailUpdateWorkerFactor,
 
 		disttae.WithCNTransferTxnLifespanThreshold(
-			s.cfg.Engine.CNTransferTxnLifespanThreshold),
+			s.cfg.Engine.CNTransferTxnLifespanThreshold,
+		),
+		disttae.WithPrefetchOnSubscribed(
+			s.cfg.Engine.PrefetchOnSubscribed,
+		),
 		disttae.WithMoTableStatsConf(s.cfg.Engine.Stats),
 		disttae.WithSQLExecFunc(internalExecutorFactory),
 		disttae.WithMoServerStateChecker(func() bool {

pkg/cnservice/server_query.go

@@ -100,6 +100,7 @@ func (s *service) initQueryCommandHandler() {
 	s.queryService.AddHandleFunc(query.CmdMethod_CtlMoTableStats, s.handleMoTableStats, false)
 	s.queryService.AddHandleFunc(query.CmdMethod_WorkspaceThreshold, s.handleWorkspaceThresholdRequest, false)
 	s.queryService.AddHandleFunc(query.CmdMethod_MinTimestamp, s.handleGetMinTimestamp, false)
+	s.queryService.AddHandleFunc(query.CmdMethod_CtlPrefetchOnSubscribed, s.handleCtlPrefetchOnSubscribed, false)
 }
 
 func (s *service) handleKillConn(ctx context.Context, req *query.Request, resp *query.Response, _ *morpc.Buffer) error {
@@ -179,6 +180,17 @@ func (s *service) handleCtlReader(ctx context.Context, req *query.Request, resp
 	return nil
 }
 
+func (s *service) handleCtlPrefetchOnSubscribed(ctx context.Context, req *query.Request, resp *query.Response, _ *morpc.Buffer) error {
+	if req == nil || req.CtlPrefetchOnSubscribedRequest == nil {
+		return moerr.NewInternalError(ctx, "bad request")
+	}
+	resp.CtlPrefetchOnSubscribedResponse = new(query.CtlPrefetchOnSubscribedResponse)
+	resp.CtlPrefetchOnSubscribedResponse.Resp = ctl.UpdateCurrentCNPrefetchOnSubscribed(
+		req.CtlPrefetchOnSubscribedRequest.Patterns,
+	)
+	return nil
+}
+
 // handleGetLockInfo sends the lock info on current cn to another cn that needs.
 func (s *service) handleGetLockInfo(ctx context.Context, req *query.Request, resp *query.Response, _ *morpc.Buffer) error {
 	resp.GetLockInfoResponse = new(query.GetLockInfoResponse)
@@ -408,7 +420,7 @@ func (s *service) handleUnsubscribeTable(ctx context.Context, req *query.Request
 	if req.UnsubscribeTable == nil {
 		return moerr.NewInternalError(ctx, "bad request")
 	}
-	err := s.storeEngine.UnsubscribeTable(ctx, req.UnsubscribeTable.DatabaseID, req.UnsubscribeTable.TableID)
+	err := s.storeEngine.UnsubscribeTable(ctx, 0, req.UnsubscribeTable.DatabaseID, req.UnsubscribeTable.TableID)
 	if err != nil {
 		resp.WrapError(err)
 		return nil

pkg/cnservice/server_query_test.go

@@ -518,10 +518,10 @@ func Test_service_handleUnsubscribeTable(t *testing.T) {
 	err := dummyErr
 	ctl := gomock.NewController(t)
 	mockEng := mock_frontend.NewMockEngine(ctl)
-	mockEng.EXPECT().UnsubscribeTable(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockEng.EXPECT().UnsubscribeTable(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	mockEngErr := mock_frontend.NewMockEngine(ctl)
-	mockEngErr.EXPECT().UnsubscribeTable(gomock.Any(), gomock.Any(), gomock.Any()).Return(err).AnyTimes()
+	mockEngErr.EXPECT().UnsubscribeTable(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(err).AnyTimes()
 
 	respWithErr := &query.Response{}
 	respWithErr.WrapError(err)
@@ -759,6 +759,24 @@ func Test_service_handleCtlReader(t *testing.T) {
 	}
 }
 
+func Test_service_handleCtlPrefetchOnSubscribed(t *testing.T) {
+	ctx := context.Background()
+
+	req := &query.Request{CtlPrefetchOnSubscribedRequest: &query.CtlPrefetchOnSubscribedRequest{
+		Patterns: []string{"^foo$"},
+	}}
+	resp := &query.Response{}
+	t.Cleanup(func() {
+		require.NoError(t, engine.SetPrefetchOnSubscribed(nil))
+	})
+	s := &service{}
+	err := s.handleCtlPrefetchOnSubscribed(ctx, req, resp, nil)
+	require.NoError(t, err)
+	require.Equal(t, &query.Response{CtlPrefetchOnSubscribedResponse: &query.CtlPrefetchOnSubscribedResponse{
+		Resp: "prefetch_on_subscribed updated, patterns: 1",
+	}}, resp)
+}
+
 func Test_service_handleRunTask(t *testing.T) {
 
 	ctx := context.Background()

pkg/cnservice/types.go

@@ -130,6 +130,13 @@ type Config struct {
 	Engine struct {
 		Type EngineType `toml:"type"`
 
+		// only prefetch the matched dbname.tablename
+		//	'^mo_catalog\.mo_tables$',
+		//	'^mysql\..*$',
+		//	'^test1\..*$',
+		//	'^test2\.t1$'
+		PrefetchOnSubscribed []string `toml:"prefetch-on-subscribed"`
+
 		MoTableStatsUseOldImpl         bool          `toml:"mo-table-stats-use-old-impl"`
 		CNTransferTxnLifespanThreshold time.Duration `toml:"cn-transfer-txn-lifespan-threshold"`
 

pkg/fileservice/local_fs.go

@@ -389,11 +389,31 @@ read_disk_cache:
 	if l.diskCache != nil {
 
 		t0 := time.Now()
+		LogEvent(ctx, str_read_disk_cache_Caches_begin)
+		// Record which entries are not done before reading from disk cache
+		undoneBefore := make(map[int]bool)
+		for i, entry := range vector.Entries {
+			undoneBefore[i] = !entry.done
+		}
 		err := readCache(ctx, l.diskCache, vector)
+		LogEvent(ctx, str_read_disk_cache_Caches_end)
 		metric.FSReadDurationReadDiskCache.Observe(time.Since(t0).Seconds())
 		if err != nil {
 			return err
 		}
+		// Count bytes actually read from disk cache (entries that became done and from disk cache)
+		var actualDiskReadBytes int64
+		for i, entry := range vector.Entries {
+			if undoneBefore[i] && entry.done && entry.fromCache == l.diskCache {
+				actualDiskReadBytes += entry.Size
+			}
+		}
+		// Record disk read size
+		if actualDiskReadBytes > 0 {
+			perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) {
+				counter.FileService.DiskReadSize.Add(actualDiskReadBytes)
+			})
+		}
 		if vector.allDone() {
 			return nil
 		}
@@ -440,10 +460,23 @@ read_disk_cache:
 		}
 	}
 
+	// Count bytes that will be read from local disk (entries that are not done yet)
+	var localDiskReadBytes int64
+	for _, entry := range vector.Entries {
+		if !entry.done {
+			localDiskReadBytes += entry.Size
+		}
+	}
 	err = l.read(ctx, vector, bytesCounter)
 	if err != nil {
 		return err
 	}
+	// Record disk read size (all bytes read from local disk)
+	if localDiskReadBytes > 0 {
+		perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) {
+			counter.FileService.DiskReadSize.Add(localDiskReadBytes)
+		})
+	}
 
 	return nil
 }