Repository: matrixorigin/matrixone
Pull request #4873 (2022-09-05 08:51:46)

txnstorage: various fixes


## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #4250 

## What this PR does / why we need it:
txnstorage: add AnyRow.String

txnstorage: fix MemHandler.rangeBatchPhysicalRows

txnstorage: update TestFrontend

txnstorage: fix BatchIter


Patch:
diff --git a/pkg/txn/storage/txn/any.go b/pkg/txn/storage/txn/any.go
index 6ef9497e0..cb4be3127 100644
--- a/pkg/txn/storage/txn/any.go
+++ b/pkg/txn/storage/txn/any.go
@@ -14,7 +14,10 @@
 
 package txnstorage
 
-import "fmt"
+import (
+	"fmt"
+	"strings"
+)
 
 type AnyKey []any
 
@@ -73,6 +76,17 @@ func (a *AnyRow) PrimaryKey() AnyKey {
 	return a.primaryKey
 }
 
+func (a *AnyRow) String() string {
+	buf := new(strings.Builder)
+	buf.WriteString("AnyRow{")
+	buf.WriteString(fmt.Sprintf("key: %+v", a.primaryKey))
+	for key, value := range a.attributes {
+		buf.WriteString(fmt.Sprintf(", %s: %v", key, value))
+	}
+	buf.WriteString("}")
+	return buf.String()
+}
+
 func NewAnyRow() *AnyRow {
 	return &AnyRow{
 		attributes: make(map[string]any),
diff --git a/pkg/txn/storage/txn/batch.go b/pkg/txn/storage/txn/batch.go
index 629fde1a8..a7c58abd3 100644
--- a/pkg/txn/storage/txn/batch.go
+++ b/pkg/txn/storage/txn/batch.go
@@ -139,7 +139,7 @@ func NewBatchIter(b *batch.Batch) BatchIter {
 			case types.T_char, types.T_varchar, types.T_json, types.T_blob:
 				info := vec.Col.(*types.Bytes)
 				if i < len(info.Offsets) {
-					str := vec.Data[info.Offsets[i] : info.Offsets[i]+info.Lengths[i]]
+					str := info.Data[info.Offsets[i] : info.Offsets[i]+info.Lengths[i]]
 					tuple = append(tuple, str)
 				} else {
 					return
@@ -240,7 +240,7 @@ func vectorAt(vec *vector.Vector, i int) any {
 
 	case types.T_char, types.T_varchar, types.T_json, types.T_blob:
 		info := vec.Col.(*types.Bytes)
-		str := vec.Data[info.Offsets[i] : info.Offsets[i]+info.Lengths[i]]
+		str := info.Data[info.Offsets[i] : info.Offsets[i]+info.Lengths[i]]
 		return str
 
 	case types.T_date:
diff --git a/pkg/txn/storage/txn/mem_handler.go b/pkg/txn/storage/txn/mem_handler.go
index c82504790..3779f9cd1 100644
--- a/pkg/txn/storage/txn/mem_handler.go
+++ b/pkg/txn/storage/txn/mem_handler.go
@@ -15,10 +15,11 @@
 package txnstorage
 
 import (
+	"crypto/rand"
 	"database/sql"
+	"encoding/binary"
 	"errors"
 	"fmt"
-	"math/rand"
 	"sort"
 	"sync"
 
@@ -967,11 +968,20 @@ func (m *MemHandler) rangeBatchPhysicalRows(
 		}
 
 		// add version
-		a := rand.Int63()
-		b := rand.Int63()
-		version := types.Decimal128FromInt64Raw(a, b)
+		//TODO use [16]byte
+		var a int64
+		err := binary.Read(rand.Reader, binary.LittleEndian, &a)
+		if err != nil {
+			return err
+		}
+		version := types.MustDecimal128FromString(fmt.Sprintf("%d", a))
 		physicalRow.attributes[nameToAttrs[rowVersionColumnName].ID] = version
 
+		// use version as primary key if no primary key is provided
+		if len(physicalRow.primaryKey) == 0 {
+			physicalRow.primaryKey = append(physicalRow.primaryKey, typeConv(version))
+		}
+
 		if err := fn(table, physicalRow); err != nil {
 			return err
 		}


Test patch:
diff --git a/pkg/txn/storage/txn/frontend_test.go b/pkg/txn/storage/txn/frontend_test.go
index 02ee7d5e6..c5b33b47b 100644
--- a/pkg/txn/storage/txn/frontend_test.go
+++ b/pkg/txn/storage/txn/frontend_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/util/trace"
 	txnengine "github.com/matrixorigin/matrixone/pkg/vm/engine/txn"
 	"github.com/matrixorigin/matrixone/pkg/vm/mempool"
+	"github.com/matrixorigin/matrixone/pkg/vm/mmu/guest"
 	"github.com/matrixorigin/matrixone/pkg/vm/mmu/host"
 	"github.com/stretchr/testify/assert"
 )
@@ -50,22 +51,25 @@ func TestFrontend(t *testing.T) {
 	}
 	frontendParameters.SetDefaultValues()
 
+	shard := logservicepb.DNShardInfo{
+		ShardID:   2,
+		ReplicaID: 2,
+	}
+	shards := []logservicepb.DNShardInfo{
+		shard,
+	}
+	dnStore := logservicepb.DNStore{
+		UUID:           uuid.NewString(),
+		ServiceAddress: "1",
+		Shards:         shards,
+	}
 	engine := txnengine.New(
 		ctx,
 		new(txnengine.ShardToSingleStatic),
 		func() (logservicepb.ClusterDetails, error) {
 			return logservicepb.ClusterDetails{
 				DNStores: []logservicepb.DNStore{
-					{
-						UUID:           uuid.NewString(),
-						ServiceAddress: "1",
-						Shards: []logservicepb.DNShardInfo{
-							{
-								ShardID:   2,
-								ReplicaID: 2,
-							},
-						},
-					},
+					dnStore,
 				},
 			}, nil
 		},
@@ -85,10 +89,14 @@ func TestFrontend(t *testing.T) {
 		storage: storage,
 	}
 
+	hostMMU := host.New(frontendParameters.HostMmuLimitation)
+	guestMMU := guest.New(frontendParameters.GuestMmuLimitation, hostMMU)
+	memoryPool := mempool.New()
+
 	pu := &config.ParameterUnit{
 		SV:            frontendParameters,
-		HostMmu:       host.New(frontendParameters.HostMmuLimitation),
-		Mempool:       mempool.New(),
+		HostMmu:       hostMMU,
+		Mempool:       memoryPool,
 		StorageEngine: engine,
 		TxnClient:     txnClient,
 		FileService:   testutil.NewFS(),
@@ -100,4 +108,25 @@ func TestFrontend(t *testing.T) {
 
 	err = frontend.InitSysTenant(ctx)
 	assert.Nil(t, err)
+
+	globalVars := new(frontend.GlobalSystemVariables)
+	frontend.InitGlobalSystemVariables(globalVars)
+
+	session := frontend.NewSession(
+		frontend.NewMysqlClientProtocol(
+			0,
+			nil, // goetty IOSession
+			1024,
+			frontendParameters,
+		),
+		guestMMU,
+		memoryPool,
+		pu,
+		globalVars,
+	)
+	session.SetRequestContext(ctx)
+
+	_, err = session.AuthenticateUser("root")
+	assert.Nil(t, err)
+
 }