Merge branch 'main' into fix-context

pkg/cnservice/server.go

@@ -223,25 +223,6 @@ func NewService(
 	srv.server = server
 	srv.storeEngine = pu.StorageEngine
 
-	srv.requestHandler = func(ctx context.Context,
-		cnAddr string,
-		message morpc.Message,
-		cs morpc.ClientSession,
-		engine engine.Engine,
-		fService fileservice.FileService,
-		lockService lockservice.LockService,
-		queryClient qclient.QueryClient,
-		hakeeper logservice.CNHAKeeperClient,
-		udfService udf.Service,
-		cli client.TxnClient,
-		aicm *defines.AutoIncrCacheManager,
-		messageAcquirer func() morpc.Message) error {
-		return nil
-	}
-	for _, opt := range options {
-		opt(srv)
-	}
-
 	// TODO: global client need to refactor
 	c, err := cnclient.NewPipelineClient(
 		cfg.UUID,

pkg/embed/cluster_test.go

@@ -26,8 +26,6 @@ import (
 )
 
 func TestBasicCluster(t *testing.T) {
-	// TODO: #17842
-	t.SkipNow()
 	c, err := NewCluster(WithCNCount(3))
 	require.NoError(t, err)
 	require.NoError(t, c.Start())

pkg/util/export/batch_processor_test.go

@@ -40,7 +40,8 @@ import (
 )
 
 func init() {
-	time.Local = time.FixedZone("CST", 0) // set time-zone +0000
+	// Tips: Op 'time.Local = time.FixedZone(...)' would cause DATA RACE against to time.Now()
+
 	logutil.SetupMOLogger(&logutil.LogConfig{
 		Level:      zapcore.DebugLevel.String(),
 		Format:     "console",

pkg/util/export/merge_test.go

@@ -47,7 +47,8 @@ import (
 )
 
 func init() {
-	time.Local = time.FixedZone("CST", 0) // set time-zone +0000
+	// Tips: Op 'time.Local = time.FixedZone(...)' would cause DATA RACE against to time.Now()
+
 	table.RegisterTableDefine(dummyTable)
 	runtime.SetupServiceBasedRuntime("", runtime.NewRuntime(metadata.ServiceType_CN, "test", logutil.GetGlobalLogger()))
 }
@@ -124,8 +125,8 @@ func TestInitCronExpr(t *testing.T) {
 				sche, err := parser.Parse(MergeTaskCronExpr)
 				require.Nil(t, err)
 
-				now := time.Unix(60, 0)
-				next := sche.Next(time.UnixMilli(now.UnixMilli()))
+				now := time.Unix(60, 0).UTC()
+				next := sche.Next(time.UnixMilli(now.UnixMilli()).UTC())
 				t.Logf("duration: %v, expr: %s, next: %v", tt.args.duration, MergeTaskCronExpr, next)
 				if tt.expectDuration > 0 {
 					require.Equal(t, tt.expectDuration, next.Sub(now))

pkg/util/export/table/table_test.go

@@ -29,7 +29,7 @@ import (
 )
 
 func init() {
-	time.Local = time.FixedZone("CST", 0) // set time-zone +0000
+	// Tips: Op 'time.Local = time.FixedZone(...)' would cause DATA RACE against to time.Now()
 }
 
 func TestNoopTableOptions_FormatDdl(t *testing.T) {
@@ -364,9 +364,9 @@ func TestColumnField_EncodedDatetime(t *testing.T) {
 		{
 			name: "Unix_Zero",
 			fields: fields{
-				cf: TimeField(time.Unix(0, 0)),
+				cf: TimeField(time.Unix(0, 0).UTC()),
 			},
-			wantT: time.Unix(0, 0),
+			wantT: time.Unix(0, 0).UTC(),
 			want:  "1970-01-01 00:00:00.000000",
 		},
 	}

pkg/util/export/table/type_test.go

@@ -27,7 +27,7 @@ import (
 )
 
 func init() {
-	time.Local = time.FixedZone("CST", 0) // set time-zone +0000
+	// Tips: Op 'time.Local = time.FixedZone(...)' would cause DATA RACE against to time.Now()
 }
 
 func TestPathBuilder(t *testing.T) {
@@ -58,7 +58,7 @@ func TestPathBuilder(t *testing.T) {
 			args: args{
 				account:  "user",
 				typ:      MergeLogTypeLogs,
-				ts:       time.Unix(0, 0),
+				ts:       time.Unix(0, 0).UTC(),
 				db:       "db",
 				name:     "table",
 				nodeUUID: "123456",
@@ -75,7 +75,7 @@ func TestPathBuilder(t *testing.T) {
 			args: args{
 				account:  "user",
 				typ:      MergeLogTypeLogs,
-				ts:       time.Unix(0, 0),
+				ts:       time.Unix(0, 0).UTC(),
 				db:       "db",
 				name:     "table",
 				nodeUUID: "123456",

pkg/util/trace/impl/motrace/buffer_pipe_test.go

@@ -52,8 +52,9 @@ func noopReportError(context.Context, error, int) {}
 var dummyBaseTime time.Time
 
 func init() {
-	time.Local = time.FixedZone("CST", 0) // set time-zone +0000
-	dummyBaseTime = time.Unix(0, 0)
+	// Tips: Op 'time.Local = time.FixedZone(...)' would cause DATA RACE against to time.Now()
+
+	dummyBaseTime = time.Unix(0, 0).UTC()
 	SV := config.ObservabilityParameters{}
 	SV.SetDefaultValues("v0.test.0")
 	SV.TraceExportInterval = 15

pkg/vm/engine/disttae/db.go

@@ -33,17 +33,9 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logtail"
 )
 
-var (
-	timeFixed             bool
-	moCatalogCreatedTime  *vector.Vector
-	moDatabaseCreatedTime *vector.Vector
-	moTablesCreatedTime   *vector.Vector
-	moColumnsCreatedTime  *vector.Vector
-)
-
 // tryAdjustThreeTablesCreatedTime analyzes the mo_tables batch and tries to adjust the created time of the three tables.
-func tryAdjustThreeTablesCreatedTimeWithBatch(b *batch.Batch) {
-	if timeFixed {
+func (e *Engine) tryAdjustThreeTablesCreatedTimeWithBatch(b *batch.Batch) {
+	if e.timeFixed {
 		return
 	}
 
@@ -55,11 +47,11 @@ func tryAdjustThreeTablesCreatedTimeWithBatch(b *batch.Batch) {
 		tname := b.Vecs[tnameIdx].GetStringAt(i)
 		if aid == 0 && tname == "mo_user" {
 			ts := vector.GetFixedAt[types.Timestamp](b.Vecs[createdTsIdx], i)
-			vector.SetFixedAt(moDatabaseCreatedTime, 0, ts)
-			vector.SetFixedAt(moTablesCreatedTime, 0, ts)
-			vector.SetFixedAt(moColumnsCreatedTime, 0, ts)
-			vector.SetFixedAt(moCatalogCreatedTime, 0, ts)
-			timeFixed = true
+			vector.SetFixedAt(e.moDatabaseCreatedTime, 0, ts)
+			vector.SetFixedAt(e.moTablesCreatedTime, 0, ts)
+			vector.SetFixedAt(e.moColumnsCreatedTime, 0, ts)
+			vector.SetFixedAt(e.moCatalogCreatedTime, 0, ts)
+			e.timeFixed = true
 			return
 		}
 	}
@@ -90,7 +82,7 @@ func (e *Engine) init(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		moCatalogCreatedTime = bat.Vecs[catalog.MO_DATABASE_CREATED_TIME_IDX]
+		e.moCatalogCreatedTime = bat.Vecs[catalog.MO_DATABASE_CREATED_TIME_IDX]
 		ibat, err := fillRandomRowidAndZeroTs(bat, m)
 		if err != nil {
 			bat.Clean(m)
@@ -119,7 +111,7 @@ func (e *Engine) init(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		moDatabaseCreatedTime = bat.Vecs[catalog.MO_TABLES_CREATED_TIME_IDX]
+		e.moDatabaseCreatedTime = bat.Vecs[catalog.MO_TABLES_CREATED_TIME_IDX]
 		ibat, err := fillRandomRowidAndZeroTs(bat, m)
 		if err != nil {
 			bat.Clean(m)
@@ -177,7 +169,7 @@ func (e *Engine) init(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		moTablesCreatedTime = bat.Vecs[catalog.MO_TABLES_CREATED_TIME_IDX]
+		e.moTablesCreatedTime = bat.Vecs[catalog.MO_TABLES_CREATED_TIME_IDX]
 		ibat, err := fillRandomRowidAndZeroTs(bat, m)
 		if err != nil {
 			bat.Clean(m)
@@ -225,7 +217,7 @@ func (e *Engine) init(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		moColumnsCreatedTime = bat.Vecs[catalog.MO_TABLES_CREATED_TIME_IDX]
+		e.moColumnsCreatedTime = bat.Vecs[catalog.MO_TABLES_CREATED_TIME_IDX]
 		ibat, err := fillRandomRowidAndZeroTs(bat, m)
 		if err != nil {
 			bat.Clean(m)

pkg/vm/engine/disttae/logtail_consumer.go

@@ -654,7 +654,7 @@ func (c *PushClient) replayCatalogCache(ctx context.Context, e *Engine) error {
 		if err = fillTsVecForSysTableQueryBatch(b, typeTs, result.Mp); err != nil {
 			return err
 		}
-		tryAdjustThreeTablesCreatedTimeWithBatch(b)
+		e.tryAdjustThreeTablesCreatedTimeWithBatch(b)
 		e.catalog.InsertTable(b)
 	}
 

pkg/vm/engine/disttae/types.go

@@ -200,6 +200,12 @@ type Engine struct {
 
 	//for message on multiCN, use uuid to get the messageBoard
 	messageCenter *message.MessageCenter
+
+	timeFixed             bool
+	moCatalogCreatedTime  *vector.Vector
+	moDatabaseCreatedTime *vector.Vector
+	moTablesCreatedTime   *vector.Vector
+	moColumnsCreatedTime  *vector.Vector
 }
 
 func (txn *Transaction) String() string {

pkg/vm/engine/tae/containers/pool.go

@@ -300,6 +300,3 @@ func newVectorElement(pool *VectorPool, t *types.Type, mp *mpool.MPool) *vectorP
 	}
 	return element
 }
-
-//go:linkname fastrand runtime.fastrand
-func fastrand() uint32

pkg/vm/engine/tae/containers/runtime.go

@@ -0,0 +1,24 @@
+// Copyright 2021 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.
+
+//go:build !go1.22
+
+package containers
+
+import (
+	_ "unsafe"
+)
+
+//go:linkname fastrand runtime.fastrand
+func fastrand() uint32

pkg/vm/engine/tae/containers/runtime_1.22.go

@@ -0,0 +1,24 @@
+// Copyright 2021 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.
+
+//go:build go1.22
+
+package containers
+
+import (
+	_ "unsafe"
+)
+
+//go:linkname fastrand runtime.cheaprand
+func fastrand() uint32