refactor mo_table_size/rows.

pkg/bootstrap/versions/v2_0_1/clusetr_upgrade_list.go → pkg/bootstrap/versions/v2_0_1/cluster_upgrade_list.go

@@ -16,20 +16,31 @@ package v2_0_1
 
 import (
 	"fmt"
-
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions"
 	"github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/frontend"
 	"github.com/matrixorigin/matrixone/pkg/util/executor"
 )
 
 var needMigrateMoPubs = false
 
 var clusterUpgEntries = []versions.UpgradeEntry{
+	upg_mo_table_stats,
 	upg_mo_pubs_add_account_id_column,
 	upg_mo_cdc_task,
 	upg_mo_cdc_watermark,
 }
 
+var upg_mo_table_stats = versions.UpgradeEntry{
+	Schema:    catalog.MO_CATALOG,
+	TableName: catalog.MO_TABLE_STATS,
+	UpgType:   versions.CREATE_NEW_TABLE,
+	UpgSql:    frontend.MoCatalogMoTableStatsDDL,
+	CheckFunc: func(txn executor.TxnExecutor, accountId uint32) (bool, error) {
+		return versions.CheckTableDefinition(txn, accountId, catalog.MO_CATALOG, catalog.MO_TABLE_STATS)
+	},
+}
+
 var upg_mo_pubs_add_account_id_column = versions.UpgradeEntry{
 	Schema:    catalog.MO_CATALOG,
 	TableName: catalog.MO_PUBS,

pkg/catalog/types.go

@@ -164,6 +164,8 @@ const (
 	MO_CDC_WATERMARK = "mo_cdc_watermark"
 
 	MO_DATA_KEY = "mo_data_key"
+
+	MO_TABLE_STATS = "mo_table_stats_alpha"
 )
 
 func IsSystemTable(id uint64) bool {

pkg/cnservice/distributed_tae.go

@@ -22,7 +22,9 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/config"
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
+	"github.com/matrixorigin/matrixone/pkg/frontend"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
+	ie "github.com/matrixorigin/matrixone/pkg/util/internalExecutor"
 	"github.com/matrixorigin/matrixone/pkg/util/status"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/blockio"
@@ -56,6 +58,10 @@ func (s *service) initDistributedTAE(
 	// start I/O pipeline
 	blockio.Start(s.cfg.UUID)
 
+	internalExecutorFactory := func() ie.InternalExecutor {
+		return frontend.NewInternalExecutor(s.cfg.UUID)
+	}
+
 	// engine
 	distributeTaeMp, err := mpool.NewMPool("distributed_tae", 0, mpool.NoFixed)
 	if err != nil {
@@ -74,6 +80,8 @@ func (s *service) initDistributedTAE(
 
 		disttae.WithCNTransferTxnLifespanThreshold(
 			s.cfg.Engine.CNTransferTxnLifespanThreshold),
+		disttae.WithMoTableStats(s.cfg.Engine.Stats),
+		disttae.WithSQLExecFunc(internalExecutorFactory),
 	)
 	pu.StorageEngine = s.storeEngine
 

pkg/cnservice/server_query.go

@@ -19,8 +19,6 @@ import (
 	"runtime/debug"
 	"strings"
 
-	"go.uber.org/zap"
-
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/common/system"
@@ -40,6 +38,8 @@ import (
 	qclient "github.com/matrixorigin/matrixone/pkg/queryservice/client"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan/function/ctl"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
+	"go.uber.org/zap"
 )
 
 func (s *service) initQueryService() error {
@@ -96,6 +96,7 @@ func (s *service) initQueryCommandHandler() {
 	s.queryService.AddHandleFunc(query.CmdMethod_FileServiceCacheEvict, s.handleFileServiceCacheEvictRequest, false)
 	s.queryService.AddHandleFunc(query.CmdMethod_MetadataCache, s.handleMetadataCacheRequest, false)
 	s.queryService.AddHandleFunc(query.CmdMethod_FaultInjection, s.handleFaultInjection, false)
+	s.queryService.AddHandleFunc(query.CmdMethod_CtlMoTableStats, s.handleMoTableStats, false)
 }
 
 func (s *service) handleKillConn(ctx context.Context, req *query.Request, resp *query.Response, _ *morpc.Buffer) error {
@@ -156,6 +157,14 @@ func (s *service) handleFaultInjection(ctx context.Context, req *query.Request,
 	return nil
 }
 
+func (s *service) handleMoTableStats(ctx context.Context, req *query.Request, resp *query.Response, _ *morpc.Buffer) error {
+	ret := disttae.HandleMoTableStatsCtl(req.CtlMoTableStatsRequest.Cmd)
+	resp.CtlMoTableStatsResponse = query.CtlMoTableStatsResponse{
+		Resp: ret,
+	}
+	return nil
+}
+
 func (s *service) handleCtlReader(ctx context.Context, req *query.Request, resp *query.Response, _ *morpc.Buffer) error {
 	resp.CtlReaderResponse = new(query.CtlReaderResponse)
 

pkg/cnservice/server_task.go

@@ -20,8 +20,6 @@ import (
 	"strings"
 	"time"
 
-	"go.uber.org/zap"
-
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
@@ -40,6 +38,8 @@ import (
 	db_holder "github.com/matrixorigin/matrixone/pkg/util/export/etl/db"
 	ie "github.com/matrixorigin/matrixone/pkg/util/internalExecutor"
 	"github.com/matrixorigin/matrixone/pkg/util/metric/mometric"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
+	"go.uber.org/zap"
 )
 
 func (s *service) adjustSQLAddress() {
@@ -277,6 +277,11 @@ func (s *service) registerExecutorsLocked() {
 			export.WithFileService(s.etlFS),
 		),
 	)
+	// init mo table stats task
+	s.task.runner.RegisterExecutor(
+		task.TaskCode_MOTableStats,
+		disttae.GetMOTableStatsExecutor(s.cfg.UUID, s.storeEngine, ieFactory))
+
 	// init metric task
 	s.task.runner.RegisterExecutor(
 		task.TaskCode_MetricStorageUsage,

pkg/cnservice/types.go

@@ -22,8 +22,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"go.uber.org/zap"
-
 	"github.com/matrixorigin/matrixone/pkg/bootstrap"
 	"github.com/matrixorigin/matrixone/pkg/clusterservice"
 	"github.com/matrixorigin/matrixone/pkg/cnservice/cnclient"
@@ -57,7 +55,9 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/util/executor"
 	"github.com/matrixorigin/matrixone/pkg/util/toml"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/options"
+	"go.uber.org/zap"
 )
 
 var (
@@ -130,7 +130,10 @@ type Config struct {
 		Type     EngineType           `toml:"type"`
 		Logstore options.LogstoreType `toml:"logstore"`
 
+		MoTableStatsUseOldImpl         bool          `toml:"mo-table-stats-use-old-impl"`
 		CNTransferTxnLifespanThreshold time.Duration `toml:"cn-transfer-txn-lifespan-threshold"`
+
+		Stats disttae.MoTableStatsConfig `toml:"stats"`
 	}
 
 	// parameters for cn-server related buffer.

pkg/frontend/authenticate.go

@@ -927,6 +927,7 @@ var (
 		"mo_snapshots":                0,
 		"mo_cdc_task":                 0,
 		"mo_cdc_watermark":            0,
+		catalog.MO_TABLE_STATS:        0,
 	}
 	sysAccountTables = map[string]struct{}{
 		catalog.MOVersionTable:       {},
@@ -967,6 +968,7 @@ var (
 		catalog.MO_RETENTION:          0,
 		"mo_cdc_task":                 0,
 		"mo_cdc_watermark":            0,
+		catalog.MO_TABLE_STATS:        0,
 	}
 	createDbInformationSchemaSql = "create database information_schema;"
 	createAutoTableSql           = MoCatalogMoAutoIncrTableDDL
@@ -1004,6 +1006,7 @@ var (
 		MoCatalogMoCdcTaskDDL,
 		MoCatalogMoCdcWatermarkDDL,
 		MoCatalogMoDataKeyDDL,
+		MoCatalogMoTableStatsDDL,
 	}
 
 	//drop tables for the tenant
@@ -7504,6 +7507,9 @@ func createTablesInMoCatalogOfGeneralTenant2(bh BackgroundExec, ca *createAccoun
 		if strings.HasPrefix(sql, "create table mo_catalog.mo_data_key") {
 			return true
 		}
+		if strings.HasPrefix(sql, fmt.Sprintf("create table mo_catalog.%s", catalog.MO_TABLE_STATS)) {
+			return true
+		}
 		return false
 	}
 

pkg/frontend/func_mo_test.go

@@ -0,0 +1,69 @@
+// Copyright 2024 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package frontend
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/matrixorigin/matrixone/pkg/sql/plan/function"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetVariables(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	ctx := context.Background()
+
+	ses := newSes(nil, ctrl)
+	defer ses.Close()
+
+	tempExecCtx := ExecCtx{
+		reqCtx: ctx,
+		ses:    ses,
+	}
+	defer tempExecCtx.Close()
+	ses.txnCompileCtx.SetExecCtx(&tempExecCtx)
+
+	proc := testutil.NewProc()
+	proc.SetResolveVariableFunc(ses.txnCompileCtx.ResolveVariable)
+
+	{
+		require.NoError(t, ses.SetSessionSysVar(ctx, function.MoTableRowsSizeForceUpdateVarName, "yes"))
+		require.True(t, function.GetForceUpdateVariable(proc))
+
+		require.NoError(t, ses.SetSessionSysVar(ctx, function.MoTableRowsSizeForceUpdateVarName, "no"))
+		require.False(t, function.GetForceUpdateVariable(proc))
+	}
+
+	{
+		require.NoError(t, ses.SetSessionSysVar(ctx, function.MoTableRowSizeUseOldImplVarName, "yes"))
+		require.True(t, function.GetUseOldImplVariable(proc))
+
+		require.NoError(t, ses.SetSessionSysVar(ctx, function.MoTableRowSizeUseOldImplVarName, "no"))
+		require.False(t, function.GetUseOldImplVariable(proc))
+	}
+
+	{
+		require.NoError(t, ses.SetSessionSysVar(ctx, function.MoTableRowSizeResetUpdateTimeVarName, "yes"))
+		require.True(t, function.GetResetUpdateTimeVariable(proc))
+
+		require.NoError(t, ses.SetSessionSysVar(ctx, function.MoTableRowSizeResetUpdateTimeVarName, "no"))
+		require.False(t, function.GetResetUpdateTimeVariable(proc))
+	}
+}

pkg/frontend/predefined.go

@@ -283,6 +283,18 @@ var (
 				update_time timestamp not null default current_timestamp on update current_timestamp,
     			primary key(account_id, key_id)
 			)`
+
+	MoCatalogMoTableStatsDDL = fmt.Sprintf(`create table mo_catalog.%s (
+    			account_id bigint signed,
+    			database_id bigint signed,
+    			table_id bigint signed,
+    			database_name varchar(255),
+    			table_name varchar(255),
+    			table_stats json,
+    			update_time datetime(6) not null,
+    			takes bigint unsigned,
+    			primary key(account_id, database_id, table_id)
+			)`, catalog.MO_TABLE_STATS)
 )
 
 // `mo_catalog` database system tables

pkg/frontend/server.go

@@ -395,6 +395,10 @@ func setPu(service string, pu *config.ParameterUnit) {
 	getServerLevelVars(service).Pu.Store(pu)
 }
 
+func SetPUForExternalUT(service string, pu *config.ParameterUnit) {
+	setPu(service, pu)
+}
+
 func getPu(service string) *config.ParameterUnit {
 	return getServerLevelVars(service).Pu.Load().(*config.ParameterUnit)
 }

pkg/frontend/variables.go

@@ -3646,6 +3646,30 @@ var gSysVarsDefs = map[string]SystemVariable{
 		Type:              InitSystemVariableStringType("optimizer_switch"),
 		Default:           "index_merge=on,index_merge_union=on,index_merge_sort_union=on,index_merge_intersection=on,engine_condition_pushdown=on,index_condition_pushdown=on,mrr=on,mrr_cost_based=on,block_nested_loop=on,batched_key_access=off,materialization=on,semijoin=on,loosescan=on,firstmatch=on,duplicateweedout=on,subquery_materialization_cost_based=on,use_index_extensions=on,condition_fanout_filter=on,derived_merge=on,use_invisible_indexes=off,skip_scan=on,hash_join=on,subquery_to_derived=off,prefer_ordering_index=on,hypergraph_optimizer=off,derived_condition_pushdown=on,hash_set_operations=on",
 	},
+	"mo_table_stats.force_update": {
+		Name:              "mo_table_stats.force_update",
+		Scope:             ScopeSession,
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              InitSystemVariableStringType("mo_table_stats.force_update"),
+		Default:           "",
+	},
+	"mo_table_stats.use_old_impl": {
+		Name:              "mo_table_stats.use_old_impl",
+		Scope:             ScopeSession,
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              InitSystemVariableStringType("mo_table_stats.use_old_impl"),
+		Default:           "",
+	},
+	"mo_table_stats.reset_update_time": {
+		Name:              "mo_table_stats.reset_update_time",
+		Scope:             ScopeSession,
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              InitSystemVariableStringType("mo_table_stats.reset_update_time"),
+		Default:           "",
+	},
 }
 
 func updateTimeZone(ctx context.Context, sess *Session, sv *SystemVariables, name string, val interface{}) error {