Repository: matrixorigin/matrixone
Pull request #8250 (2023-03-03 11:49:15)

Revert "disttae: various refactors (#8245)"


## 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 #

## What this PR does / why we need it:
This reverts commit 06f52a314d203b000027bcfd2930dbfb728f6e0c.


Patch:
diff --git a/pkg/vm/engine/disttae/db.go b/pkg/vm/engine/disttae/db.go
index 3f5033daf..41c3d968b 100644
--- a/pkg/vm/engine/disttae/db.go
+++ b/pkg/vm/engine/disttae/db.go
@@ -34,7 +34,7 @@ func (e *Engine) init(ctx context.Context, m *mpool.MPool) error {
 	{
 		parts := make(Partitions, len(e.dnMap))
 		for i := range parts {
-			parts[i] = NewPartition(nil)
+			parts[i] = NewPartition(nil, false)
 		}
 		e.partitions[[2]uint64{catalog.MO_CATALOG_ID, catalog.MO_DATABASE_ID}] = parts
 	}
@@ -42,7 +42,7 @@ func (e *Engine) init(ctx context.Context, m *mpool.MPool) error {
 	{
 		parts := make(Partitions, len(e.dnMap))
 		for i := range parts {
-			parts[i] = NewPartition(nil)
+			parts[i] = NewPartition(nil, false)
 		}
 		e.partitions[[2]uint64{catalog.MO_CATALOG_ID, catalog.MO_TABLES_ID}] = parts
 	}
@@ -50,7 +50,7 @@ func (e *Engine) init(ctx context.Context, m *mpool.MPool) error {
 	{
 		parts := make(Partitions, len(e.dnMap))
 		for i := range parts {
-			parts[i] = NewPartition(nil)
+			parts[i] = NewPartition(nil, false)
 		}
 		e.partitions[[2]uint64{catalog.MO_CATALOG_ID, catalog.MO_COLUMNS_ID}] = parts
 	}
@@ -288,6 +288,20 @@ func (e *Engine) init(ctx context.Context, m *mpool.MPool) error {
 	return nil
 }
 
+func (e *Engine) getMetaPartitions(name string) Partitions {
+	e.Lock()
+	defer e.Unlock()
+	parts, ok := e.metaTables[name]
+	if !ok { // create a new table
+		parts = make(Partitions, len(e.dnMap))
+		for i := range parts {
+			parts[i] = NewPartition(nil, true)
+		}
+		e.metaTables[name] = parts
+	}
+	return parts
+}
+
 func (e *Engine) getPartitions(databaseId, tableId uint64) Partitions {
 	e.Lock()
 	defer e.Unlock()
@@ -295,7 +309,7 @@ func (e *Engine) getPartitions(databaseId, tableId uint64) Partitions {
 	if !ok { // create a new table
 		parts = make(Partitions, len(e.dnMap))
 		for i := range parts {
-			parts[i] = NewPartition(nil)
+			parts[i] = NewPartition(nil, false)
 		}
 		e.partitions[[2]uint64{databaseId, tableId}] = parts
 	}
@@ -313,13 +327,13 @@ func (e *Engine) UpdateOfPull(ctx context.Context, dnList []DNStore, tbl *txnTab
 	if !ok { // create a new table
 		parts = make(Partitions, len(e.dnMap))
 		for i := range parts {
-			parts[i] = NewPartition(nil)
+			parts[i] = NewPartition(nil, false)
 		}
 		e.partitions[[2]uint64{databaseId, tableId}] = parts
 	}
 	e.Unlock()
 
-	for _, dn := range dnList {
+	for i, dn := range dnList {
 		part := parts[e.dnMap[dn.ServiceID]]
 
 		select {
@@ -334,7 +348,7 @@ func (e *Engine) UpdateOfPull(ctx context.Context, dnList []DNStore, tbl *txnTab
 		}
 
 		if err := updatePartitionOfPull(
-			primaryIdx, tbl, ctx, op, e, part, parts, dn,
+			i, primaryIdx, tbl, ctx, op, e, part, dn,
 			genSyncLogTailReq(part.ts, ts, databaseId, tableId),
 		); err != nil {
 			part.lock <- struct{}{}
diff --git a/pkg/vm/engine/disttae/engine.go b/pkg/vm/engine/disttae/engine.go
index ec2761b01..55576801c 100644
--- a/pkg/vm/engine/disttae/engine.go
+++ b/pkg/vm/engine/disttae/engine.go
@@ -71,6 +71,7 @@ func New(
 		txnHeap:    &transactionHeap{},
 		txns:       make(map[string]*Transaction),
 		dnMap:      dnMap,
+		metaTables: make(map[string]Partitions),
 		partitions: make(map[[2]uint64]Partitions),
 	}
 
diff --git a/pkg/vm/engine/disttae/logtail.go b/pkg/vm/engine/disttae/logtail.go
index b57ce60b9..4681c5011 100644
--- a/pkg/vm/engine/disttae/logtail.go
+++ b/pkg/vm/engine/disttae/logtail.go
@@ -16,37 +16,50 @@ package disttae
 
 import (
 	"context"
-
 	"github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/pb/api"
 )
 
-func consumeEntry(
-	ctx context.Context,
-	primaryIdx int,
-	engine *Engine,
-	partition *Partition,
-	state *PartitionState,
-	e *api.Entry,
-) error {
+func consumeEntry(idx, primaryIdx int, tbl *txnTable,
+	ctx context.Context, engine *Engine, partition *Partition, state *PartitionState, e *api.Entry) error {
 
 	state.HandleLogtailEntry(ctx, e, primaryIdx)
 
 	if e.EntryType == api.Entry_Insert {
 		if isMetaTable(e.TableName) {
-			return nil
+			vec, err := vector.ProtoVectorToVector(e.Bat.Vecs[catalog.BLOCKMETA_ID_IDX+MO_PRIMARY_OFF])
+			if err != nil {
+				return err
+			}
+			timeVec, err := vector.ProtoVectorToVector(e.Bat.Vecs[catalog.BLOCKMETA_COMMITTS_IDX+MO_PRIMARY_OFF])
+			if err != nil {
+				return err
+			}
+			vs := vector.MustTCols[uint64](vec)
+			timestamps := vector.MustTCols[types.TS](timeVec)
+			for i, v := range vs {
+				if err := tbl.parts[idx].DeleteByBlockID(ctx, timestamps[i].ToTimestamp(), v); err != nil {
+					if !moerr.IsMoErrCode(err, moerr.ErrTxnWriteConflict) {
+						return err
+					}
+				}
+			}
+			return engine.getMetaPartitions(e.TableName)[idx].Insert(ctx, -1, e.Bat, false)
 		}
 		switch e.TableId {
 		case catalog.MO_TABLES_ID:
 			bat, _ := batch.ProtoBatchToBatch(e.Bat)
-			engine.catalog.InsertTable(bat)
+			tbl.db.txn.engine.catalog.InsertTable(bat)
 		case catalog.MO_DATABASE_ID:
 			bat, _ := batch.ProtoBatchToBatch(e.Bat)
-			engine.catalog.InsertDatabase(bat)
+			tbl.db.txn.engine.catalog.InsertDatabase(bat)
 		case catalog.MO_COLUMNS_ID:
 			bat, _ := batch.ProtoBatchToBatch(e.Bat)
-			engine.catalog.InsertColumns(bat)
+			tbl.db.txn.engine.catalog.InsertColumns(bat)
 		}
 		if primaryIdx >= 0 {
 			return partition.Insert(ctx, MO_PRIMARY_OFF+primaryIdx, e.Bat, false)
@@ -54,15 +67,15 @@ func consumeEntry(
 		return partition.Insert(ctx, primaryIdx, e.Bat, false)
 	}
 	if isMetaTable(e.TableName) {
-		return nil
+		return engine.getMetaPartitions(e.TableName)[idx].Delete(ctx, e.Bat)
 	}
 	switch e.TableId {
 	case catalog.MO_TABLES_ID:
 		bat, _ := batch.ProtoBatchToBatch(e.Bat)
-		engine.catalog.DeleteTable(bat)
+		tbl.db.txn.engine.catalog.DeleteTable(bat)
 	case catalog.MO_DATABASE_ID:
 		bat, _ := batch.ProtoBatchToBatch(e.Bat)
-		engine.catalog.DeleteDatabase(bat)
+		tbl.db.txn.engine.catalog.DeleteDatabase(bat)
 	}
 	return partition.Delete(ctx, e.Bat)
 }
diff --git a/pkg/vm/engine/disttae/logtailPullModel.go b/pkg/vm/engine/disttae/logtailPullModel.go
index 1de1f7d48..1b5f0298d 100644
--- a/pkg/vm/engine/disttae/logtailPullModel.go
+++ b/pkg/vm/engine/disttae/logtailPullModel.go
@@ -29,17 +29,9 @@ import (
 )
 
 // updatePartitionOfPull the old method of log tail pull model.
-func updatePartitionOfPull(
-	primaryIdx int,
-	tbl *txnTable,
-	ctx context.Context,
-	op client.TxnOperator,
-	engine *Engine,
-	partition *Partition,
-	partitions Partitions,
-	dn DNStore,
-	req api.SyncLogTailReq,
-) error {
+func updatePartitionOfPull(idx, primaryIdx int, tbl *txnTable,
+	ctx context.Context, op client.TxnOperator, engine *Engine,
+	partition *Partition, dn DNStore, req api.SyncLogTailReq) error {
 	reqs, err := genLogTailReq(dn, req)
 	if err != nil {
 		return err
@@ -52,7 +44,7 @@ func updatePartitionOfPull(
 	state, doneMutate := partition.MutateState()
 
 	for i := range logTails {
-		if err := consumeLogTailOfPull(primaryIdx, tbl, ctx, engine, partition, partitions, state, logTails[i]); err != nil {
+		if err := consumeLogTailOfPull(idx, primaryIdx, tbl, ctx, engine, partition, state, logTails[i]); err != nil {
 			logutil.Errorf("consume %d-%s logtail error: %v\n", tbl.tableId, tbl.tableName, err)
 			return err
 		}
@@ -80,16 +72,7 @@ func getLogTail(ctx context.Context, op client.TxnOperator, reqs []txn.TxnReques
 	return logTails, nil
 }
 
-func consumeLogTailOfPull(
-	primaryIdx int,
-	tbl *txnTable,
-	ctx context.Context,
-	engine *Engine,
-	partition *Partition,
-	partitions Partitions,
-	state *PartitionState,
-	logTail *api.SyncLogTailResp,
-) (err error) {
+func consumeLogTailOfPull(idx, primaryIdx int, tbl *txnTable, ctx context.Context, engine *Engine, partition *Partition, state *PartitionState, logTail *api.SyncLogTailResp) (err error) {
 	var entries []*api.Entry
 
 	if entries, err = logtail.LoadCheckpointEntries(
@@ -99,18 +82,18 @@ func consumeLogTailOfPull(
 		tbl.tableName,
 		tbl.db.databaseId,
 		tbl.db.databaseName,
-		engine.fs); err != nil {
+		tbl.db.txn.engine.fs); err != nil {
 		return
 	}
 	for _, e := range entries {
-		if err = consumeEntry(ctx, primaryIdx,
+		if err = consumeEntry(idx, primaryIdx, tbl, ctx,
 			engine, partition, state, e); err != nil {
 			return
 		}
 	}
 
 	for i := 0; i < len(logTail.Commands); i++ {
-		if err = consumeEntry(ctx, primaryIdx,
+		if err = consumeEntry(idx, primaryIdx, tbl, ctx,
 			engine, partition, state, logTail.Commands[i]); err != nil {
 			return
 		}
diff --git a/pkg/vm/engine/disttae/logtailPushModel.go b/pkg/vm/engine/disttae/logtailPushModel.go
index 75de94b71..20351ac7a 100644
--- a/pkg/vm/engine/disttae/logtailPushModel.go
+++ b/pkg/vm/engine/disttae/logtailPushModel.go
@@ -411,22 +411,32 @@ func updatePartitionOfPush(
 	state, doneMutate := partition.MutateState()
 
 	key := e.catalog.GetTableById(dbId, tblId)
+	tbl := &txnTable{
+		db: &txnDatabase{
+			databaseId:   dbId,
+			databaseName: "",
+			txn: &Transaction{
+				engine: e,
+			},
+		},
+		parts:        partitions,
+		tableId:      key.Id,
+		tableName:    key.Name,
+		defs:         key.Defs,
+		tableDef:     key.TableDef,
+		primaryIdx:   key.PrimaryIdx,
+		clusterByIdx: key.ClusterByIdx,
+		relKind:      key.Kind,
+		viewdef:      key.ViewDef,
+		comment:      key.Comment,
+		partition:    key.Partition,
+		createSql:    key.CreateSql,
+		constraint:   key.Constraint,
+	}
 
-	if err = consumeLogTailOfPush(
-		ctx,
-		dnId,
-		dbId,
-		"",
-		key.Id,
-		key.Name,
-		key.PrimaryIdx,
-		e,
-		partition,
-		partitions,
-		state,
-		tl,
-	); err != nil {
-		logutil.Errorf("consume %d-%s log tail error: %v\n", key.Id, key.Name, err)
+	if err = consumeLogTailOfPush(ctx,
+		dnId, tbl, e, partition, state, tl); err != nil {
+		logutil.Errorf("consume %d-%s log tail error: %v\n", tbl.tableId, tbl.tableName, err)
 		return err
 	}
 
@@ -441,36 +451,26 @@ func updatePartitionOfPush(
 
 func consumeLogTailOfPush(
 	ctx context.Context,
-	idx int,
-	databaseId uint64,
-	databaseName string,
-	tableId uint64,
-	tableName string,
-	primaryIdx int,
-	engine *Engine,
-	partition *Partition,
-	partitions Partitions,
-	state *PartitionState,
-	lt *logtail.TableLogtail,
-) (err error) {
+	idx int, tbl *txnTable,
+	engine *Engine, partition *Partition, state *PartitionState, lt *logtail.TableLogtail) (err error) {
 	var entries []*api.Entry
 
 	if entries, err = logtail2.LoadCheckpointEntries(
 		ctx,
 		lt.CkpLocation,
-		tableId, tableName,
-		databaseId, databaseName, engine.fs); err != nil {
+		tbl.tableId, tbl.tableName,
+		tbl.db.databaseId, tbl.db.databaseName, tbl.db.txn.engine.fs); err != nil {
 		return
 	}
 	for _, entry := range entries {
-		if err = consumeEntry(ctx, primaryIdx,
+		if err = consumeEntry(idx, tbl.primaryIdx, tbl, ctx,
 			engine, partition, state, entry); err != nil {
 			return
 		}
 	}
 
 	for i := 0; i < len(lt.Commands); i++ {
-		if err = consumeEntry(ctx, primaryIdx,
+		if err = consumeEntry(idx, tbl.primaryIdx, tbl, ctx,
 			engine, partition, state, &lt.Commands[i]); err != nil {
 			return
 		}
diff --git a/pkg/vm/engine/disttae/partition.go b/pkg/vm/engine/disttae/partition.go
index 9755e4d83..801c276d2 100644
--- a/pkg/vm/engine/disttae/partition.go
+++ b/pkg/vm/engine/disttae/partition.go
@@ -25,6 +25,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/api"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
@@ -35,6 +36,7 @@ import (
 
 func NewPartition(
 	columnsIndexDefs []ColumnsIndexDef,
+	isMeta bool,
 ) *Partition {
 	lock := make(chan struct{}, 1)
 	lock <- struct{}{}
@@ -42,6 +44,7 @@ func NewPartition(
 		lock:             lock,
 		data:             memtable.NewTable[RowID, DataValue, *DataRow](),
 		columnsIndexDefs: columnsIndexDefs,
+		isMeta:           isMeta,
 	}
 	ret.state.Store(NewPartitionState())
 	return ret
@@ -351,19 +354,17 @@ func (p *Partition) DeleteByBlockID(ctx context.Context, ts timestamp.Timestamp,
 
 func (p *Partition) NewReader(
 	ctx context.Context,
-	txn *Transaction,
 	readerNumber int,
 	index memtable.Tuple,
 	defs []engine.TableDef,
 	tableDef *plan.TableDef,
 	skipBlocks map[uint64]uint8,
 	blks []ModifyBlockMeta,
+	ts timestamp.Timestamp,
+	fs fileservice.FileService,
 	entries []Entry,
 ) ([]engine.Reader, error) {
 
-	ts := txn.meta.SnapshotTS
-	fs := txn.engine.fs
-
 	inserts := make([]*batch.Batch, 0, len(entries))
 	deletes := make(map[types.Rowid]uint8)
 	for _, entry := range entries {
@@ -400,8 +401,9 @@ func (p *Partition) NewReader(
 
 	var newIter *partitionStateRowsIter
 	var iter partitionIter
-	if len(index) == 0 {
+	if len(index) == 0 && !p.isMeta {
 		// skip primary key indexing read
+		// and meta partition read
 		//TODO migrate to new iter
 		newIter = p.state.Load().NewRowsIter(
 			types.TimestampToTS(ts),
@@ -430,7 +432,9 @@ func (p *Partition) NewReader(
 		colIdxMp:        colIdxMp,
 		extendId2s3File: make(map[string]int),
 		s3FileService:   fs,
-		procMPool:       txn.proc.GetMPool(),
+	}
+	if p.txn != nil {
+		partReader.procMPool = p.txn.proc.GetMPool()
 	}
 	readers[0] = partReader
 	if readerNumber == 1 {
@@ -489,11 +493,3 @@ func (p *Partition) NewReader(
 	}
 	return readers, nil
 }
-
-func (p Partitions) Snapshot() []*PartitionState {
-	ret := make([]*PartitionState, 0, len(p))
-	for _, partition := range p {
-		ret = append(ret, partition.state.Load())
-	}
-	return ret
-}
diff --git a/pkg/vm/engine/disttae/partition_state.go b/pkg/vm/engine/disttae/partition_state.go
index f6c5a62c0..6ec2a5236 100644
--- a/pkg/vm/engine/disttae/partition_state.go
+++ b/pkg/vm/engine/disttae/partition_state.go
@@ -17,12 +17,10 @@ package disttae
 import (
 	"bytes"
 	"context"
-	"fmt"
 	"net/http"
 	"runtime/trace"
 	"sync/atomic"
 
-	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
@@ -56,21 +54,18 @@ type RowEntry struct {
 }
 
 func (r RowEntry) Less(than RowEntry) bool {
-	// asc
 	if r.BlockID < than.BlockID {
 		return true
 	}
 	if than.BlockID < r.BlockID {
 		return false
 	}
-	// asc
 	if r.RowID.Less(than.RowID) {
 		return true
 	}
 	if than.RowID.Less(r.RowID) {
 		return false
 	}
-	// desc
 	if than.Time.Less(r.Time) {
 		return true
 	}
@@ -81,10 +76,15 @@ func (r RowEntry) Less(than RowEntry) bool {
 }
 
 type BlockEntry struct {
-	catalog.BlockInfo
+	BlockID uint64
 
-	CreateTime types.TS
-	DeleteTime types.TS
+	MetaLocation  string
+	DeltaLocation string
+	SegmentID     uint64
+	Sorted        bool
+	CreateTime    types.TS
+	CommitTime    types.TS
+	DeleteTime    types.TS
 }
 
 func (b BlockEntry) Less(than BlockEntry) bool {
@@ -97,11 +97,6 @@ func (b BlockEntry) Less(than BlockEntry) bool {
 	return false
 }
 
-func (b *BlockEntry) Visible(ts types.TS) bool {
-	return b.CreateTime.LessEq(ts) &&
-		(b.DeleteTime.IsEmpty() || ts.Less(b.DeleteTime))
-}
-
 type PrimaryIndexEntry struct {
 	Bytes      []byte
 	RowEntryID int64
@@ -134,37 +129,6 @@ func (p *PartitionState) Copy() *PartitionState {
 	}
 }
 
-func (p *PartitionState) RowExists(rowID types.Rowid, ts types.TS) bool {
-	iter := p.Rows.Iter()
-	defer iter.Release()
-
-	blockID := blockIDFromRowID(rowID)
-	for ok := iter.Seek(RowEntry{
-		BlockID: blockID,
-		RowID:   rowID,
-		Time:    ts,
-	}); ok; ok = iter.Next() {
-		entry := iter.Item()
-		if entry.BlockID != blockID {
-			break
-		}
-		if entry.RowID != rowID {
-			break
-		}
-		if entry.Time.Greater(ts) {
-			// not visible
-			continue
-		}
-		if entry.Deleted {
-			// deleted
-			return false
-		}
-		return true
-	}
-
-	return false
-}
-
 func (p *PartitionState) HandleLogtailEntry(ctx context.Context, entry *api.Entry, primaryKeyIndex int) {
 	switch entry.EntryType {
 	case api.Entry_Insert:
@@ -285,9 +249,7 @@ func (p *PartitionState) HandleMetadataInsert(ctx context.Context, input *api.Ba
 		trace.WithRegion(ctx, "handle a row", func() {
 
 			pivot := BlockEntry{
-				BlockInfo: catalog.BlockInfo{
-					BlockID: blockID,
-				},
+				BlockID: blockID,
 			}
 			entry, ok := p.Blocks.Get(pivot)
 			if !ok {
@@ -295,10 +257,10 @@ func (p *PartitionState) HandleMetadataInsert(ctx context.Context, input *api.Ba
 			}
 
 			if location := metaLocationVector[i]; location != "" {
-				entry.MetaLoc = location
+				entry.MetaLocation = location
 			}
 			if location := deltaLocationVector[i]; location != "" {
-				entry.DeltaLoc = location
+				entry.DeltaLocation = location
 			}
 			if id := segmentIDVector[i]; id > 0 {
 				entry.SegmentID = id
@@ -308,9 +270,8 @@ func (p *PartitionState) HandleMetadataInsert(ctx context.Context, input *api.Ba
 				entry.CreateTime = t
 			}
 			if t := commitTimeVector[i]; !t.IsEmpty() {
-				entry.CommitTs = t
+				entry.CommitTime = t
 			}
-			entry.EntryState = entryStateVector[i]
 
 			p.Blocks.Set(entry)
 
@@ -343,17 +304,15 @@ func (p *PartitionState) HandleMetadataDelete(ctx context.Context, input *api.Ba
 	deleteTimeVector := vector.MustTCols[types.TS](mustVectorFromProto(input.Vecs[1]))
 
 	for i, rowID := range rowIDVector {
-		blockID := types.DecodeUint64(rowID[:8])
+		blockID := blockIDFromRowID(rowID)
 		trace.WithRegion(ctx, "handle a row", func() {
 
 			pivot := BlockEntry{
-				BlockInfo: catalog.BlockInfo{
-					BlockID: blockID,
-				},
+				BlockID: blockID,
 			}
 			entry, ok := p.Blocks.Get(pivot)
 			if !ok {
-				panic(fmt.Sprintf("invalid block id. %x", rowID))
+				entry = pivot
 			}
 
 			entry.DeleteTime = deleteTimeVector[i]
diff --git a/pkg/vm/engine/disttae/tools.go b/pkg/vm/engine/disttae/tools.go
index f8c37ed04..4985b7fe4 100644
--- a/pkg/vm/engine/disttae/tools.go
+++ b/pkg/vm/engine/disttae/tools.go
@@ -1031,10 +1031,11 @@ func isMetaTable(name string) bool {
 
 func genBlockMetas(
 	ctx context.Context,
-	blockInfos []catalog.BlockInfo,
+	rows [][]any,
 	columnLength int,
 	fs fileservice.FileService,
 	m *mpool.MPool, prefetch bool) ([]BlockMeta, error) {
+	blockInfos := catalog.GenBlockInfo(rows)
 	{
 		mp := make(map[uint64]catalog.BlockInfo) // block list
 		for i := range blockInfos {
diff --git a/pkg/vm/engine/disttae/txn.go b/pkg/vm/engine/disttae/txn.go
index ae6b33cd5..22894c0e7 100644
--- a/pkg/vm/engine/disttae/txn.go
+++ b/pkg/vm/engine/disttae/txn.go
@@ -31,6 +31,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan"
 	"github.com/matrixorigin/matrixone/pkg/txn/storage/memorystorage/memorytable"
 	"github.com/matrixorigin/matrixone/pkg/txn/storage/memorystorage/memtable"
@@ -40,50 +41,110 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
-func (txn *Transaction) getTableMeta(
-	ctx context.Context,
-	databaseId uint64,
-	tableId uint64,
-	needUpdated bool,
-	columnLength int,
-	prefetch bool,
-) (*tableMeta, error) {
-	blocks := make([][]BlockMeta, len(txn.dnStores))
-	name := genMetaTableName(tableId)
+/*
+func (txn *Transaction) getTableList(ctx context.Context, databaseId uint64) ([]string, error) {
+	rows, err := txn.getRows(ctx, "", catalog.MO_CATALOG_ID, catalog.MO_TABLES_ID, txn.dnStores[:1],
+		catalog.MoTablesTableDefs, []string{
+			catalog.MoTablesSchema[catalog.MO_TABLES_REL_NAME_IDX],
+			catalog.MoTablesSchema[catalog.MO_TABLES_RELDATABASE_ID_IDX],
+			catalog.MoTablesSchema[catalog.MO_TABLES_ACCOUNT_ID_IDX],
+		},
+		genTableListExpr(ctx, getAccountId(ctx), databaseId))
+	if err != nil {
+		return nil, err
+	}
+	tableList := make([]string, len(rows))
+	for i := range rows {
+		tableList[i] = string(rows[i][0].([]byte))
+	}
+	return tableList, nil
+}
 
-	ts := types.TimestampToTS(txn.meta.SnapshotTS)
-	if needUpdated {
-		key := [2]uint64{databaseId, tableId}
-		partitions := txn.engine.partitions[key]
+func (txn *Transaction) getTableId(ctx context.Context, databaseId uint64,
+	name string) (uint64, error) {
+	accountId := getAccountId(ctx)
+	row, err := txn.getRow(ctx, catalog.MO_CATALOG_ID, catalog.MO_TABLES_ID,
+		txn.dnStores[:1],
+		catalog.MoTablesTableDefs, []string{
+			catalog.MoTablesSchema[catalog.MO_TABLES_REL_ID_IDX],
+			catalog.MoTablesSchema[catalog.MO_TABLES_REL_NAME_IDX],
+			catalog.MoTablesSchema[catalog.MO_TABLES_RELDATABASE_ID_IDX],
+			catalog.MoTablesSchema[catalog.MO_TABLES_ACCOUNT_ID_IDX],
+		},
+		genTableIdExpr(ctx, accountId, databaseId, name))
+	if err != nil {
+		return 0, err
+	}
+	return row[0].(uint64), nil
+}
+
+func (txn *Transaction) getDatabaseList(ctx context.Context) ([]string, error) {
+	rows, err := txn.getRows(ctx, "", catalog.MO_CATALOG_ID, catalog.MO_DATABASE_ID,
+		txn.dnStores[:1],
+		catalog.MoDatabaseTableDefs, []string{
+			catalog.MoDatabaseSchema[catalog.MO_DATABASE_DAT_NAME_IDX],
+			catalog.MoDatabaseSchema[catalog.MO_DATABASE_ACCOUNT_ID_IDX],
+		},
+		genDatabaseListExpr(ctx, getAccountId(ctx)))
+	if err != nil {
+		return nil, err
+	}
+	databaseList := make([]string, len(rows))
+	for i := range rows {
+		databaseList[i] = string(rows[i][0].([]byte))
+	}
+	return databaseList, nil
+}
+
+func (txn *Transaction) getDatabaseId(ctx context.Context, name string) (uint64, error) {
+	accountId := getAccountId(ctx)
+	key := genDatabaseIndexKey(name, accountId)
+	rows, err := txn.getRowsByIndex(catalog.MO_CATALOG_ID, catalog.MO_DATABASE_ID, "",
+		txn.dnStores[:1], []string{
+			catalog.MoDatabaseSchema[catalog.MO_DATABASE_DAT_ID_IDX],
+			catalog.MoDatabaseSchema[catalog.MO_DATABASE_DAT_NAME_IDX],
+			catalog.MoDatabaseSchema[catalog.MO_DATABASE_ACCOUNT_ID_IDX],
+		}, key, genDatabaseIdExpr(ctx, accountId, name))
+	if err != nil {
+		return 0, err
+	}
+	if len(rows) != 1 {
+		return 0, moerr.NewDuplicate(ctx)
+	}
+	//	row, err := txn.getRow(ctx, catalog.MO_CATALOG_ID, catalog.MO_DATABASE_ID, txn.dnStores[:1],
+	//		catalog.MoDatabaseTableDefs, []string{
+	//			catalog.MoDatabaseSchema[catalog.MO_DATABASE_DAT_ID_IDX],
+	//			catalog.MoDatabaseSchema[catalog.MO_DATABASE_DAT_NAME_IDX],
+	//			catalog.MoDatabaseSchema[catalog.MO_DATABASE_ACCOUNT_ID_IDX],
+	//		},
+	//		genDatabaseIdExpr(accountId, name))
+	//	if err != nil {
+	//		return 0, err
+	//	}
+	return rows[0][0].(uint64), nil
+}
+*/
 
-		for i := range txn.dnStores {
-			if i >= len(partitions) {
+func (txn *Transaction) getTableMeta(ctx context.Context, databaseId uint64,
+	name string, needUpdated bool, columnLength int, prefetch bool) (*tableMeta, error) {
+	blocks := make([][]BlockMeta, len(txn.dnStores))
+	if needUpdated {
+		for i, dnStore := range txn.dnStores {
+			rows, err := txn.getRows(ctx, name, databaseId, 0,
+				[]DNStore{dnStore}, catalog.MoTableMetaDefs, catalog.MoTableMetaSchema, nil)
+			if moerr.IsMoErrCode(err, moerr.OkExpectedEOB) {
 				continue
 			}
-
-			var blockInfos []catalog.BlockInfo
-			partition := partitions[i]
-			state := partition.state.Load()
-			iter := state.Blocks.Iter()
-			for ok := iter.First(); ok; ok = iter.Next() {
-				entry := iter.Item()
-				if !entry.Visible(ts) {
-					continue
-				}
-				blockInfos = append(blockInfos, entry.BlockInfo)
+			if err != nil {
+				return nil, err
 			}
-			iter.Release()
-
-			var err error
-			blocks[i], err = genBlockMetas(ctx, blockInfos, columnLength, txn.proc.FileService,
+			blocks[i], err = genBlockMetas(ctx, rows, columnLength, txn.proc.FileService,
 				txn.proc.GetMPool(), prefetch)
 			if err != nil {
 				return nil, moerr.NewInternalError(ctx, "disttae: getTableMeta err: %v, table: %v", err.Error(), name)
 			}
-
 		}
 	}
-
 	return &tableMeta{
 		tableName: name,
 		blocks:    blocks,
@@ -253,6 +314,139 @@ func (txn *Transaction) WriteFile(typ int, databaseId, tableId uint64,
 	return nil
 }
 
+// getRow used to get a row of table based on a condition
+/*
+func (txn *Transaction) getRow(ctx context.Context, databaseId uint64, tableId uint64,
+	dnList []DNStore, defs []engine.TableDef, columns []string, expr *plan.Expr) ([]any, error) {
+	bats, err := txn.readTable(ctx, "", databaseId, tableId, defs, dnList, columns, expr)
+	if err != nil {
+		return nil, err
+	}
+	if len(bats) == 0 {
+		return nil, moerr.GetOkExpectedEOB()
+	}
+	rows := make([][]any, 0, len(bats))
+	for _, bat := range bats {
+		if bat.Length() > 0 {
+			rows = append(rows, catalog.GenRows(bat)...)
+		}
+		bat.Clean(txn.proc.Mp())
+	}
+	if len(rows) == 0 {
+		return nil, moerr.GetOkExpectedEOB()
+	}
+	if len(rows) != 1 {
+		return nil, moerr.NewInvalidInput(ctx, "table is not unique")
+	}
+	return rows[0], nil
+}
+*/
+
+// getRows used to get rows of table
+func (txn *Transaction) getRows(ctx context.Context, name string, databaseId uint64, tableId uint64,
+	dnList []DNStore, defs []engine.TableDef, columns []string, expr *plan.Expr) ([][]any, error) {
+	bats, err := txn.readTable(ctx, name, databaseId, tableId, defs, dnList, columns, expr)
+	if err != nil {
+		return nil, err
+	}
+	if len(bats) == 0 {
+		return nil, moerr.GetOkExpectedEOB()
+	}
+	rows := make([][]any, 0, len(bats))
+	for _, bat := range bats {
+		if bat.Length() > 0 {
+			rows = append(rows, catalog.GenRows(bat)...)
+		}
+		bat.Clean(txn.proc.Mp())
+	}
+	return rows, nil
+}
+
+// readTable used to get tuples of table based on a condition
+// only used to read data from catalog, for which the execution is currently single-core
+func (txn *Transaction) readTable(ctx context.Context, name string, databaseId uint64, tableId uint64,
+	defs []engine.TableDef, dnList []DNStore, columns []string, expr *plan.Expr) ([]*batch.Batch, error) {
+	var parts Partitions
+	/*
+		var writes [][]Entry
+		// consider halloween problem
+		if int64(txn.statementId)-1 > 0 {
+			writes = txn.writes[:txn.statementId-1]
+		}
+	*/
+	writes := make([]Entry, 0, len(txn.writes))
+	if len(name) == 0 { // meta table not need this
+		for i := range txn.writes {
+			for _, entry := range txn.writes[i] {
+				if entry.databaseId == databaseId &&
+					entry.tableId == tableId {
+					writes = append(writes, entry)
+				}
+			}
+		}
+	}
+	bats := make([]*batch.Batch, 0, 1)
+	accessed := make(map[string]uint8)
+	for _, dn := range dnList {
+		accessed[dn.ServiceID] = 0
+	}
+	if len(name) == 0 {
+		parts = txn.engine.getPartitions(databaseId, tableId)
+	} else {
+		parts = txn.engine.getMetaPartitions(name)
+	}
+	for i, dn := range txn.dnStores {
+		if _, ok := accessed[dn.ServiceID]; !ok {
+			continue
+		}
+		rds, err := parts[i].NewReader(ctx, 1, nil, defs, nil, nil, nil,
+			txn.meta.SnapshotTS, nil, writes)
+		if err != nil {
+			return nil, err
+		}
+		for _, rd := range rds {
+			for {
+				bat, err := rd.Read(ctx, columns, expr, txn.proc.Mp())
+				if err != nil {
+					return nil, err
+				}
+				if bat != nil {
+					bats = append(bats, bat)
+				} else {
+					break
+				}
+			}
+		}
+	}
+	if expr == nil {
+		return bats, nil
+	}
+	for i, bat := range bats {
+		vec, err := colexec.EvalExpr(bat, txn.proc, expr)
+		if err != nil {
+			return nil, err
+		}
+		bs := vector.GetColumn[bool](vec)
+		if vec.IsScalar() {
+			if !bs[0] {
+				bat.Shrink(nil)
+			}
+		} else {
+			sels := txn.proc.Mp().GetSels()
+			for i, b := range bs {
+				if b {
+					sels = append(sels, int64(i))
+				}
+			}
+			bat.Shrink(sels)
+			txn.proc.Mp().PutSels(sels)
+		}
+		vec.Free(txn.proc.Mp())
+		bats[i] = bat
+	}
+	return bats, nil
+}
+
 func (txn *Transaction) deleteBatch(bat *batch.Batch,
 	databaseId, tableId uint64) *batch.Batch {
 
diff --git a/pkg/vm/engine/disttae/txn_database.go b/pkg/vm/engine/disttae/txn_database.go
index cc5d86985..d6f08ceb8 100644
--- a/pkg/vm/engine/disttae/txn_database.go
+++ b/pkg/vm/engine/disttae/txn_database.go
@@ -124,7 +124,7 @@ func (db *txnDatabase) Relation(ctx context.Context, name string) (engine.Relati
 		constraint:   item.Constraint,
 	}
 	columnLength := len(item.TableDef.Cols) - 1 // we use this data to fetch zonemap, but row_id has no zonemap
-	meta, err := db.txn.getTableMeta(ctx, db.databaseId, item.Id,
+	meta, err := db.txn.getTableMeta(ctx, db.databaseId, genMetaTableName(item.Id),
 		true, columnLength, true)
 	if err != nil {
 		return nil, err
diff --git a/pkg/vm/engine/disttae/txn_table.go b/pkg/vm/engine/disttae/txn_table.go
index f9c841ea0..016b34e68 100644
--- a/pkg/vm/engine/disttae/txn_table.go
+++ b/pkg/vm/engine/disttae/txn_table.go
@@ -572,17 +572,9 @@ func (tbl *txnTable) newMergeReader(ctx context.Context, num int,
 		if tbl.meta != nil {
 			blks = tbl.meta.modifedBlocks[i]
 		}
-		rds0, err := tbl.parts[i].NewReader(
-			ctx,
-			tbl.db.txn,
-			num,
-			index,
-			tbl.defs,
-			tbl.tableDef,
-			tbl.skipBlocks,
-			blks,
-			writes,
-		)
+		tbl.parts[i].txn = tbl.db.txn
+		rds0, err := tbl.parts[i].NewReader(ctx, num, index, tbl.defs, tbl.tableDef,
+			tbl.skipBlocks, blks, tbl.db.txn.meta.SnapshotTS, tbl.db.txn.engine.fs, writes)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/vm/engine/disttae/txn_table_meta.go b/pkg/vm/engine/disttae/txn_table_meta.go
index fb06b4003..7a6988bed 100644
--- a/pkg/vm/engine/disttae/txn_table_meta.go
+++ b/pkg/vm/engine/disttae/txn_table_meta.go
@@ -63,7 +63,7 @@ func (tbl *txnTable) updateMeta(ctx context.Context, expr *plan.Expr) error {
 		}
 
 		columnLength := len(tbl.tableDef.Cols) - 1 //we use this data to fetch zonemap, but row_id has no zonemap
-		meta, err := tbl.db.txn.getTableMeta(ctx, tbl.db.databaseId, tbl.tableId, true, columnLength, false)
+		meta, err := tbl.db.txn.getTableMeta(ctx, tbl.db.databaseId, genMetaTableName(tbl.tableId), true, columnLength, false)
 		if err != nil {
 			return err
 		}
diff --git a/pkg/vm/engine/disttae/types.go b/pkg/vm/engine/disttae/types.go
index 14b09b987..4f6e90cdb 100644
--- a/pkg/vm/engine/disttae/types.go
+++ b/pkg/vm/engine/disttae/types.go
@@ -77,6 +77,7 @@ type Engine struct {
 	txnHeap *transactionHeap
 
 	dnMap      map[string]int
+	metaTables map[string]Partitions
 	partitions map[[2]uint64]Partitions
 
 	// XXX related to cn push model
@@ -97,6 +98,9 @@ type Partition struct {
 	columnsIndexDefs []ColumnsIndexDef
 	// last updated timestamp
 	ts timestamp.Timestamp
+	// used for block read in PartitionReader
+	txn    *Transaction
+	isMeta bool
 }
 
 // Transaction represents a transaction


Test patch:
diff --git a/pkg/vm/engine/disttae/bench_test.go b/pkg/vm/engine/disttae/bench_test.go
index 0da5e6ab1..057210a09 100644
--- a/pkg/vm/engine/disttae/bench_test.go
+++ b/pkg/vm/engine/disttae/bench_test.go
@@ -38,7 +38,7 @@ func BenchmarkPartitionInsert(b *testing.B) {
 		b.Run(fmt.Sprintf("num-rows-%v", numRows), func(b *testing.B) {
 
 			mp := mpool.MustNewZero()
-			partition := NewPartition(nil)
+			partition := NewPartition(nil, false)
 			v0 := make([]types.Rowid, numRows)
 			v1 := make([]types.TS, numRows)
 			v2 := make([]int64, numRows)
diff --git a/pkg/vm/engine/disttae/partition_state_iterators_test.go b/pkg/vm/engine/disttae/partition_state_iterators_test.go
index bdab20aba..b828b2b65 100644
--- a/pkg/vm/engine/disttae/partition_state_iterators_test.go
+++ b/pkg/vm/engine/disttae/partition_state_iterators_test.go
@@ -82,15 +82,13 @@ func TestPartitionStateRowsIter(t *testing.T) {
 	}
 
 	for i := 0; i < num; i++ {
-		ts := types.BuildTS(int64(i), 0)
-		iter := state.NewRowsIter(ts, nil, false)
+		iter := state.NewRowsIter(types.BuildTS(int64(i), 0), nil, false)
 		n := 0
 		rowIDs := make(map[types.Rowid]bool)
 		for iter.Next() {
 			n++
 			entry := iter.Entry()
 			rowIDs[entry.RowID] = true
-			require.True(t, state.RowExists(entry.RowID, ts))
 		}
 		require.Equal(t, i+1, n)
 		require.Equal(t, i+1, len(rowIDs))