Merge branch 'main' into apply_pk_filter_on_workspace

Makefile

@@ -98,7 +98,7 @@ pb: vendor-build generate-pb fmt
 RACE_OPT :=
 DEBUG_OPT :=
 CGO_DEBUG_OPT :=
-CGO_OPTS=CGO_CFLAGS="-I$(ROOT_DIR)/cgo " CGO_LDFLAGS="-L$(ROOT_DIR)/cgo -lm -lmo"
+CGO_OPTS :=
 GOLDFLAGS=-ldflags="-X '$(GO_MODULE)/pkg/version.GoVersion=$(GO_VERSION)' -X '$(GO_MODULE)/pkg/version.BranchName=$(BRANCH_NAME)' -X '$(GO_MODULE)/pkg/version.CommitID=$(LAST_COMMIT_ID)' -X '$(GO_MODULE)/pkg/version.BuildTime=$(BUILD_TIME)' -X '$(GO_MODULE)/pkg/version.Version=$(MO_VERSION)'"
 TAGS :=
 
@@ -108,7 +108,7 @@ cgo:
 
 # build mo-service binary
 .PHONY: build
-build: config cgo
+build: config
 	$(info [Build binary])
 	$(CGO_OPTS) go build $(TAGS) $(RACE_OPT) $(GOLDFLAGS) $(DEBUG_OPT) -o $(BIN_NAME) ./cmd/mo-service
 
@@ -139,7 +139,7 @@ musl:
 
 # build mo-tool
 .PHONY: mo-tool
-mo-tool: config cgo
+mo-tool: config
 	$(info [Build mo-tool tool])
 	$(CGO_OPTS) go build -o mo-tool ./cmd/mo-tool
 
@@ -158,7 +158,7 @@ debug: build
 # Excluding frontend test cases temporarily
 # Argument SKIP_TEST to skip a specific go test
 .PHONY: ut
-ut: config cgo
+ut: config
 	$(info [Unit testing])
 ifeq ($(UNAME_S),Darwin)
 	@cd optools && ./run_ut.sh UT $(SKIP_TEST)
@@ -240,7 +240,7 @@ install-static-check-tools:
 	@go install github.com/apache/skywalking-eyes/cmd/[email protected]
 
 .PHONY: static-check
-static-check: config cgo err-check
+static-check: config err-check
 	$(CGO_OPTS) go vet -vettool=`which molint` ./...
 	$(CGO_OPTS) license-eye -c .licenserc.yml header check
 	$(CGO_OPTS) license-eye -c .licenserc.yml dep check

cgo/lib.go

@@ -0,0 +1,21 @@
+// 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 cgo
+
+/*
+#cgo CFLAGS: -O3 -std=c99
+#cgo LDFLAGS: -lm
+*/
+import "C"

go.mod

@@ -215,7 +215,7 @@ require (
 replace github.com/hashicorp/memberlist => github.com/matrixorigin/memberlist v0.5.1-0.20230322082342-95015c95ee76
 
 replace (
-	github.com/elastic/gosigar v0.14.2 => github.com/matrixorigin/gosigar v0.14.3-0.20240721074812-0d5bde0e9f91
+	github.com/elastic/gosigar v0.14.2 => github.com/matrixorigin/gosigar v0.14.3-0.20241204071856-40aab500bfac
 	github.com/fagongzi/goetty/v2 v2.0.3-0.20230628075727-26c9a2fd5fb8 => github.com/matrixorigin/goetty/v2 v2.0.0-20240611082008-a4de209fff3d
 	github.com/lni/dragonboat/v4 v4.0.0-20220815145555-6f622e8bcbef => github.com/matrixorigin/dragonboat/v4 v4.0.0-20241019050137-1c6138e9cf8b
 	github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4 => github.com/matrixorigin/goutils v1.3.1-0.20220604063047-388d67b4dbc4

go.sum

@@ -497,8 +497,8 @@ github.com/matrixorigin/dragonboat/v4 v4.0.0-20241019050137-1c6138e9cf8b h1:DQMm
 github.com/matrixorigin/dragonboat/v4 v4.0.0-20241019050137-1c6138e9cf8b/go.mod h1:bR6ZGoUwApH4/P4+AezNGT0xehljg+j+Z/Q2Fz+Y4a0=
 github.com/matrixorigin/goetty/v2 v2.0.0-20240611082008-a4de209fff3d h1:wSlkJlWXZ3if1sH8Bc/lUUOWhTw91lgYGSFOHqy1tcw=
 github.com/matrixorigin/goetty/v2 v2.0.0-20240611082008-a4de209fff3d/go.mod h1:OwIBpVwRW1HjF/Jhc2Av3UvG2NygMg+bdqGxZaqwhU0=
-github.com/matrixorigin/gosigar v0.14.3-0.20240721074812-0d5bde0e9f91 h1:+P8GoCsNjqb3Vtrx0gPHxhoVPBXXLDwdy3wsvGhHemM=
-github.com/matrixorigin/gosigar v0.14.3-0.20240721074812-0d5bde0e9f91/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
+github.com/matrixorigin/gosigar v0.14.3-0.20241204071856-40aab500bfac h1:KRPOwOcSZRfWT7w8mr7BmFLoFB75ljgqjkg47xkN/Pc=
+github.com/matrixorigin/gosigar v0.14.3-0.20241204071856-40aab500bfac/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
 github.com/matrixorigin/goutils v1.3.1-0.20220604063047-388d67b4dbc4 h1:+SmZP2bG+YcO/ntzjCleCu6hFTJiue7Oj2tftpJKTlU=
 github.com/matrixorigin/goutils v1.3.1-0.20220604063047-388d67b4dbc4/go.mod h1:LIHvF0fflR+zyXUQFQOiHPpKANf3UIr7DFIv5CBPOoU=
 github.com/matrixorigin/memberlist v0.5.1-0.20230322082342-95015c95ee76 h1:MpmqMPooJ0Ea7W4ldIGbQV4D3z+sEiCu6C6aTibiwiQ=

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/common/bitmap/cbitmap.go

@@ -15,13 +15,13 @@
 package bitmap
 
 /*
-#include "mo.h"
-
-#cgo CFLAGS: -I../../../cgo
-#cgo LDFLAGS: -L../../../cgo -lmo -lm
+#include "../../../cgo/mo.h"
 */
 import "C"
-import "unsafe"
+import (
+	_ "github.com/matrixorigin/matrixone/cgo"
+	"unsafe"
+)
 
 func (n *Bitmap) cPtr() *C.uint64_t {
 	return (*C.uint64_t)(unsafe.Pointer(&n.data[0]))

pkg/common/moerr/error.go

@@ -225,6 +225,7 @@ const (
 	ErrPrevCheckpointNotFinished  uint16 = 20635
 	ErrCantDelGCChecker           uint16 = 20636
 	ErrTxnUnknown                 uint16 = 20637
+	ErrTxnControl                 uint16 = 20638
 
 	// Group 7: lock service
 	// ErrDeadLockDetected lockservice has detected a deadlock and should abort the transaction if it receives this error
@@ -468,6 +469,7 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	ErrCantCompileForPrepare:      {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "can not compile for prepare"},
 	ErrCantDelGCChecker:           {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "can't delete gc checker"},
 	ErrTxnUnknown:                 {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "txn commit status is unknown: %s"},
+	ErrTxnControl:                 {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "txn control error: %s"},
 
 	// Group 7: lock service
 	ErrDeadLockDetected:        {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "deadlock detected"},

pkg/common/moerr/error_no_ctx.go

@@ -334,6 +334,10 @@ func NewCantDelGCCheckerNoCtx() *Error {
 	return newError(Context(), ErrCantDelGCChecker)
 }
 
+func NewTxnControlErrorNoCtxf(format string, args ...any) *Error {
+	return newError(Context(), ErrTxnControl, fmt.Sprintf(format, args...))
+}
+
 func NewNotFoundNoCtx() *Error {
 	return newError(Context(), ErrNotFound)
 }

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
@@ -3612,7 +3615,7 @@ func doDropAccount(ctx context.Context, bh BackgroundExec, ses *Session, da *dro
 
 	dropAccountFunc := func() (rtnErr error) {
 		ses.Infof(ctx, "dropAccount %s sql: %s", da.Name, getAccountIdNamesSql)
-		_, nameInfoMap, rtnErr := getAccounts(ctx, bh)
+		_, nameInfoMap, rtnErr := getAccounts(ctx, bh, true)
 		if rtnErr != nil {
 			return rtnErr
 		}
@@ -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
 	}
 
@@ -7669,7 +7675,7 @@ func createTablesInInformationSchemaOfGeneralTenant(ctx context.Context, bh Back
 // createSubscription insert records into mo_subs of To-All-Publications
 func createSubscription(ctx context.Context, bh BackgroundExec, newTenant *TenantInfo) (err error) {
 	// get all accounts
-	accIdInfoMap, accNameInfoMap, err := getAccounts(ctx, bh)
+	accIdInfoMap, accNameInfoMap, err := getAccounts(ctx, bh, false)
 	if err != nil {
 		return
 	}

pkg/frontend/authenticate_test.go

@@ -7126,7 +7126,7 @@ func Test_doDropAccount(t *testing.T) {
 		bh.sql2result["commit;"] = nil
 		bh.sql2result["rollback;"] = nil
 
-		sql := getAccountIdNamesSql
+		sql := getAccountIdNamesSql + " for update"
 		mrs := newMrsForGetAllAccounts([][]interface{}{
 			{uint64(0), "sys", "open", uint64(1), nil},
 			{uint64(1), "acc", "open", uint64(1), nil},
@@ -7187,7 +7187,7 @@ func Test_doDropAccount(t *testing.T) {
 		bh.sql2result["commit;"] = nil
 		bh.sql2result["rollback;"] = nil
 
-		sql := getAccountIdNamesSql
+		sql := getAccountIdNamesSql + " for update"
 		bh.sql2result[sql] = newMrsForGetAllAccounts([][]interface{}{})
 
 		sql, _ = getSqlForDeleteAccountFromMoAccount(context.TODO(), mustUnboxExprStr(stmt.Name))
@@ -7234,7 +7234,7 @@ func Test_doDropAccount(t *testing.T) {
 		bh.sql2result["commit;"] = nil
 		bh.sql2result["rollback;"] = nil
 
-		sql := getAccountIdNamesSql
+		sql := getAccountIdNamesSql + " for update"
 		bh.sql2result[sql] = newMrsForGetAllAccounts([][]interface{}{})
 
 		sql, _ = getSqlForDeleteAccountFromMoAccount(context.TODO(), mustUnboxExprStr(stmt.Name))

pkg/frontend/back_exec.go

@@ -999,6 +999,8 @@ func (backSes *backSession) GetSessionSysVar(name string) (interface{}, error) {
 			return int64(0), nil
 		}
 		return int64(1), nil
+	case "mo_table_stats.force_update", "mo_table_stats.use_old_impl", "mo_table_stats.reset_update_time":
+		return backSes.upstream.GetSessionSysVar(name)
 	}
 	return nil, nil
 }